API Reference

Document

class butty.Document

async save(*, mode: Literal['auto', 'update', 'insert', 'upsert'] = 'auto') → T

Save the document instance to database.

Parameters:

mode – Save operation mode: - “auto”: Insert if document has no identity (None), otherwise update - “insert”: Force insert (raises DuplicateKeyError if identity exists) - “update”: Force update (raises DocumentNotFound if identity doesn’t exist) - “upsert”: Insert or update(instance must have an identity).

Returns:

The saved document instance (with generated identity if new).

Raises:

• DuplicateKeyError: In insert mode when document with same identity exists.

• DocumentNotFound: In update mode when document doesn’t exist.

async classmethod get(id_: ID_T, /) → T

Get document by its identity value.

Parameters:

id – Document identity value.

Returns:

Found document.

Raises:

• DocumentNotFound: If document doesn’t exist.

async classmethod find_one(query: Query, /) → T

Find a single document matching the query.

Parameters:

query – Query to match documents against.

Returns:

First matching document.

Raises:

• DocumentNotFound: If no document matches the query.

async classmethod find_one_or_none(query: Query, /) → T | None

Find a single document matching the query or return None if not found.

Parameters:

query – Query to match documents against.

Returns:

First matching document or None if none match.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

async classmethod find(query: Query | None = None, /, *, sort: Query | None = None, skip: int | None = None, limit: int | None = None) → list[T]

Find documents matching the query.

Parameters:

• query – Optional query to filter documents.

• sort – Optional sorting criteria.

• skip – Optional number of documents to skip.

• limit – Optional maximum number of documents to return.

Returns:

List of matching documents.

classmethod find_iter(query: Query | None = None, /, *, sort: Query | None = None, skip: int | None = None, limit: int | None = None) → AsyncIterable[T]

Find documents matching the query.

Parameters:

• query – Optional query to filter documents.

• sort – Optional sorting criteria.

• skip – Optional number of documents to skip.

• limit – Optional maximum number of documents to return.

Returns:

Async iterable of matching documents.

async classmethod count_documents(query: Query | None = None, /) → int

Count documents matching the query.

Parameters:

query – Optional query to filter documents.

Returns:

Number of matching documents.

async classmethod find_and_count(query: Query | None = None, /, *, sort: Query | None = None, skip: int | None = None, limit: int | None = None) → tuple[list[T], int]

Find documents and get total count in one operation.

Parameters:

• query – Optional query to filter documents.

• sort – Optional sorting criteria.

• skip – Optional number of documents to skip.

• limit – Optional maximum number of documents to return.

Returns:

Tuple of (list of matching documents, total count).

async classmethod update_document(id_: ID_T, update: Query, /, *, upsert: bool = False) → T

Update document by identity value.

Parameters:

• id – Document identity value to update.

• update – Update operations i.g. Inc, Set etc.

• upsert – Create document if doesn’t exist.

Returns:

Updated/upserted document.

Raises:

• DocumentNotFound: If document doesn’t exist and upsert=False.

async delete() → T

Delete the current document fromDB.

Returns:

The deleted document instance with None identity.

Raises:

• DocumentNotFound: If document doesn’t exist in database.

async before_delete() → T

Hook called before deleting a document.

Returns:

The document instance.

Engine

class butty.Engine(db: ~motor.core.AgnosticDatabase, *, collection_name_format: ~typing.Callable[[~typing.Type[~butty.document.Document[Any]]], str] = <function Engine.<lambda>>, link_name_format: ~typing.Callable[[~butty.compat.ModelFieldInfo], str] = <function Engine.<lambda>>)

__init__(db: ~motor.core.AgnosticDatabase, *, collection_name_format: ~typing.Callable[[~typing.Type[~butty.document.Document[Any]]], str] = <function Engine.<lambda>>, link_name_format: ~typing.Callable[[~butty.compat.ModelFieldInfo], str] = <function Engine.<lambda>>)

Initialize the MongoDB engine with database connection and naming formats.

Parameters:

• db – MongoDB database connection.

• collection_name_format – Function to generate collection names from models.

