diff --git a/src/poetry/installation/executor.py b/src/poetry/installation/executor.py index 33bc924dfc0..4103da299f0 100644 --- a/src/poetry/installation/executor.py +++ b/src/poetry/installation/executor.py @@ -16,6 +16,7 @@ from cleo.io.null_io import NullIO from poetry.core.packages.utils.link import Link +from requests.utils import atomic_open from poetry.installation.chef import Chef from poetry.installation.chef import ChefBuildError @@ -29,7 +30,6 @@ from poetry.utils.authenticator import Authenticator from poetry.utils.cache import ArtifactCache from poetry.utils.env import EnvCommandError -from poetry.utils.helpers import atomic_open from poetry.utils.helpers import get_file_hash from poetry.utils.helpers import pluralize from poetry.utils.helpers import remove_directory diff --git a/src/poetry/utils/helpers.py b/src/poetry/utils/helpers.py index ab2f369aa6e..fb14e819079 100644 --- a/src/poetry/utils/helpers.py +++ b/src/poetry/utils/helpers.py @@ -14,13 +14,14 @@ from typing import TYPE_CHECKING from typing import Any +from requests.utils import atomic_open + from poetry.utils.constants import REQUESTS_TIMEOUT if TYPE_CHECKING: from collections.abc import Callable from collections.abc import Iterator - from io import BufferedWriter from poetry.core.packages.package import Package from requests import Session @@ -38,24 +39,6 @@ def directory(path: Path) -> Iterator[Path]: os.chdir(cwd) -@contextmanager -def atomic_open(filename: str | os.PathLike[str]) -> Iterator[BufferedWriter]: - """ - write a file to the disk in an atomic fashion - - Taken from requests.utils - (https://github.com/psf/requests/blob/7104ad4b135daab0ed19d8e41bd469874702342b/requests/utils.py#L296) - """ - tmp_descriptor, tmp_name = tempfile.mkstemp(dir=os.path.dirname(filename)) - try: - with os.fdopen(tmp_descriptor, "wb") as tmp_handler: - yield tmp_handler - os.replace(tmp_name, filename) - except BaseException: - os.remove(tmp_name) - raise - - def _on_rm_error(func: Callable[[str], None], path: str, exc_info: Exception) -> None: if not os.path.exists(path): return