Skip to content

Find

Quantum methods for searching documents.

Find

Quantum methods for searching documents.

Source code in src/scruby/mixins/find.py 
 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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class Find:
    """Quantum methods for searching documents."""

    @final
    @staticmethod
    def _task_find(
        filter_fn: Callable,
        hash_reduce_left: int,
        branch_number: int,
        class_model: Any,
        stop_event: Event,
    ) -> list[Any] | None:
        """Task for find documents.

        This method is for internal use.

        Returns:
            List of documents or None.
        """
        # Suppress warning - RuntimeWarning: coroutine 'Find._task_find' was never awaited
        warnings.filterwarnings("ignore", category=RuntimeWarning)
        # Variable initialization
        collection_name = class_model.__name__
        branch_number_as_hash: str = f"{branch_number:08x}"[hash_reduce_left:]
        docs: dict[str, Any] = {}
        result: list[Any] = []

        match hash_reduce_left:
            case 7:
                docs = DocCache.cache[collection_name][branch_number_as_hash[0]]
            case 6:
                docs = DocCache.cache[collection_name][branch_number_as_hash[0]][branch_number_as_hash[1]]
            case 5:
                docs = DocCache.cache[collection_name][branch_number_as_hash[0]][branch_number_as_hash[1]][
                    branch_number_as_hash[2]
                ]
            case _ as unreachable:
                assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

        for _, doc in docs.items():
            if stop_event.is_set():
                return None
            if filter_fn(doc):
                result.append(doc)
        return result or None

    @final
    def find_one(
        self,
        filter_fn: Callable,
        include_fields: set[str] | None = None,
        exclude_fields: set[str] | None = None,
        return_type: ReturnType = ReturnType.MODEL,
    ) -> Any | None:
        """Synchronous method for find one document matching the filter.

        Attention:
            - 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.
            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:
            Document or None.
        """
        # Variable initialization
        hash_reduce_left: int = self._hash_reduce_left
        assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `find_one` method."

        model_dump_kwargs = {"include": include_fields, "exclude": exclude_fields}
        search_task_fn: Callable = self._task_find
        branch_numbers: range = range(self._max_number_branch)
        class_model: Any = self._class_model
        stop_signal = Event()
        doc: Any | None = None

        # Run quantum loop
        with ThreadPoolExecutor(self._max_workers) as executor:
            futures: list[Future] = [
                executor.submit(
                    search_task_fn,
                    filter_fn,
                    hash_reduce_left,
                    branch_number,
                    class_model,
                    stop_signal,
                )
                for branch_number in branch_numbers
            ]
            for future in as_completed(futures):
                docs = future.result()
                if docs is not None:
                    # Get first document
                    doc = docs[0]
                    # Cancel all pending tasks in the queue instantly
                    executor.shutdown(wait=False, cancel_futures=True)
                    # Trigger the event to tell running tasks to exit
                    stop_signal.set()
                    # Stop loop
                    break
        # Return document
        match return_type.value:
            case 1:
                return doc
            case 2:
                return doc.model_dump_json(**model_dump_kwargs) if doc is not None else None
            case 3:
                return doc.model_dump(**model_dump_kwargs) if doc is not None else None
            case _ as unreachable:
                assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

    @final
    def find_many(
        self,
        filter_fn: Callable = lambda _: True,
        limit_docs: int = 100,
        page_number: int = 1,
        sort_fn: Callable | None = lambda doc: doc.created_at,
        sort_reverse: bool = True,
        include_fields: set[str] | None = None,
        exclude_fields: set[str] | None = None,
        return_type: ReturnType = ReturnType.MODEL,
    ) -> list[Any] | str | None:
        """Synchronous method for find many documents matching the filter.

        Attention:
            - 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.
                                  By default, it searches all documents.
            limit_docs (int): Limiting the number of documents.
                              Default = 100.
            page_number (int): For pagination.
                               Default = 1.
                               Number of documents per page = limit_docs.
            sort_fn (Callable | None): Sort the list of documents.
                                       By default, documents are sorted by creation date.
            sort_reverse: (bool): Sorting direction.
                                  By default, sort descending (newest to oldest).
            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:
            Document list or None.
        """
        if __debug__:
            if limit_docs <= 0:
                msg = "Method: `find_many` => The `limit_docs` parameter must not be less than one."
                raise AssertionError(msg)
            if page_number <= 0:
                msg = "Method: `find_many` => The `page_number` parameter must not be less than one."
                raise AssertionError(msg)
        # Variable initialization
        hash_reduce_left: int = self._hash_reduce_left
        assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `find_many` method."

        model_dump_kwargs = {"include": include_fields, "exclude": exclude_fields}
        search_task_fn: Callable = self._task_find
        branch_numbers: range = range(self._max_number_branch)
        class_model: Any = self._class_model
        stop_signal = Event()
        stop_outer_loop: bool = False
        counter: int = 0
        number_docs_skippe: int = limit_docs * (page_number - 1) if page_number > 1 else 0
        result: list[Any] = []

        # Run quantum loop
        with ThreadPoolExecutor(self._max_workers) as executor:
            futures: list[Future] = [
                executor.submit(
                    search_task_fn,
                    filter_fn,
                    hash_reduce_left,
                    branch_number,
                    class_model,
                    stop_signal,
                )
                for branch_number in branch_numbers
            ]
            for future in as_completed(futures):
                docs = future.result()
                if docs is not None:
                    for doc in docs:
                        if number_docs_skippe == 0:
                            if counter >= limit_docs:
                                # Cancel all pending tasks in the queue instantly
                                executor.shutdown(wait=False, cancel_futures=True)
                                # Trigger the event to tell running tasks to exit
                                stop_signal.set()
                                # Stop loops
                                stop_outer_loop = True
                                break
                            result.append(doc)
                            counter += 1
                        else:
                            number_docs_skippe -= 1
                if stop_outer_loop:
                    break
        # Sorting
        if sort_fn is not None:
            result.sort(key=sort_fn, reverse=sort_reverse)
        # Return a document list
        match return_type.value:
            case 1:
                return result or None
            case 2:
                return (
                    f"[{','.join([doc.model_dump_json(**model_dump_kwargs) for doc in result])}]"
                    if result is not None
                    else None
                )
            case 3:
                return [doc.model_dump(**model_dump_kwargs) for doc in result] if result is not None else None
            case _ as unreachable:
                assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