• link_name_format – Function to generate field names for links/relations.

bind(*documents: Type[Document[Any]]) → Self

Bind document models to this engine instance.

Parameters:

documents – Document models to bind. If empty, uses all registered documents.

Returns:

The engine instance for chaining.

unbind() → Self

Unbind all document models from this engine instance.

Returns:

The engine instance for chaining.

async init() → Self

Initialize the engine by creating database indexes.

Returns:

The engine instance for chaining.

Fields

class butty.fields.KnownExtra(value)

An enumeration.

butty.fields.OnDelete

Defines possible behaviors when linked document is deleted.

Possible values: - nothing: Take no action (default) - cascade: Delete this document when linked document is deleted - propagate: Delete linked documents when this document is deleted

alias of Literal[‘nothing’, ‘propagate’, ‘cascade’]

butty.fields.IndexedField(default: Any = PydanticUndefined, *, unique: bool = False, **kwargs: Any) → Any

Creates an indexed field in MongoDB.

Parameters:

• default – Default field value.

• unique – Whether to create unique index (defaults to False if not specified).

• kwargs – Additional Pydantic Field arguments.

Returns:

Field definition with index metadata.

butty.fields.IdentityField(default: Any = PydanticUndefined, *, identity_provider: IdentityProvider | None = None, identity_provider_factory: IdentityProviderFactory | None = None, **kwargs: Any) → Any

Creates a document identity field.

Parameters:

• default – Default field value.

• identity_provider – Identity provider.

• identity_provider_factory – Factory for identity provider.

• kwargs – Additional Pydantic Field arguments.

Returns:

Field definition with identity metadata.

butty.fields.VersionField(default: Any = PydanticUndefined, *, version_provider: VersionProvider, **kwargs: Any) → Any

Creates a document version field for optimistic concurrency control.

Parameters:

• default – Default field value.

• version_provider – Version provider implementation.

• kwargs – Additional Pydantic Field arguments.

Returns:

Field definition with version metadata.

butty.fields.LinkField(default: Any = PydanticUndefined, *, link_name: str | None = None, link_ignore: bool = False, on_delete: Literal['nothing', 'propagate', 'cascade'] = 'nothing', **kwargs: Any) → Any

Creates a field that links to another document.

Parameters:

• default – Default field value.

• link_name – Custom field name for storing link in MongoDB.

• link_ignore – If True, skip this field during link processing.

• on_delete – Behavior when linked document is deleted.

• kwargs – Additional Pydantic Field arguments.

Returns:

Field definition with link metadata.

butty.fields.BackLinkField(default: Any = PydanticUndefined, **kwargs: Any) → Any

Creates a field that represents a back reference from another document.

Parameters:

• default – Default field value.

• kwargs – Additional Pydantic Field arguments.

Returns:

Field definition with backlink metadata.

Query

butty.query.F(butty_field: Any) → ButtyField

Casts model field to ButtyField to use in queries.

butty.query.Q(query: dict[str, Any] | dict[ButtyField, Any] | ButtyQuery | None) → dict[str, Any]

Builds MongoDB query from ButtyQuery or dict, where keys can be ButtyFields.

butty.query.Set(query: dict[str, Any] | dict[ButtyField, Any] | ButtyQuery) → dict[str, Any] | dict[ButtyField, Any] | ButtyQuery

Create a MongoDB $set update operation.

Parameters:

query – Field/value mappings to set

Returns:

MongoDB update query with $set operator

butty.query.Inc(query: dict[str, Any] | dict[ButtyField, Any] | ButtyQuery) → dict[str, Any] | dict[ButtyField, Any] | ButtyQuery

Create a MongoDB $inc update operation.

Parameters:

query – Field/value mappings to set

Returns:

MongoDB update query with $inc operator

Errors

exception butty.errors.ButtyError

Base exception class for all Butty-specific errors.

exception butty.errors.ButtyValueError

Raised when invalid values are provided to Butty operations.

This covers: - Field values violating model constraints - Invalid method parameters - Type mismatches in operations

exception butty.errors.DocumentNotFound(doc_model: DocModel, op: str, query: Query)