Use tomli for reading the lock file

pyproject.toml

@@ -48,7 +48,7 @@ generate-setup-file = false
 python = "^3.7"
 
 poetry-core = "^1.3.2"
-poetry-plugin-export = "^1.1.2"
+poetry-plugin-export = "^1.2.0"
 "backports.cached-property" = { version = "^1.0.2", python = "<3.8" }
 cachecontrol = { version = "^0.12.9", extras = ["filecache"] }
 cleo = "^1.0.0a5"
@@ -67,6 +67,7 @@ platformdirs = "^2.5.2"
 requests = "^2.18"
 requests-toolbelt = ">=0.9.1,<0.11.0"
 shellingham = "^1.5"
+tomli = { version = "^2.0.1", python = "<3.11" }
 # exclude 0.11.2 and 0.11.3 due to https://github.com/sdispater/tomlkit/issues/225
 tomlkit = ">=0.11.1,<1.0.0,!=0.11.2,!=0.11.3"
 trove-classifiers = "^2022.5.19"

src/poetry/console/commands/add.py

@@ -241,7 +241,7 @@ def handle(self) -> int:
 
         # Refresh the locker
         self.poetry.set_locker(
-            self.poetry.locker.__class__(self.poetry.locker.lock.path, poetry_content)
+            self.poetry.locker.__class__(self.poetry.locker.lock, poetry_content)
         )
         self.installer.set_locker(self.poetry.locker)
 

src/poetry/console/commands/remove.py

@@ -103,7 +103,7 @@ def handle(self) -> int:
 
         # Refresh the locker
         self.poetry.set_locker(
-            self.poetry.locker.__class__(self.poetry.locker.lock.path, poetry_content)
+            self.poetry.locker.__class__(self.poetry.locker.lock, poetry_content)
         )
         self.installer.set_locker(self.poetry.locker)
 

src/poetry/packages/locker.py

@@ -24,15 +24,15 @@
 from tomlkit import document
 from tomlkit import inline_table
 from tomlkit import table
-from tomlkit.exceptions import TOMLKitError
+
+from poetry.utils._compat import tomllib
 
 
 if TYPE_CHECKING:
     from poetry.core.packages.directory_dependency import DirectoryDependency
     from poetry.core.packages.file_dependency import FileDependency
     from poetry.core.packages.url_dependency import URLDependency
     from poetry.core.packages.vcs_dependency import VCSDependency
-    from tomlkit.items import Table
     from tomlkit.toml_document import TOMLDocument
 
     from poetry.repositories.lockfile_repository import LockfileRepository
@@ -53,17 +53,17 @@ class Locker:
     _relevant_keys = [*_legacy_keys, "group"]
 
     def __init__(self, lock: str | Path, local_config: dict[str, Any]) -> None:
-        self._lock = TOMLFile(lock)
+        self._lock = lock if isinstance(lock, Path) else Path(lock)
         self._local_config = local_config
-        self._lock_data: TOMLDocument | None = None
+        self._lock_data: dict[str, Any] | None = None
         self._content_hash = self._get_content_hash()
 
     @property
-    def lock(self) -> TOMLFile:
+    def lock(self) -> Path:
         return self._lock
 
     @property
-    def lock_data(self) -> TOMLDocument:
+    def lock_data(self) -> dict[str, Any]:
         if self._lock_data is None:
             self._lock_data = self._get_lock_data()
 
@@ -79,7 +79,8 @@ def is_fresh(self) -> bool:
         """
         Checks whether the lock file is still up to date with the current hash.
         """
-        lock = self._lock.read()
+        with self.lock.open("rb") as f:
+            lock = tomllib.load(f)
         metadata = lock.get("metadata", {})
 
         if "content-hash" in metadata:
@@ -111,7 +112,7 @@ def locked_repository(self) -> LockfileRepository:
             source_type = source.get("type")
             url = source.get("url")
             if source_type in ["directory", "file"]:
-                url = self._lock.path.parent.joinpath(url).resolve().as_posix()
+                url = self.lock.parent.joinpath(url).resolve().as_posix()
 
             name = info["name"]
             package = Package(
@@ -196,7 +197,7 @@ def locked_repository(self) -> LockfileRepository:
                         package.marker = parse_marker(split_dep[1].strip())
 
             for dep_name, constraint in info.get("dependencies", {}).items():
-                root_dir = self._lock.path.parent
+                root_dir = self.lock.parent
                 if package.source_type == "directory":
                     # root dir should be the source of the package relative to the lock
                     # path
@@ -267,11 +268,8 @@ def set_lock_data(self, root: Package, packages: list[Package]) -> bool:
         return do_write
 
     def _write_lock_data(self, data: TOMLDocument) -> None:
-        self.lock.write(data)
-
-        # Checking lock file data consistency
-        if data != self.lock.read():
-            raise RuntimeError("Inconsistent lock file data.")
+        lockfile = TOMLFile(self.lock)
+        lockfile.write(data)
 
         self._lock_data = None
 
@@ -292,16 +290,17 @@ def _get_content_hash(self) -> str:
 
         return sha256(json.dumps(relevant_content, sort_keys=True).encode()).hexdigest()
 
-    def _get_lock_data(self) -> TOMLDocument:
-        if not self._lock.exists():
+    def _get_lock_data(self) -> dict[str, Any]:
+        if not self.lock.exists():
             raise RuntimeError("No lockfile found. Unable to read locked packages")
 
-        try:
-            lock_data: TOMLDocument = self._lock.read()
-        except TOMLKitError as e:
-            raise RuntimeError(f"Unable to read the lock file ({e}).")
+        with self.lock.open("rb") as f:
+            try:
+                lock_data = tomllib.load(f)
+            except tomllib.TOMLDecodeError as e:
+                raise RuntimeError(f"Unable to read the lock file ({e}).")
 
-        metadata = cast("Table", lock_data["metadata"])
+        metadata = lock_data["metadata"]
         lock_version = Version.parse(metadata.get("lock-version", "1.0"))
         current_version = Version.parse(self._VERSION)
         accepted_versions = parse_constraint(self._READ_VERSION_RANGE)
@@ -441,7 +440,7 @@ def _dump_package(self, package: Package) -> dict[str, Any]:
                 url = Path(
                     os.path.relpath(
                         Path(url).resolve(),
-                        Path(self._lock.path.parent).resolve(),
+                        Path(self.lock.parent).resolve(),
                     )
                 ).as_posix()
 

src/poetry/utils/_compat.py

@@ -7,6 +7,14 @@
 
 # TODO: use try/except ImportError when
 # https://github.com/python/mypy/issues/1393 is fixed
+
+if sys.version_info < (3, 11):
+    # compatibility for python <3.11
+    import tomli as tomllib
+else:
+    import tomllib  # nopycln: import
+
+
 if sys.version_info < (3, 10):
     # compatibility for python <3.10
     import importlib_metadata as metadata
@@ -67,4 +75,5 @@ def list_to_shell_command(cmd: list[str]) -> str:
     "list_to_shell_command",
     "metadata",
     "to_str",
+    "tomllib",
 ]

