Skip to content

AtomicFile

File like object supporting writing and reading in quasi atomic manor.

All reading and writing is done under lock and writing is done with temporary files and os.replace().

Example#

1
2
3
4
5
6
7
8
9
>>> tmp = getfixture("tmp_path")
>>> with AtomicFile(tmp / "some_file.txt") as file:
...     file.write_text("Example content")
...
>>> with AtomicFile(tmp / "some_file.txt") as file:
...     file.read_text()
...
'Example content'
>>>
Source code in magic_storage/_atomic_file.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class AtomicFile:
    """File like object supporting writing and reading in quasi atomic manor.

    All reading and writing is done under lock and writing is done with
    temporary files and os.replace().

    Example
    -------
    ```
    >>> tmp = getfixture("tmp_path")
    >>> with AtomicFile(tmp / "some_file.txt") as file:
    ...     file.write_text("Example content")
    ...
    >>> with AtomicFile(tmp / "some_file.txt") as file:
    ...     file.read_text()
    ...
    'Example content'
    >>>
    ```
    """

    def __init__(self, file_path: str | Path) -> None:
        file_path = Path(file_path)
        self._file = file_path.absolute()
        self._lock_file = (
            file_path.parent / f"{file_path.name}.lock"
        ).absolute()
        self._lock = FileLock(self._lock_file)

    def __enter__(self) -> AtomicFile:
        self._file.touch(0o777, True)
        logging.debug(f"Created {self._file}.")
        self._lock.acquire()
        logging.debug(f"Acquired {self._lock_file}.")
        return self

    def read_text(self, **kwargs: Any) -> str:
        """Read data from file. Requires lock to be acquired with context
        manager.

        Parameters
        ----------
        **kwargs
            Keyword arguments passed to Path.read_text()

        Returns
        -------
        str
            data from file.
        """
        assert self._lock.is_locked
        value = self._file.read_text(**kwargs)
        logging.debug(f"Read text to {self._file}.")
        return value

    def write_text(
        self,
        content: str,
        encoding: str = "utf-8",
        errors: str = "strict",
    ) -> None:
        """Write data to file. Requires lock to be acquired with context
        manager.

        Parameters
        ----------
        content : str
            Content to be saved.
        encoding : str, optional
            Encoding to use, by default "utf-8"
        errors : str, optional
            Error mode, same rules as for open(), by default "strict"
        """
        assert self._lock.is_locked
        temp = tempfile.NamedTemporaryFile(
            mode="wt",
            delete=False,
            suffix=self._file.name,
            dir=self._file.parent,
            encoding=encoding,
            errors=errors,
        )
        temp.write(content)
        temp.flush()
        temp.close()
        os.replace(temp.name, self._file)
        logging.debug(f"Wrote text to {self._file}.")

    def read_bytes(self, **kwargs: Any) -> bytes:
        """Read data from file. Requires lock to be acquired with context
        manager.

        Parameters
        ----------
        **kwargs
            Keyword arguments passed to Path.read_bytes()

        Returns
        -------
        str
            data from file.
        """
        assert self._lock.is_locked
        value = self._file.read_bytes(**kwargs)
        logging.debug(f"Read text to {self._file}.")
        return value

    def write_bytes(self, content: bytes) -> None:
        """Write data to file. Requires lock to be acquired with context
        manager.

        Parameters
        ----------
        content : str
            Content to be saved.
        """
        assert self._lock.is_locked
        temp = tempfile.NamedTemporaryFile(
            mode="wb",
            delete=False,
            suffix=self._file.name,
            dir=self._file.parent,
        )
        temp.write(content)
        temp.flush()
        temp.close()
        os.replace(temp.name, self._file)
        logging.debug(f"Wrote bytes to {self._file}.")

    def __exit__(
        self,
        _exception_type: Optional[Type[BaseException]],
        _exception_value: Optional[BaseException],
        _traceback: Traceback,
    ) -> None:
        self._lock.release()
        logging.debug(f"Released {self._lock_file}.")

read_bytes(**kwargs) #

Read data from file. Requires lock to be acquired with context manager.

Parameters:

Name Type Description Default
**kwargs Any

Keyword arguments passed to Path.read_bytes()

 {}

Returns:

Type Description
str

data from file.
Source code in magic_storage/_atomic_file.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def read_bytes(self, **kwargs: Any) -> bytes:
    """Read data from file. Requires lock to be acquired with context
    manager.

    Parameters
    ----------
    **kwargs
        Keyword arguments passed to Path.read_bytes()

    Returns
    -------
    str
        data from file.
    """
    assert self._lock.is_locked
    value = self._file.read_bytes(**kwargs)
    logging.debug(f"Read text to {self._file}.")
    return value

read_text(**kwargs) #

Read data from file. Requires lock to be acquired with context manager.

Parameters:

Name Type Description Default
**kwargs Any

Keyword arguments passed to Path.read_text()

 {}

Returns:

Type Description
str

data from file.
Source code in magic_storage/_atomic_file.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def read_text(self, **kwargs: Any) -> str:
    """Read data from file. Requires lock to be acquired with context
    manager.

    Parameters
    ----------
    **kwargs
        Keyword arguments passed to Path.read_text()

    Returns
    -------
    str
        data from file.
    """
    assert self._lock.is_locked
    value = self._file.read_text(**kwargs)
    logging.debug(f"Read text to {self._file}.")
    return value

write_bytes(content) #

Write data to file. Requires lock to be acquired with context manager.

Parameters:

Name Type Description Default
content str

Content to be saved.

 required
Source code in magic_storage/_atomic_file.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def write_bytes(self, content: bytes) -> None:
    """Write data to file. Requires lock to be acquired with context
    manager.

    Parameters
    ----------
    content : str
        Content to be saved.
    """
    assert self._lock.is_locked
    temp = tempfile.NamedTemporaryFile(
        mode="wb",
        delete=False,
        suffix=self._file.name,
        dir=self._file.parent,
    )
    temp.write(content)
    temp.flush()
    temp.close()
    os.replace(temp.name, self._file)
    logging.debug(f"Wrote bytes to {self._file}.")

write_text(content, encoding='utf-8', errors='strict') #

Write data to file. Requires lock to be acquired with context manager.

Parameters:

Name Type Description Default
content str

Content to be saved.

 required
encoding str, optional

Encoding to use, by default "utf-8"

 'utf-8'
errors str, optional

Error mode, same rules as for open(), by default "strict"

 'strict'
Source code in magic_storage/_atomic_file.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def write_text(
    self,
    content: str,
    encoding: str = "utf-8",
    errors: str = "strict",
) -> None:
    """Write data to file. Requires lock to be acquired with context
    manager.

    Parameters
    ----------
    content : str
        Content to be saved.
    encoding : str, optional
        Encoding to use, by default "utf-8"
    errors : str, optional
        Error mode, same rules as for open(), by default "strict"
    """
    assert self._lock.is_locked
    temp = tempfile.NamedTemporaryFile(
        mode="wt",
        delete=False,
        suffix=self._file.name,
        dir=self._file.parent,
        encoding=encoding,
        errors=errors,
    )
    temp.write(content)
    temp.flush()
    temp.close()
    os.replace(temp.name, self._file)
    logging.debug(f"Wrote text to {self._file}.")
Last update: August 5, 2022
Created: August 5, 2022