Skip to content

Keys

Methods for working with keys.

Keys

Methods for working with keys.

Source code in src/scruby/mixins/keys.py 
 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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
class Keys:
    """Methods for working with keys."""

    @final
    async def add_doc(self, doc: Any) -> None:
        """Asynchronous method for adding document to collection.

        Args:
            doc (Any): Value of key. Type, derived from `ScrubyModel`.

        Returns:
            None.
        """
        # Check if the Model matches the collection
        if not isinstance(doc, self._class_model):
            doc_class_name = doc.__class__.__name__
            collection_name = self._class_model.__name__
            msg = (
                "Method: `add_doc` > Parameter: `doc` => "
                + f"Model `{doc_class_name}` does not match collection `{collection_name}`!"
            )
            raise TypeError(msg)

        # If a password field is present, it must not be empty
        if "password" in self.model_fields and not bool(doc.password):
            msg = "Method: `add_doc` => The `password` field is empty"
            raise ValueError(msg)

        # Get the path to the collection cell
        leaf_path, prepared_key = await self._get_leaf_path(doc.key)
        # Init a `created_at` and `updated_at` fields
        tz = ZoneInfo("UTC")
        doc.created_at = datetime.now(tz)
        doc.updated_at = datetime.now(tz)
        # Convert doc to json
        doc_json: str = doc.model_dump_json()

        async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
            # Raise an exception if the key is exists
            if await leaf_db.exists(prepared_key):
                raise KeyAlreadyExistsError()
            # Add a new document to the database
            await leaf_db.set(prepared_key, doc_json)
        # Update document counter
        await self._counter_documents(1)

    @final
    async def update_doc(self, doc: Any) -> None:
        """Asynchronous method for updating document to collection.

        Args:
            doc (Any): Value of key. Type `ScrubyModel`.

        Returns:
            None.
        """
        # Check if the Model matches the collection
        if not isinstance(doc, self._class_model):
            doc_class_name = doc.__class__.__name__
            collection_name = self._class_model.__name__
            msg = (
                f"Method: `update_doc` > Parameter: `doc` => Model `{doc_class_name}` "
                f"does not match collection `{collection_name}`!"
            )
            raise TypeError(msg)

        # If a password field is present, it must not be empty
        if "password" in self.model_fields and not bool(doc.password):
            msg = "Method: `update_doc` => The `password` field is empty"
            raise ValueError(msg)

        # Get the path to the collection cell
        leaf_path, prepared_key = await self._get_leaf_path(doc.key)
        # Update a `updated_at` field
        doc.updated_at = datetime.now(ZoneInfo("UTC"))
        # Convert doc to json.
        doc_json: str = doc.model_dump_json()

        async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
            # Raise an exception if the key is missing
            if not await leaf_db.exists(prepared_key):
                raise KeyNotExistsError()
            # Update document to the database
            await leaf_db.set(prepared_key, doc_json)

    @final
    async def get_doc(
        self,
        key: str,
        include_fields: set[str] | None = None,
        exclude_fields: set[str] | None = None,
        return_type: ReturnType = ReturnType.MODEL,
    ) -> Any | None:
        """Asynchronous method for getting document from collection the by key.

        Args:
            key (str): Key name.
            include_fields: (set[str] | None): A set of fields to include in the output.
                                               Available for `ReturnType.JSON` and `ReturnType.DICT`.
            exclude_fields: (set[str] | None): A set of fields to exclude from the output.
                                               Available for `ReturnType.JSON` and `ReturnType.DICT`.
            return_type (ReturnType): ScrubyModel, JSON-string or Dictionary.

        Returns:
            Value of key or KeyError.
        """
        if not isinstance(key, str):
            raise KeyError("The key is not a string.")

        # Get the path to the collection cell
        leaf_path, prepared_key = await self._get_leaf_path(key)

        async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
            # If the key is missing, return None
            if not await leaf_db.exists(prepared_key):
                return None

            # Get json by key
            doc = self._class_model.model_validate_json(await leaf_db.get(prepared_key))

            # Return document
            match return_type.value:
                case 1:
                    return doc
                case 2:
                    return doc.model_dump_json(include=include_fields, exclude=exclude_fields)
                case 3:
                    return doc.model_dump(include=include_fields, exclude=exclude_fields)
                case _ as unreachable:
                    assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

    @final
    async def has_key(self, key: str) -> bool:
        """Asynchronous method for checking presence of key in collection.

        Args:
            key (str): Key name.

        Returns:
            True, if the key is present.
        """
        # Get path to cell of collection.
        leaf_path, prepared_key = await self._get_leaf_path(key)

        async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
            return await leaf_db.exists(prepared_key)

    @final
    async def delete_doc(self, key: str) -> None:
        """Asynchronous method for deleting document from collection the by key.

        Args:
            key (str): Key name.

        Returns:
            None.
        """
        # The path to the database cell.
        leaf_path, prepared_key = await self._get_leaf_path(key)

        # Deleting key.
        async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
            # Raise an exception if the key is missing
            if not await leaf_db.exists(prepared_key):
                raise KeyNotExistsError()

            await leaf_db.delete(prepared_key)
            await self._counter_documents(-1)

