|
1 | | -"""Internal adapter around the :mod:`safeatomic` package.""" |
| 1 | +"""Internal adapter around the :mod:`safeatomic` package. |
| 2 | +
|
| 3 | +This module centralises imports from :mod:`safeatomic` and provides thin |
| 4 | +wrappers so that :mod:`atomicwrites` continues to expose the historic API. |
| 5 | +The wrappers keep the original function signatures while delegating all |
| 6 | +implementation details to :mod:`safeatomic`. |
| 7 | +""" |
2 | 8 |
|
3 | 9 | from __future__ import annotations |
4 | 10 |
|
5 | | -__all__ = ["AtomicWriter", "atomic_write", "move_atomic", "replace_atomic"] |
| 11 | +from typing import Any |
| 12 | +import inspect |
6 | 13 |
|
7 | 14 | try: # pragma: no cover - handled in tests |
8 | | - from safeatomic import ( |
9 | | - AtomicWriter, |
10 | | - atomic_write, |
11 | | - move_atomic, |
12 | | - replace_atomic, |
13 | | - ) |
| 15 | + import safeatomic as _safeatomic # type: ignore |
14 | 16 | except Exception as exc: # pragma: no cover - makes dependency explicit |
15 | 17 | raise ImportError( |
16 | 18 | "python-atomicwrites now depends on the 'safeatomic' package." |
17 | 19 | " Please install safeatomic to use this compatibility wrapper." |
18 | 20 | ) from exc |
| 21 | + |
| 22 | +__all__ = ["atomic_write", "replace_atomic", "move_atomic", "AtomicWriter"] |
| 23 | + |
| 24 | + |
| 25 | +def atomic_write(path: Any, writer_cls: type | None = None, **cls_kwargs: Any): |
| 26 | + """Proxy to :func:`safeatomic.atomic_write` keeping backward signature.""" |
| 27 | + |
| 28 | + if writer_cls is None: |
| 29 | + writer_cls = AtomicWriter |
| 30 | + return _safeatomic.atomic_write(path, writer_cls=writer_cls, **cls_kwargs) |
| 31 | + |
| 32 | + |
| 33 | +def replace_atomic(src: Any, dst: Any): |
| 34 | + """Proxy to :func:`safeatomic.replace_atomic`.""" |
| 35 | + |
| 36 | + return _safeatomic.replace_atomic(src, dst) |
| 37 | + |
| 38 | + |
| 39 | +def move_atomic(src: Any, dst: Any): |
| 40 | + """Proxy to :func:`safeatomic.move_atomic`.""" |
| 41 | + |
| 42 | + return _safeatomic.move_atomic(src, dst) |
| 43 | + |
| 44 | + |
| 45 | +class AtomicWriter(_safeatomic.AtomicWriter): |
| 46 | + """Compatibility wrapper around :class:`safeatomic.AtomicWriter`. |
| 47 | +
|
| 48 | + The original :mod:`atomicwrites` exposed an ``AtomicWriter`` with the |
| 49 | + signature ``(path, mode="w", overwrite=False, **open_kwargs)``. The |
| 50 | + underlying :mod:`safeatomic` implementation may use different parameter |
| 51 | + names (for example ``permit_overwrite`` instead of ``overwrite``). This |
| 52 | + wrapper translates the historic arguments to whatever the wrapped class |
| 53 | + expects so that existing code keeps working. |
| 54 | + """ |
| 55 | + |
| 56 | + def __init__(self, path: Any, mode: str = "w", overwrite: bool = False, **open_kwargs: Any) -> None: |
| 57 | + params = inspect.signature(_safeatomic.AtomicWriter.__init__).parameters |
| 58 | + kwargs = dict(open_kwargs) |
| 59 | + if "mode" in params: |
| 60 | + kwargs["mode"] = mode |
| 61 | + elif "file_mode" in params: |
| 62 | + kwargs["file_mode"] = mode |
| 63 | + else: |
| 64 | + kwargs["mode"] = mode |
| 65 | + |
| 66 | + if "overwrite" in params: |
| 67 | + kwargs["overwrite"] = overwrite |
| 68 | + elif "permit_overwrite" in params: |
| 69 | + kwargs["permit_overwrite"] = overwrite |
| 70 | + elif overwrite: |
| 71 | + raise TypeError("safeatomic.AtomicWriter does not support overwrite flag") |
| 72 | + |
| 73 | + if "dir" in kwargs and "dir" not in params and "tmp_dir" in params: |
| 74 | + kwargs["tmp_dir"] = kwargs.pop("dir") |
| 75 | + |
| 76 | + super().__init__(path, **kwargs) |
| 77 | + |
0 commit comments