tests/conftest.py

@@ -412,7 +412,7 @@ def _factory(
         poetry = Factory().create_poetry(project_dir)
 
         locker = TestLocker(
-            poetry.locker.lock.path, locker_config or poetry.locker._local_config
+            poetry.locker.lock, locker_config or poetry.locker._local_config
         )
         locker.write()
 

tests/helpers.py

@@ -181,7 +181,7 @@ def reset_poetry(self) -> None:
         self._poetry.set_pool(poetry.pool)
         self._poetry.set_config(poetry.config)
         self._poetry.set_locker(
-            TestLocker(poetry.locker.lock.path, self._poetry.local_config)
+            TestLocker(poetry.locker.lock, self._poetry.local_config)
         )
 
 

tests/packages/test_locker.py

@@ -231,7 +231,9 @@ def test_locker_properly_loads_extras(locker: Locker):
 content-hash = "c3d07fca33fba542ef2b2a4d75bf5b48d892d21a830e2ad9c952ba5123a52f77"
 """  # noqa: E800
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     packages = locker.locked_repository().packages
 
@@ -294,7 +296,9 @@ def test_locker_properly_loads_nested_extras(locker: Locker):
 content-hash = "123456789"
 """  # noqa: E800
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     repository = locker.locked_repository()
     assert len(repository.packages) == 3
@@ -359,7 +363,9 @@ def test_locker_properly_loads_extras_legacy(locker: Locker):
 content-hash = "123456789"
 """  # noqa: E800
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     repository = locker.locked_repository()
     assert len(repository.packages) == 2
@@ -399,7 +405,9 @@ def test_locker_properly_loads_subdir(locker: Locker) -> None:
 python-versions = "*"
 content-hash = "115cf985d932e9bf5f540555bbdd75decbb62cac81e399375fc19f6277f8c1d8"
 """
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     repository = locker.locked_repository()
     assert len(repository.packages) == 1
@@ -495,7 +503,9 @@ def test_locker_properly_assigns_metadata_files(locker: Locker) -> None:
     {file = "demo-1.0-py3-none-any.whl", hash = "sha256"},
 ]
 """
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     repository = locker.locked_repository()
     assert len(repository.packages) == 5
@@ -687,7 +697,9 @@ def test_locker_should_emit_warnings_if_lock_version_is_newer_but_allowed(
 """
     caplog.set_level(logging.WARNING, logger="poetry.packages.locker")
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     _ = locker.lock_data
 
@@ -717,7 +729,9 @@ def test_locker_should_raise_an_error_if_lock_version_is_newer_and_not_allowed(
 """  # noqa: E800
     caplog.set_level(logging.WARNING, logger="poetry.packages.locker")
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     with pytest.raises(RuntimeError, match="^The lock file is not compatible"):
         _ = locker.lock_data
@@ -775,7 +789,9 @@ def test_locker_should_neither_emit_warnings_nor_raise_error_for_lower_compatibl
 """
     caplog.set_level(logging.WARNING, logger="poetry.packages.locker")
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
 
     _ = locker.lock_data
 
@@ -970,7 +986,10 @@ def test_locked_repository_uses_root_dir_of_package(
 content-hash = "115cf985d932e9bf5f540555bbdd75decbb62cac81e399375fc19f6277f8c1d8"
 """  # noqa: E800
 
-    locker.lock.write(tomlkit.parse(content))
+    data = tomlkit.parse(content)
+    with open(locker.lock, "w", encoding="utf-8") as f:
+        f.write(data.as_string())
+
     create_dependency_patch = mocker.patch(
         "poetry.factory.Factory.create_dependency", autospec=True
     )
@@ -983,7 +1002,7 @@ def test_locked_repository_uses_root_dir_of_package(
     root_dir = call_kwargs["root_dir"]
     assert root_dir.match("*/lib/libA")
     # relative_to raises an exception if not relative - is_relative_to comes in py3.9
-    assert root_dir.relative_to(locker.lock.path.parent.resolve()) is not None
+    assert root_dir.relative_to(locker.lock.parent.resolve()) is not None
 
 
 @pytest.mark.parametrize(
@@ -1017,7 +1036,7 @@ def test_content_hash_with_legacy_is_compatible(
         relevant_content[key] = local_config.get(key)
 
     locker = locker.__class__(
-        lock=locker.lock.path,
+        lock=locker.lock,
         local_config=local_config,
     )