Skip to content

Update

Methods for updating documents.

Update

Methods for updating documents.

Source code in src/scruby/mixins/update.py 
 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
class Update:
    """Methods for updating documents."""

    @final
    @staticmethod
    async def _task_update(
        filter_fn: Callable,
        db_root: str,
        hash_reduce_left: int,
        branch_number: int,
        class_model: Any,
        new_data: dict[str, Any],
    ) -> int:
        """Asynchronous task for find documents.

        This method is for internal use.

        Returns:
            The number of updated documents.
        """
        collection_name = class_model.__name__
        branch_number_as_hash: str = f"{branch_number:08x}"[hash_reduce_left:]
        separated_hash: str = "/".join(list(branch_number_as_hash))
        leaf_path = Path(
            *(
                db_root,
                collection_name,
                separated_hash,
                "leaf.json",
            ),
        )
        counter: int = 0

        if await leaf_path.exists():
            data_json: bytes = await leaf_path.read_bytes()
            data: dict[str, str] = orjson.loads(data_json) or {}
            new_state: dict[str, str] = {}
            for doc_name, val in data.items():
                doc = class_model.model_validate_json(val)
                if filter_fn(doc):
                    for field_name, value in new_data.items():
                        doc.__dict__[field_name] = value
                    new_state[doc_name] = doc.model_dump_json()
                    # Save updated documents to cache
                    match hash_reduce_left:
                        case 7:
                            DocCache.cache[collection_name][branch_number_as_hash[0]][doc_name] = doc
                        case 6:
                            DocCache.cache[collection_name][branch_number_as_hash[0]][branch_number_as_hash[1]][
                                doc_name
                            ] = doc
                        case 5:
                            DocCache.cache[collection_name][branch_number_as_hash[0]][branch_number_as_hash[1]][
                                branch_number_as_hash[2]
                            ][doc_name] = doc
                        case _ as unreachable:
                            assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]
                    # Update counter
                    counter += 1
                else:
                    new_state[doc_name] = val
            # Save updated documents to the database
            await leaf_path.write_bytes(orjson.dumps(new_state))
        return counter

    @final
    async def update_many(
        self,
        new_data: dict[str, Any],
        filter_fn: Callable = lambda _: True,
    ) -> int:
        """Asynchronous method for updates one or more documents matching the filter.

        Attention:
            - For a complex case, a custom task may be needed.
            - See documentation on creating custom tasks.
            - The search is based on the effect of a quantum loop.
            - The search effectiveness depends on the number of processor threads.

        Args:
            filter_fn (Callable): A function that execute the conditions of filtering.
            new_data (dict[str, Any]): New data for the fields that need to be updated.

        Returns:
            The number of updated documents.
        """
        # Variable initialization
        hash_reduce_left: int = self._hash_reduce_left
        assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `update_many` method."

        update_task_fn: Callable = self._task_update
        branch_numbers: range = range(self._max_number_branch)
        hash_reduce_left: int = self._hash_reduce_left
        db_root: str = self._db_root
        class_model: Any = self._class_model
        counter: int = 0

        # Run quantum loop
        with ThreadPoolExecutor(self._max_workers) as executor:
            futures: list[Future] = [
                executor.submit(
                    update_task_fn,
                    filter_fn,
                    db_root,
                    hash_reduce_left,
                    branch_number,
                    class_model,
                    copy.deepcopy(new_data),
                )
                for branch_number in branch_numbers
            ]
            for future in as_completed(futures):
                counter += await future.result()
        return counter

update_many(new_data, filter_fn=lambda _: True) async

Asynchronous method for updates one or more documents matching the filter.

Attention
  • For a complex case, a custom task may be needed.
  • See documentation on creating custom tasks.
  • The search is based on the effect of a quantum loop.
  • The search effectiveness depends on the number of processor threads.

Parameters:

Name Type Description Default
filter_fn Callable

A function that execute the conditions of filtering.

 lambda _: True
new_data dict[str, Any]

New data for the fields that need to be updated.

 required

Returns:

Type Description
int

The number of updated documents.
Source code in src/scruby/mixins/update.py 
 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
@final
async def update_many(
    self,
    new_data: dict[str, Any],
    filter_fn: Callable = lambda _: True,
) -> int:
    """Asynchronous method for updates one or more documents matching the filter.

    Attention:
        - For a complex case, a custom task may be needed.
        - See documentation on creating custom tasks.
        - The search is based on the effect of a quantum loop.
        - The search effectiveness depends on the number of processor threads.

    Args:
        filter_fn (Callable): A function that execute the conditions of filtering.
        new_data (dict[str, Any]): New data for the fields that need to be updated.

    Returns:
        The number of updated documents.
    """
    # Variable initialization
    hash_reduce_left: int = self._hash_reduce_left
    assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `update_many` method."

    update_task_fn: Callable = self._task_update
    branch_numbers: range = range(self._max_number_branch)
    hash_reduce_left: int = self._hash_reduce_left
    db_root: str = self._db_root
    class_model: Any = self._class_model
    counter: int = 0

    # Run quantum loop
    with ThreadPoolExecutor(self._max_workers) as executor:
        futures: list[Future] = [
            executor.submit(
                update_task_fn,
                filter_fn,
                db_root,
                hash_reduce_left,
                branch_number,
                class_model,
                copy.deepcopy(new_data),
            )
            for branch_number in branch_numbers
        ]
        for future in as_completed(futures):
            counter += await future.result()
    return counter