add_doc(doc) async

Asynchronous method for adding document to collection.

Parameters:

Name Type Description Default
doc Any

Value of key. Type, derived from ScrubyModel.

 required

Returns:

Type Description
None

None.
Source code in src/scruby/mixins/keys.py 
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
@final
async def add_doc(self, doc: Any) -> None:
    """Asynchronous method for adding document to collection.

    Args:
        doc (Any): Value of key. Type, derived from `ScrubyModel`.

    Returns:
        None.
    """
    # Check if the Model matches the collection
    if not isinstance(doc, self._class_model):
        doc_class_name = doc.__class__.__name__
        collection_name = self._class_model.__name__
        msg = (
            "Method: `add_doc` > Parameter: `doc` => "
            + f"Model `{doc_class_name}` does not match collection `{collection_name}`!"
        )
        raise TypeError(msg)

    # If a password field is present, it must not be empty
    if "password" in self.model_fields and not bool(doc.password):
        msg = "Method: `add_doc` => The `password` field is empty"
        raise ValueError(msg)

    # Get the path to the collection cell
    leaf_path, prepared_key = await self._get_leaf_path(doc.key)
    # Init a `created_at` and `updated_at` fields
    tz = ZoneInfo("UTC")
    doc.created_at = datetime.now(tz)
    doc.updated_at = datetime.now(tz)
    # Convert doc to json
    doc_json: str = doc.model_dump_json()

    async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
        # Raise an exception if the key is exists
        if await leaf_db.exists(prepared_key):
            raise KeyAlreadyExistsError()
        # Add a new document to the database
        await leaf_db.set(prepared_key, doc_json)
    # Update document counter
    await self._counter_documents(1)

delete_doc(key) async

Asynchronous method for deleting document from collection the by key.

Parameters:

Name Type Description Default
key str

Key name.

 required

Returns:

Type Description
None

None.
Source code in src/scruby/mixins/keys.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@final
async def delete_doc(self, key: str) -> None:
    """Asynchronous method for deleting document from collection the by key.

    Args:
        key (str): Key name.

    Returns:
        None.
    """
    # The path to the database cell.
    leaf_path, prepared_key = await self._get_leaf_path(key)

    # Deleting key.
    async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
        # Raise an exception if the key is missing
        if not await leaf_db.exists(prepared_key):
            raise KeyNotExistsError()

        await leaf_db.delete(prepared_key)
        await self._counter_documents(-1)

get_doc(key, include_fields=None, exclude_fields=None, return_type=ReturnType.MODEL) async

Asynchronous method for getting document from collection the by key.

Parameters:

Name Type Description Default
key str

Key name.

 required
include_fields set[str] | None

(set[str] | None): A set of fields to include in the output. Available for ReturnType.JSON and ReturnType.DICT.

 None
