scruby 2.0.1

Asynchronous library for building and managing a hybrid database, by scheme of key-value.

Scruby (small shrub)

Asynchronous library for building and managing a hybrid database,
by scheme of key-value.

The library uses fractal-tree addressing and
the search for documents based on the effect of a quantum loop.

The size of each collection is 16|256|4096 branches,
each branch can store one or more keys.

The value of any key in collection can be obtained in 1-3 steps,
thereby achieving high performance.

The effectiveness of the search for documents based on a quantum loop,
requires a large number of processor threads.

Parameter `ScrubyConfig.HASH_REDUCE_LEFT - Scruby.run(hash_reduce_left = 7)`:
7 = 16 branches in collection (is default) -> ~16000+ docs (for development).
6 = 256 branches in collection -> ~256000+ docs (for small projects).
5 = 4096 branches in collection -> ~4096000+ (for large projects).

Installation

uv add scruby

Run

# Run Development:
uv run python main.py
# Run Production:
uv run python -OOP main.py

Usage

"""Working with keys."""

import anyio
from datetime import datetime
from zoneinfo import ZoneInfo
from typing import Annotated
from pydantic import EmailStr, Field
from pydantic_extra_types.phone_numbers import PhoneNumber, PhoneNumberValidator
from scruby import Scruby, ScrubyModel, ScrubyConfig
from pprint import pprint as pp


class User(ScrubyModel):
    """User model."""
    first_name: str = Field(strict=True)
    last_name: str = Field(strict=True)
    birthday: datetime = Field(strict=True)
    email: EmailStr = Field(strict=True)
    phone: Annotated[PhoneNumber, PhoneNumberValidator(number_format="E164")] = Field(frozen=True)
    # key is always at bottom
    key: str = Field(
        strict=True,
        frozen=True,
        default_factory=lambda data: data["phone"],
    )


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

    # Get/Create a User collection
    user_coll = await Scruby.collection(User)

    # Create user
    user = User(
        first_name="John",
        last_name="Smith",
        birthday=datetime(1970, 1, 1, tzinfo=ZoneInfo("UTC")),
        email="[e-mail]",
        phone="[phone]",
    )

    # Add user to collection
    await user_coll.add_doc(user)

    # Update user data in a collection
    await user_coll.update_doc(user)

    # Get user details
    user = user_coll.get_doc("[phone]")
    pp(user)
    user_coll.get_doc("key missing")  # => None

    # Check for the presence of a key in the collection
    user_coll.has_key("[phone]")  # => True

    # Delete a document by key
    await user_coll.delete_doc("[phone]")

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


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

"""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 pydantic import Field
from scruby import Scruby, ScrubyModel, ScrubyConfig
from pprint import pprint as pp


class Phone(ScrubyModel):
    """Phone model."""
    brand: str = Field(strict=True, frozen=True)
    model: str = Field(strict=True, frozen=True)
    screen_diagonal: float = Field(strict=True)
    matrix_type: str = Field(strict=True)
    # key is always at bottom
    key: str = Field(
        strict=True,
        frozen=True,
        default_factory=lambda data: f"{data['brand']}:{data['model']}",
    )


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

    # Get/Create a Phone collection
    phone_coll = await Scruby.collection(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!")

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


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

"""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 Scruby, ScrubyModel, ScrubyConfig
from pprint import pprint as pp


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


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

    # Get/Create a Car collection
    car_coll = await Scruby.collection(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!")

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


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