Skip to content

tinydantic.tinydb.storages ¤

TinyDB storage classes.

YAMLStorage ¤

Bases: Storage

Store the data in a YAML file.

Source code in src/tinydantic/tinydb/storages.py 
 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
class YAMLStorage(Storage):
    """Store the data in a YAML file."""

    def __init__(
        self,
        path: str,
        *,
        create_dirs: bool = False,
        encoding: str | None = None,
        access_mode: str = "r+",
        **kwargs: Any,
    ) -> None:
        """Create a new instance.

        Also creates the storage file, if it doesn't exist and the
        access mode is appropriate for writing.

        Note: Using an access mode other than `r` or `r+` will probably
        lead to data loss or data corruption!

        Args:
            path: Where to store the YAML data.
            create_dirs: Whether to create all missing parent
                directories of the storage file.
            encoding: The encoding used to open the storage file.
            access_mode: Mode in which the file is opened (`r`, `r+`).
            **kwargs: Additional keyword arguments passed to
                `yaml.dump()` when writing data.
        """
        super().__init__()

        self._mode = access_mode
        self.kwargs = kwargs

        if access_mode not in ("r", "rb", "r+", "rb+"):
            warnings.warn(
                "Using an `access_mode` other than 'r', 'rb', 'r+' "
                "or 'rb+' can cause data loss or corruption",
                stacklevel=2,
            )

        # Create the file if it doesn't exist and creating is allowed by
        # the access mode
        if any(
            character in self._mode for character in ("+", "w", "a")
        ):  # any of the writing modes
            touch(path, create_dirs=create_dirs)

        # Open the file for reading/writing
        self._handle = Path(path).open(mode=self._mode, encoding=encoding)  # noqa: SIM115

    def close(self) -> None:
        """Close the file object."""
        self._handle.close()

    def read(self) -> dict[str, dict[str, Any]] | None:
        """Read data from storage.

        Returns:
            A python dict containing the database data from storage,
            or `None` if the storage file is empty.
        """
        # Get the file size by moving the cursor to the file end and
        # reading its location
        self._handle.seek(0, os.SEEK_END)
        size = self._handle.tell()

        if not size:
            # File is empty, so we return ``None`` so TinyDB can
            # properly initialize the database
            return None

        # Return the cursor to the beginning of the file
        self._handle.seek(0)

        # Load the YAML contents of the file
        return yaml.safe_load(self._handle)

    def write(self, data: dict[str, dict[str, Any]]) -> None:
        """Write data to storage."""
        # Move the cursor to the beginning of the file just in case
        self._handle.seek(0)

        # Serialize the database state using the user-provided arguments
        serialized = yaml.dump(data, **self.kwargs)

        # Write the serialized data to the file
        try:
            self._handle.write(serialized)
        except io.UnsupportedOperation as exc:
            err_msg = (
                f'Cannot write to the database. Access mode is "{self._mode}"'
            )
            raise OSError(err_msg) from exc

        # Ensure the file has been written
        self._handle.flush()
        os.fsync(self._handle.fileno())

        # Remove data that is behind the new cursor in case the file has
        # gotten shorter
        self._handle.truncate()

__init__ ¤

__init__(
    path: str,
    *,
    create_dirs: bool = False,
    encoding: str | None = None,
    access_mode: str = "r+",
    **kwargs: Any
) -> None

Create a new instance.

Also creates the storage file, if it doesn't exist and the access mode is appropriate for writing.

Note: Using an access mode other than r or r+ will probably lead to data loss or data corruption!

Parameters:

Name Type Description Default
path str

Where to store the YAML data.

 required
create_dirs bool

Whether to create all missing parent directories of the storage file.

 False
encoding str | None

The encoding used to open the storage file.

 None
access_mode str

Mode in which the file is opened (r, r+).

 'r+'
**kwargs Any

Additional keyword arguments passed to yaml.dump() when writing data.

 {}
Source code in src/tinydantic/tinydb/storages.py 
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
def __init__(
    self,
    path: str,
    *,
    create_dirs: bool = False,
    encoding: str | None = None,
    access_mode: str = "r+",
    **kwargs: Any,
) -> None:
    """Create a new instance.

    Also creates the storage file, if it doesn't exist and the
    access mode is appropriate for writing.

    Note: Using an access mode other than `r` or `r+` will probably
    lead to data loss or data corruption!

    Args:
        path: Where to store the YAML data.
        create_dirs: Whether to create all missing parent
            directories of the storage file.
        encoding: The encoding used to open the storage file.
        access_mode: Mode in which the file is opened (`r`, `r+`).
        **kwargs: Additional keyword arguments passed to
            `yaml.dump()` when writing data.
    """
    super().__init__()

    self._mode = access_mode
    self.kwargs = kwargs

    if access_mode not in ("r", "rb", "r+", "rb+"):
        warnings.warn(
            "Using an `access_mode` other than 'r', 'rb', 'r+' "
            "or 'rb+' can cause data loss or corruption",
            stacklevel=2,
        )

    # Create the file if it doesn't exist and creating is allowed by
    # the access mode
    if any(
        character in self._mode for character in ("+", "w", "a")
    ):  # any of the writing modes
        touch(path, create_dirs=create_dirs)

    # Open the file for reading/writing
    self._handle = Path(path).open(mode=self._mode, encoding=encoding)  # noqa: SIM115

close ¤

close() -> None

Close the file object.

Source code in src/tinydantic/tinydb/storages.py 
75
76
77
def close(self) -> None:
    """Close the file object."""
    self._handle.close()

read ¤

read() -> dict[str, dict[str, Any]] | None

Read data from storage.

Returns:

Type Description
dict[str, dict[str, Any]] | None

A python dict containing the database data from storage,
dict[str, dict[str, Any]] | None

or None if the storage file is empty.
Source code in src/tinydantic/tinydb/storages.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def read(self) -> dict[str, dict[str, Any]] | None:
    """Read data from storage.

    Returns:
        A python dict containing the database data from storage,
        or `None` if the storage file is empty.
    """
    # Get the file size by moving the cursor to the file end and
    # reading its location
    self._handle.seek(0, os.SEEK_END)
    size = self._handle.tell()

    if not size:
        # File is empty, so we return ``None`` so TinyDB can
        # properly initialize the database
        return None

    # Return the cursor to the beginning of the file
    self._handle.seek(0)

    # Load the YAML contents of the file
    return yaml.safe_load(self._handle)

write ¤

write(data: dict[str, dict[str, Any]]) -> None

Write data to storage.

Source code in src/tinydantic/tinydb/storages.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def write(self, data: dict[str, dict[str, Any]]) -> None:
    """Write data to storage."""
    # Move the cursor to the beginning of the file just in case
    self._handle.seek(0)

    # Serialize the database state using the user-provided arguments
    serialized = yaml.dump(data, **self.kwargs)

    # Write the serialized data to the file
    try:
        self._handle.write(serialized)
    except io.UnsupportedOperation as exc:
        err_msg = (
            f'Cannot write to the database. Access mode is "{self._mode}"'
        )
        raise OSError(err_msg) from exc

    # Ensure the file has been written
    self._handle.flush()
    os.fsync(self._handle.fileno())

    # Remove data that is behind the new cursor in case the file has
    # gotten shorter
    self._handle.truncate()