find_many(filter_fn=lambda _: True, limit_docs=100, page_number=1, sort_fn=lambda doc: doc.created_at, sort_reverse=True, include_fields=None, exclude_fields=None, return_type=ReturnType.MODEL)

Synchronous method for find many documents matching the filter.

Attention
  • 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. By default, it searches all documents.

 lambda _: True
limit_docs int

Limiting the number of documents. Default = 100.

 100
page_number int

For pagination. Default = 1. Number of documents per page = limit_docs.

 1
sort_fn Callable | None

Sort the list of documents. By default, documents are sorted by creation date.

 lambda doc: created_at
sort_reverse bool

(bool): Sorting direction. By default, sort descending (newest to oldest).

 True
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
list[Any] | str | None

Document list or None.
Source code in src/scruby/mixins/find.py 
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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
@final
def find_many(
    self,
    filter_fn: Callable = lambda _: True,
    limit_docs: int = 100,
    page_number: int = 1,
    sort_fn: Callable | None = lambda doc: doc.created_at,
    sort_reverse: bool = True,
    include_fields: set[str] | None = None,
    exclude_fields: set[str] | None = None,
    return_type: ReturnType = ReturnType.MODEL,
) -> list[Any] | str | None:
    """Synchronous method for find many documents matching the filter.

    Attention:
        - 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.
                              By default, it searches all documents.
        limit_docs (int): Limiting the number of documents.
                          Default = 100.
        page_number (int): For pagination.
                           Default = 1.
                           Number of documents per page = limit_docs.
        sort_fn (Callable | None): Sort the list of documents.
                                   By default, documents are sorted by creation date.
        sort_reverse: (bool): Sorting direction.
                              By default, sort descending (newest to oldest).
        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:
        Document list or None.
    """
    if __debug__:
        if limit_docs <= 0:
            msg = "Method: `find_many` => The `limit_docs` parameter must not be less than one."
            raise AssertionError(msg)
        if page_number <= 0:
            msg = "Method: `find_many` => The `page_number` parameter must not be less than one."
            raise AssertionError(msg)
    # Variable initialization
    hash_reduce_left: int = self._hash_reduce_left
    assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `find_many` method."

    model_dump_kwargs = {"include": include_fields, "exclude": exclude_fields}
    search_task_fn: Callable = self._task_find
    branch_numbers: range = range(self._max_number_branch)
    class_model: Any = self._class_model
    stop_signal = Event()
    stop_outer_loop: bool = False
    counter: int = 0
    number_docs_skippe: int = limit_docs * (page_number - 1) if page_number > 1 else 0
    result: list[Any] = []

    # Run quantum loop
    with ThreadPoolExecutor(self._max_workers) as executor:
        futures: list[Future] = [
            executor.submit(
                search_task_fn,
                filter_fn,
                hash_reduce_left,
                branch_number,
                class_model,
                stop_signal,
            )
            for branch_number in branch_numbers
        ]
        for future in as_completed(futures):
            docs = future.result()
            if docs is not None:
                for doc in docs:
                    if number_docs_skippe == 0:
                        if counter >= limit_docs:
                            # Cancel all pending tasks in the queue instantly
                            executor.shutdown(wait=False, cancel_futures=True)
                            # Trigger the event to tell running tasks to exit
                            stop_signal.set()
                            # Stop loops
                            stop_outer_loop = True
                            break
                        result.append(doc)
                        counter += 1
                    else:
                        number_docs_skippe -= 1
            if stop_outer_loop:
                break
    # Sorting
    if sort_fn is not None:
        result.sort(key=sort_fn, reverse=sort_reverse)
    # Return a document list
    match return_type.value:
        case 1:
            return result or None
        case 2:
            return (
                f"[{','.join([doc.model_dump_json(**model_dump_kwargs) for doc in result])}]"
                if result is not None
                else None
            )
        case 3:
            return [doc.model_dump(**model_dump_kwargs) for doc in result] if result is not None else None
        case _ as unreachable:
            assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

