ramifice 0.8.45

ORM-pseudo-like API MongoDB for Python language.

ramifice

ORM-pseudo-like API MongoDB for Python language.

Ramifice is built around PyMongo.
For simulate relationship Many-to-One and Many-to-Many,
a simplified alternative (Types of selective fields with dynamic addition of elements) is used.
The project is more concentrated for web development or for applications with a graphic interface.

Supports MongoDB 3.6, 4.0, 4.2, 4.4, 5.0, 6.0, 7.0, and 8.0.
For more information see PyMongo.

Documentation

Online browsable documentation is available at https://kebasyaty.github.io/ramifice/.

Requirements

View the list of requirements.

Installation

1. Install MongoDB (if not installed):
   

2. Run:

# Fedora:
sudo dnf install gettext
gettext --version
# Ubuntu:
sudo apt install gettext
gettext --version
# Windows:
https://mlocati.github.io/articles/gettext-iconv-windows.html
gettext --version

cd project_name
uv add ramifice

3. Add config and public directories in root of your project:
   Download config directory Download public directory

Usage

It is recommended to look at examples here.

import re
import asyncio
from typing import Any
from datetime import datetime
from pprint import pprint as pp

from pymongo import AsyncMongoClient
from ramifice import (
    NamedTuple,
    model,
    translations,
    Migration,
    to_human_size,
)
from ramifice.fields import (
    ImageField,
    PasswordField,
    TextField,
)


@model(service_name="Accounts")
class User:
    """Model of User."""

    def fields(self) -> None:
        """Adding fields."""
        # For custom translations.
        gettext = translations.gettext
        # ngettext = translations.ngettext
        self.avatar = ImageField(
            label=gettext("Avatar"),
            default="public/media/default/no-photo.png",
            # Directory for images inside media directory.
            target_dir="users/avatars",
            # Available 4 sizes from lg to xs or None.
            # Hint: By default = None
            thumbnails={"lg": 512, "md": 256, "sm": 128, "xs": 64},
            # The maximum size of the original image in bytes.
            # Hint: By default = 2 MB
            max_size=524288,  # 0.5 MB = 512 KB = 524288 Bytes (in binary)
            warning=[
                gettext("Maximum size: {}").format(to_human_size(524288)),
            ],
        )
        self.username = TextField(
            label=gettext("Username"),
            maxlength=150,
            required=True,
            unique=True,
            warning=[
                gettext("Allowed chars: {}").format("a-z A-Z 0-9 _"),
            ],
        )
        self.password = PasswordField(
            label=gettext("Password"),
        )
        self.сonfirm_password = PasswordField(
            label=gettext("Confirm password"),
            # If true, the value of this field is not saved in the database.
            ignored=True,
        )

    # Optional method
    async def add_validation(self) -> NamedTuple:
        """Additional validation of fields."""
        gettext = translations.gettext
        cd, err = self.get_clean_data()

        # Check username
        if re.match(r"^[a-zA-Z0-9_]+$", cd.username) is None:
            err.update("username", gettext("Allowed chars: {}").format("a-z A-Z 0-9 _"))

        # Check password
        if cd._id is None and (cd.password != cd.сonfirm_password):
            err.update("password", gettext("Passwords do not match!"))

        return err


async def main():
    client = AsyncMongoClient()

    await Migration(
        database_name="test_db",
        mongo_client=client,
    ).migrate()

    # If you need to change the language of translation.
    # Hint: For Ramifice by default = "en"
    translations.change_locale("en")

    user = User()
    # user.avatar.from_path("public/media/default/no-photo.png")
    user.username.value = "pythondev"
    user.password.value = "12345678"
    user.сonfirm_password.value = "12345678"

    # Create User.
    if not await user.save():
        # Convenient to use during development.
        user.print_err()

    # Update User.
    user.username.value = "pythondev_123"
    if not await user.save():
        user.print_err()

    print("User details:")
    user_details = await User.find_one_to_raw_doc(
        # {"_id": user.id.value}
        {f"username": user.username.value}
    )
    if user_details is not None:
        pp(user_details)
    else:
        print("No User!")

    # Close connection.
    await client.close()


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

Model Parameters

( only service_name is a required parameter )

Parameter	Default	Description
service_name	no	Examples: Accounts | Smartphones | Washing machines | etc ...
fixture_name	None	The name of the fixture in the config/fixtures directory (without extension). Examples: SiteSettings | AppSettings | etc ...
db_query_docs_limit	1000	Limiting the number of request results.
is_create_doc	True	Can a Model create new documents in a collection? Set to False if you only need one document in the collection and the Model is using a fixture.
is_update_doc	True	Can a Model update documents in a collection?
is_delete_doc	True	Can a Model remove documents from a collection?

Example:

@model(
    service_name="ServiceName",
    fixture_name="FixtureName",
    db_query_docs_limit=1000,
    is_create_doc = True,
    is_update_doc = True,
    is_delete_doc = True,
)
class User:
    def fields(self):
        self.username = TextField(
            label=gettext("Username"),
            required=True,
            unique=True,
        )

Class methods

List of frequently used methods:

# Gets an estimate of the count of documents in a collection using collection metadata.
count: int = await User.estimated_document_count()

# Gets an estimate of the count of documents in a collection using collection metadata.
q_filter = {"first_name": "John"}
count: int = await User.count_documents(q_filter)

