Skip to content

Update

Methods for updating documents.

Update

Methods for updating documents.

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

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

        This method is for internal use.

        Returns:
            The number of updated documents.
        """
        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,
                class_model.__name__,
                separated_hash,
                "leaf.dbm",
            ),
        )
        counter: int = 0

        if await leaf_path.exists():
            async with aiodbm.open(str(leaf_path), flag="c", mode=mode) as leaf_db:
                keys = await leaf_db.keys()

                for key in keys:
                    doc_json = await leaf_db.get(key)
                    doc = class_model.model_validate_json(doc_json)
                    if filter_fn(doc):
                        for field_name, value in new_data.items():
                            doc.__dict__[field_name] = value
                        await leaf_db.set(key, doc.model_dump_json())
                        counter += 1
        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
        mode = self._mode
        counter: int = 0

        # Run quantum loop
        with ThreadPoolExecutor(self._max_workers) as executor:
            futures: list[Future] = [
                executor.submit(
                    update_task_fn,
                    filter_fn,
                    branch_number,
                    hash_reduce_left,
                    db_root,
                    class_model,
                    mode,
                    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 
 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
@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
    mode = self._mode
    counter: int = 0

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

        for future in as_completed(futures):
            counter += await future.result()

    return counter