|
1 | 1 | from pathlib import Path |
| 2 | +from typing import ClassVar |
2 | 3 |
|
3 | 4 | from tomlkit.api import dumps, parse |
4 | 5 | from tomlkit.exceptions import TOMLKitError |
|
11 | 12 | ) |
12 | 13 |
|
13 | 14 |
|
14 | | -def read_pyproject_toml() -> TOMLDocument: |
15 | | - return pyproject_toml_io_manager._opener.read() |
| 15 | +class UnexpectedPyprojectTOMLOpenError(Exception): |
| 16 | + """Raised when the 'pyproject.toml' opener is accessed unexpectedly.""" |
| 17 | + |
16 | 18 |
|
| 19 | +class UnexpectedPyprojectTOMLCloseError(Exception): |
| 20 | + """Raised when the 'pyproject.toml' opener is closed unexpectedly.""" |
17 | 21 |
|
18 | | -def write_pyproject_toml(toml_document: TOMLDocument) -> None: |
19 | | - return pyproject_toml_io_manager._opener.write(toml_document) |
20 | 22 |
|
| 23 | +class UnexpectedPyprojectTOMLIOError(Exception): |
| 24 | + """Raised when an unexpected attempt is made to read or write 'pyproject.toml'.""" |
21 | 25 |
|
22 | | -class UnexpectedPyprojectTOMLReadError(Exception): |
23 | | - """Raised when the pyproject.toml is read unexpectedly.""" |
24 | 26 |
|
| 27 | +class PyprojectTOMLManager: |
| 28 | + _content_by_path: ClassVar[dict[Path, TOMLDocument | None]] = {} |
25 | 29 |
|
26 | | -class PyprojectTOMLOpener: |
27 | 30 | def __init__(self) -> None: |
28 | | - self.path = Path.cwd() / "pyproject.toml" |
29 | | - self.content = TOMLDocument() |
30 | | - self.open = False |
31 | | - self._set = False |
32 | | - |
33 | | - def read(self) -> TOMLDocument: |
34 | | - if not self._set: |
35 | | - msg = """The pyproject.toml opener has not been set yet.""" |
| 31 | + self._path = (Path.cwd() / "pyproject.toml").resolve() |
| 32 | + |
| 33 | + def __enter__(self) -> Self: |
| 34 | + if self.is_locked(): |
| 35 | + msg = ( |
| 36 | + f"The 'pyproject.toml' file is already in use by another instance of " |
| 37 | + f"'{self.__class__.__name__}'." |
| 38 | + ) |
36 | 39 | raise UnexpectedPyprojectTOMLOpenError(msg) |
37 | 40 |
|
38 | | - if not self.open: |
| 41 | + self.lock() |
| 42 | + return self |
| 43 | + |
| 44 | + def __exit__(self, exc_type: None, exc_value: None, traceback: None) -> None: |
| 45 | + if not self.is_locked(): |
| 46 | + # This could happen if we decide to delete the file. |
| 47 | + return |
| 48 | + |
| 49 | + self.write_file() |
| 50 | + self.unlock() |
| 51 | + |
| 52 | + def get(self) -> TOMLDocument: |
| 53 | + self._validate_lock() |
| 54 | + |
| 55 | + if self._content is None: |
39 | 56 | self.read_file() |
40 | | - self.open = True |
| 57 | + assert self._content is not None |
41 | 58 |
|
42 | | - return self.content |
| 59 | + return self._content |
43 | 60 |
|
44 | | - def write(self, toml_document: TOMLDocument) -> None: |
45 | | - if not self._set: |
46 | | - msg = """The pyproject.toml opener has not been set yet.""" |
47 | | - raise UnexpectedPyprojectTOMLOpenError(msg) |
| 61 | + def commit(self, toml_document: TOMLDocument) -> None: |
| 62 | + self._validate_lock() |
48 | 63 |
|
49 | | - self.content = toml_document |
| 64 | + self._content = toml_document |
50 | 65 |
|
51 | 66 | def write_file(self) -> None: |
52 | | - self.path.write_text(dumps(self.content)) |
| 67 | + self._validate_lock() |
| 68 | + |
| 69 | + if self._content is None: |
| 70 | + # No changes made, nothing to write. |
| 71 | + return |
| 72 | + |
| 73 | + self._path.write_text(dumps(self._content)) |
53 | 74 |
|
54 | 75 | def read_file(self) -> None: |
| 76 | + self._validate_lock() |
| 77 | + |
| 78 | + if self._content is not None: |
| 79 | + msg = ( |
| 80 | + "The 'pyproject.toml' file has already been read, use 'get()' to " |
| 81 | + "access the content." |
| 82 | + ) |
| 83 | + raise UnexpectedPyprojectTOMLIOError(msg) |
55 | 84 | try: |
56 | | - self.content = parse(self.path.read_text()) |
| 85 | + self._content = parse(self._path.read_text()) |
57 | 86 | except FileNotFoundError: |
58 | 87 | msg = "'pyproject.toml' not found in the current directory." |
59 | 88 | raise PyprojectTOMLNotFoundError(msg) |
60 | 89 | except TOMLKitError as err: |
61 | 90 | msg = f"Failed to decode 'pyproject.toml': {err}" |
62 | 91 | raise PyprojectTOMLDecodeError(msg) from None |
63 | 92 |
|
64 | | - def __enter__(self) -> Self: |
65 | | - self._set = True |
66 | | - return self |
67 | | - |
68 | | - def __exit__(self, exc_type: None, exc_value: None, traceback: None) -> None: |
69 | | - self.write_file() |
70 | | - self._set = False |
71 | | - |
72 | | - |
73 | | -class UnexpectedPyprojectTOMLOpenError(Exception): |
74 | | - """Raised when the pyproject.toml opener is accessed unexpectedly.""" |
75 | | - |
76 | | - |
77 | | -class PyprojectTOMLIOManager: |
78 | | - def __init__(self) -> None: |
79 | | - self._opener = PyprojectTOMLOpener() |
80 | | - self._set = False |
81 | | - |
82 | 93 | @property |
83 | | - def opener(self) -> PyprojectTOMLOpener: |
84 | | - if not self._opener._set: |
85 | | - self._set = False |
| 94 | + def _content(self) -> TOMLDocument | None: |
| 95 | + return self._content_by_path[self._path] |
86 | 96 |
|
87 | | - if not self._set: |
88 | | - msg = """The pyproject.toml opener has not been set to open yet.""" |
89 | | - raise UnexpectedPyprojectTOMLOpenError(msg) |
| 97 | + @_content.setter |
| 98 | + def _content(self, value: TOMLDocument | None) -> None: |
| 99 | + self._content_by_path[self._path] = value |
90 | 100 |
|
91 | | - return self._opener |
| 101 | + def _validate_lock(self) -> None: |
| 102 | + if not self.is_locked(): |
| 103 | + msg = ( |
| 104 | + f"The 'pyproject.toml' file has not been opened yet. Please enter the " |
| 105 | + f"context manager, e.g. 'with {self.__class__.__name__}():'" |
| 106 | + ) |
| 107 | + raise UnexpectedPyprojectTOMLIOError(msg) |
92 | 108 |
|
93 | | - def open(self) -> PyprojectTOMLOpener: |
94 | | - self._opener = PyprojectTOMLOpener() |
95 | | - return self._opener |
| 109 | + def is_locked(self) -> bool: |
| 110 | + return self._path in self._content_by_path |
96 | 111 |
|
| 112 | + def lock(self) -> None: |
| 113 | + self._content = None |
97 | 114 |
|
98 | | -pyproject_toml_io_manager = PyprojectTOMLIOManager() |
| 115 | + def unlock(self) -> None: |
| 116 | + self._content_by_path.pop(self._path, None) |
0 commit comments