MetaGPT

1
#!/usr/bin/env python
2
# -*- coding: utf-8 -*-
3
"""
4
@Time    : 2023/11/22
5
@Author  : mashenquan
6
@File    : dependency_file.py
7
@Desc: Implementation of the dependency file described in Section 2.2.3.2 of RFC 135.
8
"""
9
from __future__ import annotations
10

11
import json
12
import re
13
from pathlib import Path
14
from typing import Set
15

16
from metagpt.utils.common import aread, awrite
17
from metagpt.utils.exceptions import handle_exception
18

19

20
class DependencyFile:
21
    """A class representing a DependencyFile for managing dependencies.
22

23
    :param workdir: The working directory path for the DependencyFile.
24
    """
25

26
    def __init__(self, workdir: Path | str):
27
        """Initialize a DependencyFile instance.
28

29
        :param workdir: The working directory path for the DependencyFile.
30
        """
31
        self._dependencies = {}
32
        self._filename = Path(workdir) / ".dependencies.json"
33

34
    async def load(self):
35
        """Load dependencies from the file asynchronously."""
36
        if not self._filename.exists():
37
            return
38
        json_data = await aread(self._filename)
39
        json_data = re.sub(r"\\+", "/", json_data)  # Compatible with windows path
40
        self._dependencies = json.loads(json_data)
41

42
    @handle_exception
43
    async def save(self):
44
        """Save dependencies to the file asynchronously."""
45
        data = json.dumps(self._dependencies)
46
        await awrite(filename=self._filename, data=data)
47

48
    async def update(self, filename: Path | str, dependencies: Set[Path | str], persist=True):
49
        """Update dependencies for a file asynchronously.
50

51
        :param filename: The filename or path.
52
        :param dependencies: The set of dependencies.
53
        :param persist: Whether to persist the changes immediately.
54
        """
55
        if persist:
56
            await self.load()
57

58
        root = self._filename.parent
59
        try:
60
            key = Path(filename).relative_to(root).as_posix()
61
        except ValueError:
62
            key = filename
63
        key = str(key)
64
        if dependencies:
65
            relative_paths = []
66
            for i in dependencies:
67
                try:
68
                    s = str(Path(i).relative_to(root).as_posix())
69
                except ValueError:
70
                    s = str(i)
71
                relative_paths.append(s)
72

73
            self._dependencies[key] = relative_paths
74
        elif key in self._dependencies:
75
            del self._dependencies[key]
76

77
        if persist:
78
            await self.save()
79

80
    async def get(self, filename: Path | str, persist=True):
81
        """Get dependencies for a file asynchronously.
82

83
        :param filename: The filename or path.
84
        :param persist: Whether to load dependencies from the file immediately.
85
        :return: A set of dependencies.
86
        """
87
        if persist:
88
            await self.load()
89

90
        root = self._filename.parent
91
        try:
92
            key = Path(filename).relative_to(root).as_posix()
93
        except ValueError:
94
            key = filename
95
        return set(self._dependencies.get(str(key), {}))
96

97
    def delete_file(self):
98
        """Delete the dependency file."""
99
        self._filename.unlink(missing_ok=True)
100

101
    @property
102
    def exists(self):
103
        """Check if the dependency file exists."""
104
        return self._filename.exists()
105