exclude_fields set[str] | None

(set[str] | None): A set of fields to exclude from the output. Available for ReturnType.JSON and ReturnType.DICT.

 None
return_type ReturnType

ScrubyModel, JSON-string or Dictionary.

 MODEL

Returns:

Type Description
Any | None

Value of key or KeyError.
Source code in src/scruby/mixins/keys.py 
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
153
154
@final
async def get_doc(
    self,
    key: str,
    include_fields: set[str] | None = None,
    exclude_fields: set[str] | None = None,
    return_type: ReturnType = ReturnType.MODEL,
) -> Any | None:
    """Asynchronous method for getting document from collection the by key.

    Args:
        key (str): Key name.
        include_fields: (set[str] | None): A set of fields to include in the output.
                                           Available for `ReturnType.JSON` and `ReturnType.DICT`.
        exclude_fields: (set[str] | None): A set of fields to exclude from the output.
                                           Available for `ReturnType.JSON` and `ReturnType.DICT`.
        return_type (ReturnType): ScrubyModel, JSON-string or Dictionary.

    Returns:
        Value of key or KeyError.
    """
    if not isinstance(key, str):
        raise KeyError("The key is not a string.")

    # Get the path to the collection cell
    leaf_path, prepared_key = await self._get_leaf_path(key)

    async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
        # If the key is missing, return None
        if not await leaf_db.exists(prepared_key):
            return None

        # Get json by key
        doc = self._class_model.model_validate_json(await leaf_db.get(prepared_key))

        # Return document
        match return_type.value:
            case 1:
                return doc
            case 2:
                return doc.model_dump_json(include=include_fields, exclude=exclude_fields)
            case 3:
                return doc.model_dump(include=include_fields, exclude=exclude_fields)
            case _ as unreachable:
                assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

has_key(key) async

Asynchronous method for checking presence of key in collection.

Parameters:

Name Type Description Default
key str

Key name.

 required

Returns:

Type Description
bool

True, if the key is present.
Source code in src/scruby/mixins/keys.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@final
async def has_key(self, key: str) -> bool:
    """Asynchronous method for checking presence of key in collection.

    Args:
        key (str): Key name.

    Returns:
        True, if the key is present.
    """
    # Get path to cell of collection.
    leaf_path, prepared_key = await self._get_leaf_path(key)

    async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
        return await leaf_db.exists(prepared_key)

update_doc(doc) async

Asynchronous method for updating document to collection.

Parameters:

Name Type Description Default
doc Any

Value of key. Type ScrubyModel.

 required

Returns:

Type Description
None

None.
Source code in src/scruby/mixins/keys.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
103
104
105
106
107
108
@final
async def update_doc(self, doc: Any) -> None:
    """Asynchronous method for updating document to collection.

    Args:
        doc (Any): Value of key. Type `ScrubyModel`.

    Returns:
        None.
    """
    # Check if the Model matches the collection
    if not isinstance(doc, self._class_model):
        doc_class_name = doc.__class__.__name__
        collection_name = self._class_model.__name__
        msg = (
            f"Method: `update_doc` > Parameter: `doc` => Model `{doc_class_name}` "
            f"does not match collection `{collection_name}`!"
        )
        raise TypeError(msg)

    # If a password field is present, it must not be empty
    if "password" in self.model_fields and not bool(doc.password):
        msg = "Method: `update_doc` => The `password` field is empty"
        raise ValueError(msg)

    # Get the path to the collection cell
    leaf_path, prepared_key = await self._get_leaf_path(doc.key)
    # Update a `updated_at` field
    doc.updated_at = datetime.now(ZoneInfo("UTC"))
    # Convert doc to json.
    doc_json: str = doc.model_dump_json()

    async with aiodbm.open(str(leaf_path), flag="c", mode=self._mode) as leaf_db:
        # Raise an exception if the key is missing
        if not await leaf_db.exists(prepared_key):
            raise KeyNotExistsError()
        # Update document to the database
        await leaf_db.set(prepared_key, doc_json)