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}",  # {num} - there is no need to do this, this is just an example
            year=2025,
            power_reserve=600,
        )
        await car_coll.add_doc(car)

    # Find all cars.
    car_list: list[Car] | None = await 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 = await 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 = await 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 = await 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 = await 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 = await 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)