Firedantic

Database models for Firestore using Pydantic base models.

Installation

The package is available on PyPI:

pip install firedantic

Usage

In your application you will need to configure the firestore db client and optionally the collection prefix, which by default is empty.

from os import environ
from unittest.mock import Mock

import google.auth.credentials
from firedantic import configure
from google.cloud.firestore import Client

# Firestore emulator must be running if using locally.
if environ.get("FIRESTORE_EMULATOR_HOST"):
    client = Client(
        project="firedantic-test",
        credentials=Mock(spec=google.auth.credentials.Credentials)
    )
else:
    client = Client()

configure(client, prefix="firedantic-test-")

Once that is done, you can start defining your Pydantic models, e.g:

from pydantic import BaseModel

from firedantic import Model

class Owner(BaseModel):
    """Dummy owner Pydantic model."""
    first_name: str
    last_name: str


class Company(Model):
    """Dummy company Firedantic model."""
    __collection__ = "companies"
    company_id: str
    owner: Owner

# Now you can use the model to save it to Firestore
owner = Owner(first_name="John", last_name="Doe")
company = Company(company_id="1234567-8", owner=owner)
company.save()

# Prints out the firestore ID of the Company model
print(company.id)

# Reloads model data from the database
company.reload()

Querying is done via a MongoDB-like find():

from firedantic import Model
import firedantic.operators as op
from google.cloud.firestore import Query

class Product(Model):
    __collection__ = "products"
    product_id: str
    stock: int
    unit_value: int


Product.find({"product_id": "abc-123"})
Product.find({"stock": {">=": 3}})
# or
Product.find({"stock": {op.GTE: 3}})
Product.find({"stock": {">=": 1}}, order_by=[('unit_value', Query.ASCENDING)], limit=25, offset=50)
Product.find(order_by=[('unit_value', Query.ASCENDING), ('stock', Query.DESCENDING)], limit=2)

The query operators are found at https://firebase.google.com/docs/firestore/query-data/queries#query_operators.

Async usage

Firedantic can also be used in an async way, like this:

import asyncio
from os import environ
from unittest.mock import Mock

import google.auth.credentials
from google.cloud.firestore import AsyncClient

from firedantic import AsyncModel, configure

# Firestore emulator must be running if using locally.
if environ.get("FIRESTORE_EMULATOR_HOST"):
    client = AsyncClient(
        project="firedantic-test",
        credentials=Mock(spec=google.auth.credentials.Credentials),
    )
else:
    client = AsyncClient()

configure(client, prefix="firedantic-test-")


class Person(AsyncModel):
    __collection__ = "persons"
    name: str


async def main():
    alice = Person(name="Alice")
    await alice.save()
    print(f"Saved Alice as {alice.id}")
    bob = Person(name="Bob")
    await bob.save()
    print(f"Saved Bob as {bob.id}")

    found_alice = await Person.find_one({"name": "Alice"})
    print(f"Found Alice: {found_alice.id}")
    assert alice.id == found_alice.id

    found_bob = await Person.get_by_id(bob.id)
    assert bob.id == found_bob.id
    print(f"Found Bob: {found_bob.id}")

    await alice.delete()
    print("Deleted Alice")
    await bob.delete()
    print("Deleted Bob")


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

Subcollections

Subcollections in Firestore are basically dynamically named collections.

Firedantic supports them via the SubCollection and SubModel classes, by creating dynamic classes with collection name determined based on the "parent" class it is in reference to using the model_for() method.

from typing import Optional, Type

from firedantic import AsyncModel, AsyncSubCollection, AsyncSubModel, ModelNotFoundError


class UserStats(AsyncSubModel):
    id: Optional[str] = None
    purchases: int = 0

    class Collection(AsyncSubCollection):
        # Can use any properties of the "parent" model
        __collection_tpl__ = "users/{id}/stats"


class User(AsyncModel):
    __collection__ = "users"
    name: str


async def get_user_purchases(user_id: str, period: str = "2021") -> int:
    user = await User.get_by_id(user_id)
    stats_model: Type[UserStats] = UserStats.model_for(user)
    try:
        stats = await stats_model.get_by_id(period)
    except ModelNotFoundError:
        stats = stats_model()
    return stats.purchases