find_one(filter_fn, include_fields=None, exclude_fields=None, return_type=ReturnType.MODEL)

Synchronous method for find one document matching the filter.

Attention
  • 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.

 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

Document or None.
Source code in src/scruby/mixins/find.py 
 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
@final
def find_one(
    self,
    filter_fn: Callable,
    include_fields: set[str] | None = None,
    exclude_fields: set[str] | None = None,
    return_type: ReturnType = ReturnType.MODEL,
) -> Any | None:
    """Synchronous method for find one document matching the filter.

    Attention:
        - 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.
        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:
        Document or None.
    """
    # Variable initialization
    hash_reduce_left: int = self._hash_reduce_left
    assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `find_one` method."

    model_dump_kwargs = {"include": include_fields, "exclude": exclude_fields}
    search_task_fn: Callable = self._task_find
    branch_numbers: range = range(self._max_number_branch)
    class_model: Any = self._class_model
    stop_signal = Event()
    doc: Any | None = None

    # Run quantum loop
    with ThreadPoolExecutor(self._max_workers) as executor:
        futures: list[Future] = [
            executor.submit(
                search_task_fn,
                filter_fn,
                hash_reduce_left,
                branch_number,
                class_model,
                stop_signal,
            )
            for branch_number in branch_numbers
        ]
        for future in as_completed(futures):
            docs = future.result()
            if docs is not None:
                # Get first document
                doc = docs[0]
                # Cancel all pending tasks in the queue instantly
                executor.shutdown(wait=False, cancel_futures=True)
                # Trigger the event to tell running tasks to exit
                stop_signal.set()
                # Stop loop
                break
    # Return document
    match return_type.value:
        case 1:
            return doc
        case 2:
            return doc.model_dump_json(**model_dump_kwargs) if doc is not None else None
        case 3:
            return doc.model_dump(**model_dump_kwargs) if doc is not None else None
        case _ as unreachable:
            assert_never(Never(unreachable))  # pyrefly: ignore[not-callable]

ReturnType

Bases: Enum

Return type for a find_one and find_many methods.

Members
  • MODEL: ScrubyModel type.
  • JSON: JSON-string type.
  • DICT: Dictionary type.
Source code in src/scruby/mixins/find.py 
21
22
23
24
25
26
27
28
29
30
31
32
class ReturnType(Enum):
    """Return type for a find_one and find_many methods.

    Members:
        - `MODEL:` ScrubyModel type.
        - `JSON:` JSON-string type.
        - `DICT:` Dictionary type.
    """

    MODEL = 1
    JSON = 2
    DICT = 3