Skip to content

Find one document

Find one document matching the filter

main.py
"""Find one document 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 Phone(ScrubyModel):
    """Phone model."""
    brand: Annotated[str, Field(frozen=True)]
    model: Annotated[str, Field(frozen=True)]
    screen_diagonal: float
    matrix_type: str
    # 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 `Phone`.
    phone_coll = Scruby(Phone)

    # Create phone.
    phone = Phone(
        brand="Samsung",
        model="Galaxy A26",
        screen_diagonal=6.7,
        matrix_type="Super AMOLED",
    )

    # Add phone to collection.
    await phone_coll.add_doc(phone)

    # Find phone by brand.
    phone_details: Phone | None = phone_coll.find_one(
        filter_fn=lambda doc: doc.brand == "Samsung",
    )
    if phone_details is not None:
        pp(phone_details)
    else:
        print("No Phone!")

    # Find phone by model.
    phone_details: Phone | None = phone_coll.find_one(
        filter_fn=lambda doc: doc.model == "Galaxy A26",
    )
    if phone_details is not None:
        pp(phone_details)
    else:
        print("No Phone!")

    # Return phone in JSON format
    phone_details: str | None = phone_coll.find_one(
        filter_fn=lambda doc: doc.model == "Galaxy A26",
        return_type=ReturnType.JSON,
    )

    # Return phone in Dict format
    phone_details: dict | None = phone_coll.find_one(
        filter_fn=lambda doc: doc.model == "Galaxy A26",
        return_type=ReturnType.DICT,
    )

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


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