Skip to content

Find many documents

Find many documents matching the filter

main.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 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
"""Find many documents matching the filter.

The search is based on the effect of a quantum loop.
The search effectiveness depends on the number of processor threads.
"""

import anyio
from typing import Annotated
from pydantic import Field
from scruby import ReturnType, Scruby, ScrubyModel
from pprint import pprint as pp


class Car(ScrubyModel):
    """Car model."""
    brand: Annotated[str, Field(frozen=True)]
    model: Annotated[str, Field(frozen=True)]
    year: int
    power_reserve: int
    # key is always at bottom
    key: Annotated[
        str,
        Field(
            frozen=True,
            default_factory=lambda data: f"{data['brand']}:{data['model']}",
        ),
    ]


async def main() -> None:
    """Example."""
    # Activate database.
    Scruby.run()

    # Get collection `Car`.
    car_coll = Scruby(Car)

    # Create cars.
    for num in range(1, 10):
        car = Car(
            brand="Mazda",
            model=f"EZ-6 {num}",
            year=2025,
            power_reserve=600,
        )
        await car_coll.add_doc(car)

    # Find all cars.
    car_list: list[Car] | None = car_coll.find_many()
    if car_list is not None:
        pp(car_list)
    else:
        print("No cars!")

    # Find cars by brand and year.
    car_list: list[Car] | None = car_coll.find_many(
        filter_fn=lambda doc: doc.brand == "Mazda" and doc.year == 2025,
    )
    if car_list is not None:
        pp(car_list)
    else:
        print("No cars!")

    # Pagination
    car_list: list[Car] | None = car_coll.find_many(
        filter_fn=lambda doc: doc.brand == "Mazda",
        limit_docs=5,
        page_number=2,
    )
    if car_list is not None:
        pp(car_list)
    else:
        print("No cars!")

    # Sorting
    car_list: list[Car] | None = car_coll.find_many(
        filter_fn=lambda doc: doc.brand == "Mazda",
        sort_fn=lambda doc: (doc.brand, doc.updated_at),
        sort_reverse=True,
    )
    if car_list is not None:
        pp(car_list)
    else:
        print("No cars!")

    # Return cars in JSON format
    car_list: str | None = car_coll.find_many(
        filter_fn=lambda doc: doc.brand == "Mazda",
        return_type=ReturnType.JSON,
    )

    # Return cars in Dict format
    car_list: list[dict] | None = car_coll.find_many(
        filter_fn=lambda doc: doc.brand == "Mazda",
        return_type=ReturnType.DICT,
    )

    # Full database deletion.
    # Hint: The main purpose is tests.
    Scruby.napalm()


if __name__ == "__main__":
    anyio.run(main)