Scruby
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|4294967296 branches,
each branch can store one or more keys.
The value of any key in collection can be obtained in 1-8 steps,
thereby achieving high performance.
The effectiveness of the search for documents based on a quantum loop,
requires a large number of processor threads.
Version 4.0
All reasons that could lead to unrealistic production requirements have been eliminated.
Now uses `aiodbm` to work with documents on the file system.
The principle of creating custom tasks has been updated.
See documentation.
Parameter `Scruby.run(hash_reduce_left = 7)`:
7 = 16 branches in collection (default) -> Docs: ~16000+, RAM: 2G+, CPU: 2+ (for development).
6 = 256 branches in collection -> Docs: ~256000+, RAM: 2G+, CPU: 2+ (for small projects).
0 = 4294967296 branches in collection -> Docs: ~4,294967296×10¹²+, RAM: 2G+, CPU: 2+ (only operations with keys are available).
If you notice the production server slowing down,
you will need to add RAM and CPU.
Installation
uv add scruby
Run
# Run Development:
uv run python main.py
# Run Production:
uv run python -OOP main.py
Usage
"""Operations 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
class User(ScrubyModel):
"""User model."""
first_name: str
last_name: str
birthday: datetime
email: EmailStr
phone: Annotated[
PhoneNumber,
PhoneNumberValidator(number_format="E164"),
Field(strict=False),
]
# key is always at bottom
key: Annotated[
str,
Field(
frozen=True,
default_factory=lambda data: data["phone"],
),
]
async def main() -> None:
"""Example."""
# Activate database.
Scruby.run()
# Get access to the collection
user_coll = Scruby(User)
# Create user
user = User(
first_name="John",
last_name="Smith",
birthday=datetime(1970, 1, 1, tzinfo=ZoneInfo("UTC")),
email="John_Smith@gmail.com",
phone="+447986123456",
)
# 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 = await user_coll.get_doc("+447986123456")
await user_coll.get_doc("key missing") # => None
# Check for the presence of a key in the collection
await user_coll.has_key("+447986123456") # => True
# Delete a document by key
await user_coll.delete_doc("+447986123456")
# Get collection name
user_coll.collection_name() # => User
# Get collection list
coll_list = Scruby.collection_list() # => ["User"]
# Get the number of documents in the collection from metadata
await user_coll.estimated_document_count() # => 1
# Get the number of documents comparable to the filter
await user_coll.count_documents(filter_fn=lambda doc: doc.first_name == "John") == 1
# Clear collection
Scruby.clear_collection("User")
# Full database deletion
# Hint: The main purpose is tests
Scruby.napalm()
if __name__ == "__main__":
anyio.run(main)
"""Operations with passwords.
To operations with a password, use only special methods.
Do not use this field directly.
"""
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 CryptModel, Scruby, ScrubyModel
class User(ScrubyModel, CryptModel):
"""User model."""
username: str
first_name: str
last_name: str
birthday: datetime
email: EmailStr
phone: Annotated[
PhoneNumber,
PhoneNumberValidator(number_format="E164"),
Field(strict=False),
]
# key is always at bottom
key: Annotated[
str,
Field(
frozen=True,
default_factory=lambda data: data["username"],
),
]
async def main() -> None:
"""Example."""
# Activate database.
Scruby.run()
# Create/get the `User` collection.
user_coll = Scruby(User)
# Create user
user = User(
username="user_1",
first_name="John",
last_name="Smith",
birthday=datetime(1970, 1, 1, tzinfo=ZoneInfo("UTC")),
email="John_Smith@gmail.com",
phone="+447986123456",
)
test_pass = "user_pass_123"
# Add user password
user.set_password(test_pass)
# Add user to collection
await user_coll.add_doc(user)
# Get user details
user_details = await user_coll.get_doc("user_1")
# Check a password
user_details.password_is_valid(test_pass) # True
# Update existing password
new_test_pass = "new_user_pass_123"
user_details.update_password(test_pass, new_test_pass)
# Update user data in a collection
await user_coll.update_doc(user_details)
# Full database deletion.
# Hint: The main purpose is tests.
Scruby.napalm()
if __name__ == "__main__":
anyio.run(main)
"""Find one document matching the filter."""
import anyio
from typing import Annotated
from pydantic import Field
from scruby import ReturnType, Scruby, ScrubyModel
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 access to the collection
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 = await phone_coll.find_one(
filter_fn=lambda doc: doc.brand == "Samsung",
)
# Find phone by model
phone_details: Phone | None = await phone_coll.find_one(
filter_fn=lambda doc: doc.model == "Galaxy A26",
)
# Return phone in JSON format
phone_details: str | None = await phone_coll.find_one(
filter_fn=lambda doc: doc.model == "Galaxy A26",
return_type=ReturnType.JSON,
)
# Return phone in Dictionary format
phone_details: dict | None = await 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)
"""Find many documents matching the filter."""
import anyio
from typing import Annotated
from pydantic import Field
from scruby import ReturnType, Scruby, ScrubyModel
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 access to the collection
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()
# 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,
)
# Pagination
car_list: list[Car] | None = await car_coll.find_many(
filter_fn=lambda doc: doc.brand == "Mazda",
limit_docs=5,
page_number=2,
)
# 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,
)
# 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 Dictionary format
car_list: list[dict] | None = await car_coll.find_many(
filter_fn=lambda doc: doc.brand == "Mazda",
return_type=ReturnType.DICT,
)
# Update one or more documents matching the filter
count_updated = await car_coll.update_many(
new_data={"brand": "BMW"},
filter_fn=lambda doc: doc.brand == "Mazda",
)
# Delete one or more documents matching the filter
count_deleted = await car_coll.delete_many(
filter_fn=lambda doc: doc.brand == "BMW",
)
# Full database deletion
# Hint: The main purpose is tests
Scruby.napalm()
if __name__ == "__main__":
anyio.run(main)
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
No source distribution files available for this release.See tutorial on generating distribution archives.
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
scruby-4.0.1-py3-none-any.whl
(41.0 kB
view details)
File details
Details for the file scruby-4.0.1-py3-none-any.whl.
File metadata
- Download URL: scruby-4.0.1-py3-none-any.whl
- Upload date:
- Size: 41.0 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
uv/0.12.1 {"installer":{"name":"uv","version":"0.12.1","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Fedora Linux","version":"44","id":"","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
97ae24c9947fc93a5d46ebc3a9d15e258a23161eb68c85277d6d895aff25c797
|
|
| MD5 |
5db92973d8abc4451c0ad1ee7fe37c2e
|
|
| BLAKE2b-256 |
0fd8ef5e6ae48ac6a6a08649bad468762f2c8275c7e701eb30acdc74f8efe968
|
