Main Usage#

This is the most general use-case for Alaric, where-in you use the Document class.

class alaric.Document(database: AsyncIOMotorDatabase, document_name: str, converter: Type[T] | None = None)#
__init__(database: AsyncIOMotorDatabase, document_name: str, converter: Type[T] | None = None)#
Parameters:

  • database (AsyncIOMotorDatabase) – The database we are connected to

  • document_name (str) – What this _document should be called

  • converter (Optional[Type[T]]) – An optional class to try to convert all data-types which return either Dict or List into

1from motor.motor_asyncio import AsyncIOMotorClient
2
3client = AsyncIOMotorClient(connection_url)
4database = client["my_database"]
5config_document = Document(database, "config")
async bulk_insert(data: List[Dict]) → None#

Given a List of Dictionaries, bulk insert all the given dictionaries in a single call.

Parameters:

data (List[Dict]) – The data to bulk insert

1# Insert 25 documents
2await Document.bulk_insert(
3    {"_id": i}
4    for i in range(25)
5)
async change_field_to(filter_dict: Dict[str, Any] | Buildable | Filterable, field: str, new_value: Any) → None#

Modify a single field and change the value.

Parameters:

  • filter_dict (Union[Dict[Any, Any], Buildable, Filterable]) – A dictionary to use as a filter or AQ object.

  • field (str) – The key for the field to increment

  • new_value (Any) – What the field should get changed to

1# Assuming a data structure of
2# {"_id": 1, "prefix": "!"}
3await Document.change_field_to({"_id": 1}, "prefix", "?")
4
5# This will now look like
6# {"_id": 1, "prefix": "?"}
property collection_name: str#

The connected collections name.

async count(filter_dict: Dict[Any, Any] | Buildable | Filterable) → int#

Return a count of how many items match the filter.

Parameters:

filter_dict (Union[Dict[Any, Any], Buildable, Filterable]) – The count filer.

Returns:

How many items matched the filter.

Return type:

int

1# How many items have the `enabled` field set to True
2count: int = await Document.count({"enabled": True})
create_cursor() → Cursor#
async delete(filter_dict: Dict | Buildable | Filterable) → DeleteResult | None#

Delete an item from the Document if an item with the provided filter exists.

Parameters:

filter_dict (Union[Dict, Buildable, Filterable]) – A dictionary to use as a filter or AQ object.

Returns:

The result of deletion if it occurred.

Return type:

Optional[DeleteResult]

1# Delete items with a `prefix` of `!`
2await Document.delete({"prefix": "!"})
async delete_all() → None#

Delete all data associated with this document.

Notes

This will attempt to complete the operation in a single call, however, if that fails it will fall back to manually deleting items one by one.

Warning

There is no going back if you call this accidentally.

This also currently doesn’t appear to work.

property document_name: str#

Same as collection_name()

async find(filter_dict: Dict[str, Any] | Buildable | Filterable, projections: Dict[str, Any] | Projection | None = None, *, try_convert: bool = True) → Dict[str, Any] | Type[T] | None#

Find and return one item.

Parameters:

  • filter_dict (Union[Dict, Buildable, Filterable]) – A dictionary to use as a filter or AQ object.

  • projections (Optional[Union[Dict[str, Any], Projection]]) – Specify the data you want returned from matching queries.

  • try_convert (bool) –

    Whether to attempt to run convertors on returned data.

    Defaults to True

Returns:

The result of the query

Return type:

Optional[Union[Dict[str, Any], Type[T]]]

1# Find the document where the `_id` field is equal to `my_id`
2data: dict = await Document.find({"_id": "my_id"})
async find_many(filter_dict: Dict[str, Any] | Buildable | Filterable, projections: Dict[str, Any] | Projection | None = None, *, try_convert: bool = True) → List[Dict[str, Any] | Type[T]]#

Find and return all items matching the given filter.