# Runs an aggregation framework pipeline.
from bson.bson import BSON
pipeline = [
    {"$unwind": "$tags"},
    {"$group": {"_id": "$tags", "count": {"$sum": 1}}},
    {"$sort": BSON([("count", -1), ("_id", -1)])},
]
docs = await User.aggregate(pipeline)

# Finds the distinct values for a specified field across a single collection.
q_filter = "key_name"
values = await User.distinct(q_filter)

# Get collection name.
name = await User.collection_name()

# The full name is of the form database_name.collection_name.
name = await User.collection_full_name()

# Get AsyncBatabase for the current Model.
database = await User.database()

# Get AsyncCollection for the current Model.
collection = await User.collection()

# Find a single document.
q_filter = {"email": "[e-mail]"}
mongo_doc = await User.find_one(q_filter)

# Create object instance from Mongo document.
q_filter = {"email": "[e-mail]"}
mongo_doc = await User.find_one(q_filter)
user = User.from_mongo_doc(mongo_doc)

# Find a single document and converting to raw document.
q_filter = {"email": "[e-mail]"}
raw_doc = await User.find_one_to_raw_doc(q_filter)

# Find a single document and convert it to a Model instance.
q_filter = {"email": "[e-mail]"}
user = await User.find_one_to_instance(q_filter)

# Find a single document and convert it to a JSON string.
q_filter = {"email": "[e-mail]"}
json = await User.find_one_to_json(q_filter)

# Find a single document and delete it.
q_filter = {"email": "[e-mail]"}
delete_result = await User.delete_one(q_filter)

# Find a single document and delete it, return original.
q_filter = {"email": "[e-mail]"}
mongo_doc = await User.find_one_and_delete(q_filter)

# Find documents.
q_filter = {"first_name": "John"}
mongo_docs = await User.find_many(q_filter)

# Find documents and convert to a raw documents.
q_filter = {"first_name": "John"}
raw_docs = await User.find_many_to_raw_docs(q_filter)

# Find documents and convert to a json string.
q_filter = {"email": "[e-mail]"}
json = await User.find_many_to_json(q_filter)

# Find documents matching with Model.
q_filter = {"email": "[e-mail]"}
delete_result = await User.delete_many(q_filter)

# Creates an index on this collection.
from pymongo import ASCENDING
keys = [("email", ASCENDING)]
result: str = await User.create_index(keys, name="idx_email")

# Drops the specified index on this collection.
User.drop_index("idx_email")

# Create one or more indexes on this collection.
from pymongo import ASCENDING, DESCENDING
index_1 = IndexModel([("username", DESCENDING), ("email", ASCENDING)], name="idx_username_email")
index_2 = IndexModel([("first_name", DESCENDING)], name="idx_first_name")
result: list[str] = await User.create_indexes([index_1, index_2])

# Drops all indexes on this collection.
User.drop_index()

# Get information on this collection’s indexes.
result = await User.index_information()

# Get a cursor over the index documents for this collection.
async for index in await User.list_indexes():
    print(index)

# Units Management.
# Management for `choices` parameter in dynamic field types.
# Units are stored in a separate collection.
from ramifice import Unit
unit = Unit(
  field="field_name",  # The name of the dynamic field.
  title={"en": "Title", "ru": "Заголовок"},  # The name of the choice item.
  value="Some text ...",  # The value of the choice item.
                          # Hint: float | int | str
  is_delete=False, # True - if you need to remove the item of choice.
                   # by default = False (add item to choice)
)
await User.unit_manager(unit)

Instance methods

List of frequently used methods:

# Check data validity.
# The main use is to check data from web forms.
# It is also used to verify Models that do not migrate to the database.
user = User()
if not await user.is_valid():
    user.print_err()  # Convenient to use during development.

# Create or update document in database.
# This method pre-uses the `check` method.
user = User()
if not await user.save():
    user.print_err()  # Convenient to use during development.

# Delete document from database.
user = User()
await user.delete()
# or
await user.delete(remove_files=False)

# Verification, replacement and recoverang of password.
user = User()
await user.verify_password(password="12345678")
await user.update_password(  # + verify_password
  old_password="12345678",
  new_password="O2eA4GIr38KGGlS",
)

General auxiliary methods

from ramifice import to_human_size
from ramifice.utils.tools import (
    get_file_size,
    hash_to_obj_id,
    is_color,
    is_email,
    is_ip,
    is_mongo_id,
    is_password,
    is_phone,
    is_url,
    normal_email,
)

# Validate Password.
if is_password("12345678"):
    ...

# Validate Email address.
if await is_email("[e-mail]"):
    ...

# Normalizing email address.
# Use this before requeste to a database.
# For example, on the login page.
email: str | None = normal_email("[e-mail]")  # None, if not valid

# Validate URL address.
if is_url("https://www.google.com"):
    ...

# Validate IP address.
if is_ip("127.0.0.1"):
    ...

# Validate Color code.
if is_color("#000"):
    ...

# Validate Phone number.
if is_phone("[phone]"):
    ...

# Validation of the Mongodb identifier.
if is_mongo_id("666f6f2d6261722d71757578"):
    ...

# Get ObjectId from hash string.
from bson.objectid import ObjectId
_id: ObjectId | None = hash_to_obj_id("666f6f2d6261722d71757578")

# Convert number of bytes to readable format.
size: str = to_human_size(2097152)  # => 2 MB

# Get file size in bytes.
path = "public/media/default/no_doc.odt"
size: int = get_file_size(path)  # => 9843

Changelog

View the change history.

License

This project is licensed under the MIT.