Skip to content

Keys

Methods for working with keys.

Keys

Methods for working with keys.

Source code in src/scruby/mixins/keys.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
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
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 = (
                f"(add_doc) Parameter `doc` => Model `{doc_class_name}` does not match collection `{collection_name}`!"
            )
            raise TypeError(msg)
        # The path to cell of collection.
        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()
        # Write key-value to collection.
        if await leaf_path.exists():
            # Add new key.
            data_json: bytes = await leaf_path.read_bytes()
            data: dict = orjson.loads(data_json) or {}
            try:
                data[prepared_key]
            except KeyError:
                data[prepared_key] = doc_json
                await leaf_path.write_bytes(orjson.dumps(data))
            else:
                raise KeyAlreadyExistsError()
        else:
            # Add new document to a blank leaf.
            await leaf_path.write_bytes(orjson.dumps({prepared_key: doc_json}))
        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"(update_doc) Parameter `doc` => Model `{doc_class_name}` "
                f"does not match collection `{collection_name}`!"
            )
            raise TypeError(msg)
        # The path to cell of collection.
        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()
        # Update the existing key.
        if await leaf_path.exists():
            # Update the existing key.
            data_json: bytes = await leaf_path.read_bytes()
            data: dict = orjson.loads(data_json) or {}
            try:
                data[prepared_key]
                data[prepared_key] = doc_json
                await leaf_path.write_bytes(orjson.dumps(data))
            except KeyError:
                err = KeyNotExistsError()
                raise err from None
        else:
            msg: str = f"`update_doc` - The key `{doc.key}` is missing!"
            raise KeyError(msg)

    @final
    async def get_doc(self, key: str) -> Any:
        """Asynchronous method for getting document from collection the by key.

        Args:
            key (str): Key name.

        Returns:
            Value of key or KeyError.
        """
        # The path to the database cell.
        leaf_path, prepared_key = await self._get_leaf_path(key)
        # Get value of key.
        if await leaf_path.exists():
            data_json: bytes = await leaf_path.read_bytes()
            data: dict = orjson.loads(data_json) or {}
            obj: Any = self._class_model.model_validate_json(data[prepared_key])
            return obj
        msg: str = f"`get_doc` - The key `{key}` is missing!"
        raise KeyError(msg)

    @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)
        # Checking whether there is a key.
        if await leaf_path.exists():
            data_json: bytes = await leaf_path.read_bytes()
            data: dict = orjson.loads(data_json) or {}
            try:
                data[prepared_key]
                return True
            except KeyError:
                return False
        return False

    @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.
        if await leaf_path.exists():
            data_json: bytes = await leaf_path.read_bytes()
            data: dict = orjson.loads(data_json) or {}
            del data[prepared_key]
            await leaf_path.write_bytes(orjson.dumps(data))
            await self._counter_documents(-1)
            return
        msg: str = f"`delete_doc` - The key `{key}` is missing!"
        raise KeyError(msg)

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 
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
@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 = (
            f"(add_doc) Parameter `doc` => Model `{doc_class_name}` does not match collection `{collection_name}`!"
        )
        raise TypeError(msg)
    # The path to cell of collection.
    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()
    # Write key-value to collection.
    if await leaf_path.exists():
        # Add new key.
        data_json: bytes = await leaf_path.read_bytes()
        data: dict = orjson.loads(data_json) or {}
        try:
            data[prepared_key]
        except KeyError:
            data[prepared_key] = doc_json
            await leaf_path.write_bytes(orjson.dumps(data))
        else:
            raise KeyAlreadyExistsError()
    else:
        # Add new document to a blank leaf.
        await leaf_path.write_bytes(orjson.dumps({prepared_key: doc_json}))
    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 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@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.
    if await leaf_path.exists():
        data_json: bytes = await leaf_path.read_bytes()
        data: dict = orjson.loads(data_json) or {}
        del data[prepared_key]
        await leaf_path.write_bytes(orjson.dumps(data))
        await self._counter_documents(-1)
        return
    msg: str = f"`delete_doc` - The key `{key}` is missing!"
    raise KeyError(msg)

get_doc(key) async

Asynchronous method for getting document from collection the by key.

Parameters:

Name Type Description Default
key str

Key name.

 required

Returns:

Type Description
Any

Value of key or KeyError.
Source code in src/scruby/mixins/keys.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@final
async def get_doc(self, key: str) -> Any:
    """Asynchronous method for getting document from collection the by key.

    Args:
        key (str): Key name.

    Returns:
        Value of key or KeyError.
    """
    # The path to the database cell.
    leaf_path, prepared_key = await self._get_leaf_path(key)
    # Get value of key.
    if await leaf_path.exists():
        data_json: bytes = await leaf_path.read_bytes()
        data: dict = orjson.loads(data_json) or {}
        obj: Any = self._class_model.model_validate_json(data[prepared_key])
        return obj
    msg: str = f"`get_doc` - The key `{key}` is missing!"
    raise KeyError(msg)

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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@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)
    # Checking whether there is a key.
    if await leaf_path.exists():
        data_json: bytes = await leaf_path.read_bytes()
        data: dict = orjson.loads(data_json) or {}
        try:
            data[prepared_key]
            return True
        except KeyError:
            return False
    return False

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 
 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
@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"(update_doc) Parameter `doc` => Model `{doc_class_name}` "
            f"does not match collection `{collection_name}`!"
        )
        raise TypeError(msg)
    # The path to cell of collection.
    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()
    # Update the existing key.
    if await leaf_path.exists():
        # Update the existing key.
        data_json: bytes = await leaf_path.read_bytes()
        data: dict = orjson.loads(data_json) or {}
        try:
            data[prepared_key]
            data[prepared_key] = doc_json
            await leaf_path.write_bytes(orjson.dumps(data))
        except KeyError:
            err = KeyNotExistsError()
            raise err from None
    else:
        msg: str = f"`update_doc` - The key `{doc.key}` is missing!"
        raise KeyError(msg)