Parameters:

  • filter_dict (Union[Dict[str, Any], Buildable, Filterable]) – A dictionary to use as a filter or AQ object.

  • projections (Optional[Union[Dict[str, Any], Projection]]) – Specify the data you want returned from matching queries.

  • try_convert (bool) –

    Whether to attempt to run convertors on returned data.

    Defaults to True

Returns:

The result of the query

Return type:

List[Union[Dict[str, Any], Type[T]]]

Notes

This uses a cursor internally, consider using them for more complicated queries.

1# Find all documents where the key `my_field` is `true`
2data: list[dict] = await Document.find_many({"my_field": True})
async get_all(filter_dict: Dict[str, Any] | Buildable | Filterable | None = None, projections: Dict[str, Any] | Projection | None = None, *args: Any, try_convert: bool = True, **kwargs: Any) → List[Dict[str, Any] | Type[T] | None]#

Fetches and returns all items which match the given filter.

Parameters:

  • filter_dict (Optional[Union[Dict[str, Any], Buildable, Filterable]]) – A dictionary to use as a filter or AQ object.

  • projections (Optional[Union[Dict[str, Any], Projection]]) – Specify the data you want returned from matching queries.

  • try_convert (bool) –

    Whether to attempt to run convertors on returned data.

    Defaults to True

Returns:

The items matching the filter

Return type:

List[Optional[Union[Dict[str, Any], Type[T]]]]

1data: list[dict] = await Document.get_all()
async increment(filter_dict: Dict[str, Any] | Buildable | Filterable, field: str, amount: int | float) → None#

Increment the provided field.

Parameters:

  • filter_dict (Union[Dict[str, Any], Buildable, Filterable]) – The ‘thing’ we want to increment

  • field (str) – The key for the field to increment

  • amount (Union[int, float]) – How much to increment (or decrement) by

1# Assuming a data structure of
2# {"_id": 1, "counter": 4}
3await Document.increment({"_id": 1}, "counter", 1)
4
5# Now looks like
6# {"_id": 1, "counter": 5}

Notes

You can also use negative numbers to decrease the count of a field.

async insert(data: Dict[str, Any] | Saveable) → None#

Insert the provided data into the document.

Parameters:

data (Union[Dict[str, Any], Saveable]) – The data to insert

1# If you don't provide an _id,
2# Mongo will generate one for you automatically
3await Document.insert({"_id": 1, "data": "hello world"})
property raw_collection: AsyncIOMotorCollection#

The connection collection instance.

property raw_database: AsyncIOMotorDatabase#

Access to the database instance.

async unset(filter_dict: Dict[str, Any] | Buildable | Filterable, field: Any) → None#

Delete a given field on a collection

Parameters:

  • filter_dict (Union[Dict[str, Any], Buildable, Filterable]) – The fields to match on (Think _id)

  • field (Any) – The field to remove

1# Assuming we have a document that looks like
2# {"_id": 1, "field_one": True, "field_two": False}
3await Document.unset({"_id": 1}, "field_two")
4
5# This data will now look like the following
6# {"_id": 1, "field_one": True}
async update(filter_dict: Dict[str, Any] | Buildable | Filterable, update_data: Dict[str, Any] | Saveable, option: str = 'set', *args: Any, **kwargs: Any) → None#

Performs an UPDATE operation.

Parameters:
1# Update the document with an `_id` of 1
2# So that it now equals the second argument
3await Document.update({"_id": 1}, {"_id": 1, "data": "new data"})
async upsert(filter_dict: Dict[str, Any] | Buildable | Filterable, update_data: Dict[str, Any] | Saveable, option: str = 'set', *args: Any, **kwargs: Any) → None#

Performs an UPSERT operation.

Parameters:
1# Update the document with an `_id` of `1`
2# So that it now equals the second argument
3# NOTE: If a document with an `_id` of `1`
4# does not exist, then this method will
5# insert the data instead.
6await Document.upsert({"_id": 1}, {"_id": 1, "data": "new data"})