Composite Indexes and TTL Policies

Firedantic has support for defining composite indexes and TTL policies as well as creating them.

Composite indexes

Composite indexes of a collection are defined in __composite_indexes__, which is a list of all indexes to be created.

To define an index, you can use collection_index or collection_group_index, depending on the query scope of the index. Each of these takes in an arbitrary amount of tuples, where the first element is the field name and the second is the order (ASCENDING/DESCENDING).

The set_up_composite_indexes and async_set_up_composite_indexes functions are used to create indexes.

For more details, see the example further down.

TTL Policies

The field used for the TTL policy should be a datetime field and the name of the field should be defined in __ttl_field__. The set_up_ttl_policies and async_set_up_ttl_policies functions are used to set up the policies.

Note: The TTL policies can not be set up in the Firestore emulator.

Examples

Below are examples (both sync and async) to show how to use Firedantic to set up composite indexes and TTL policies.

The examples use async_set_up_composite_indexes_and_ttl_policies and set_up_composite_indexes_and_ttl_policies functions to set up both composite indexes and TTL policies. However, you can use separate functions to set up only either one of them.

Composite Index and TTL Policy Example (sync)

from datetime import datetime

from firedantic import (
    collection_index,
    collection_group_index,
    configure,
    get_all_subclasses,
    Model,
    set_up_composite_indexes_and_ttl_policies,
)
from google.cloud.firestore import Client, Query
from google.cloud.firestore_admin_v1 import FirestoreAdminClient


class ExpiringModel(Model):
    __collection__ = "expiringModel"
    __ttl_field__ = "expire"
    __composite_indexes__ = [
        collection_index(("content", Query.ASCENDING), ("expire", Query.DESCENDING)),
        collection_group_index(("content", Query.DESCENDING), ("expire", Query.ASCENDING)),
    ]

    expire: datetime
    content: str


def main():
    configure(Client(), prefix="firedantic-test-")
    set_up_composite_indexes_and_ttl_policies(
        gcloud_project="my-project",
        models=get_all_subclasses(Model),
        client=FirestoreAdminClient(),
    )
    # or use set_up_composite_indexes / set_up_ttl_policies functions separately


if __name__ == "__main__":
    main()

Composite Index and TTL Policy Example (async)

import asyncio
from datetime import datetime

from firedantic import (
    AsyncModel,
    async_set_up_composite_indexes_and_ttl_policies,
    collection_index,
    collection_group_index,
    configure,
    get_all_subclasses,
)
from google.cloud.firestore import AsyncClient, Query
from google.cloud.firestore_admin_v1.services.firestore_admin import (
    FirestoreAdminAsyncClient,
)


class ExpiringModel(AsyncModel):
    __collection__ = "expiringModel"
    __ttl_field__ = "expire"
    __composite_indexes__ = [
        collection_index(("content", Query.ASCENDING), ("expire", Query.DESCENDING)),
        collection_group_index(("content", Query.DESCENDING), ("expire", Query.ASCENDING)),
    ]

    expire: datetime
    content: str


async def main():
    configure(AsyncClient(), prefix="firedantic-test-")
    await async_set_up_composite_indexes_and_ttl_policies(
        gcloud_project="my-project",
        models=get_all_subclasses(AsyncModel),
        client=FirestoreAdminAsyncClient(),
    )
    # or await async_set_up_composite_indexes / async_set_up_ttl_policies separately


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

Transactions

Firedantic has basic support for Firestore Transactions. The following methods can be used in a transaction:

• Model.delete(transaction=transaction)
• Model.find_one(transaction=transaction)
• Model.find(transaction=transaction)
• Model.get_by_doc_id(transaction=transaction)
• Model.get_by_id(transaction=transaction)
• Model.reload(transaction=transaction)
• Model.save(transaction=transaction)
• SubModel.get_by_id(transaction=transaction)

When using transactions, note that read operations must come before write operations.

Transaction examples

In this example (async and sync version of it below), we are updating a City to increment and decrement the population of it, both using an instance method and a standalone function. Please note that the @async_transactional and @transactional decorators always expect the first argument of the wrapped function to be transaction; i.e. you can not