Fractal Repositories

Fractal Repositories is an implementation of the repository pattern of Domain Driven Design (DDD) for building SOLID logic for your Python applications.

Installation

pip install fractal-repositories

Development

Setup the development environment by running:

make deps
pre-commit install

Happy coding.

Occasionally you can run:

make lint

This is not explicitly necessary because the git hook does the same thing.

Do not disable the git hooks upon commit!

Usage

To get started, define an Entity you would like to store:

from dataclasses import dataclass

from fractal_repositories.core.entity import Entity


@dataclass
class DemoModel(Entity):
    id: str
    name: str

Next, define the generic Repository:

from abc import ABC

from fractal_repositories.core.repositories import Repository


class DemoRepository(Repository[DemoModel], ABC):
    entity = DemoModel

Now choose your target storage system, e.g., Google Firestore and define the adapter:

from fractal_repositories.contrib.gcp.firestore.mixins import FirestoreRepositoryMixin


class FirestoreDemoRepository(DemoRepository, FirestoreRepositoryMixin[DemoModel]):
    ...

Compare-and-swap

update() is last-write-wins. Two callers that read the same row, change it and write it back both succeed, and the second silently erases the first. Usually that is tolerable. When the field being written is the thing being competed over -- rotating a refresh token, claiming a job, incrementing a counter -- it is not: the loser hands its holder a value the store no longer has, and nothing reports an error.

compare_and_swap() makes the check and the write one indivisible step:

session = repo.find_one(Specification.parse(id=session_id))
seen = session.refresh_token_hash

session.refresh_token_hash = new_hash
if repo.compare_and_swap(
    session, expected=Specification.parse(refresh_token_hash=seen)
):
    ...  # the row was still as we read it, and is now ours
else:
    ...  # someone else got there first -- re-read and decide, do not overwrite

expected describes what must still hold for the stored row, normally the value this caller read. The return value is the outcome, not an error: False means the row had already moved on (or is gone) and was left exactly as the other writer left it.

Backend support

A backend that cannot make the check and the write indivisible raises NotImplementedError rather than quietly degrading to an update() -- a compare-and-swap that silently is not one is worse than none at all, because the caller stops watching for the race it believes it ruled out. Test supports_compare_and_swap when the same code runs against several backends.

The wrappers (MemoizedRepository, CachedRepository, FieldPermissionsRepository) pass it through and report the inner repository's support. A memoized wrapper evicts its cache entry when a swap loses, since what it holds describes a row someone else has since rewritten.

Utilities

FieldPermissionsRepository

FieldPermissionsRepository wraps any repository to enforce field-level read masking and write validation based on the caller's roles. It is useful when different users should see or modify only certain fields of the same entity — for example, a superadmin can set a sensitive field that regular users can never read or overwrite.

Defining field permissions on an entity

Add a field_permissions() classmethod to your entity. Each entry maps a field name to its read_roles and/or write_roles:

from dataclasses import dataclass
from fractal_repositories.core.entity import Entity


@dataclass
class Order(Entity):
    id: str
    amount: float
    internal_notes: str = ""  # only visible/writable by staff

    @classmethod
    def field_permissions(cls):
        return {
            "internal_notes": {
                "read_roles": ["staff"],
                "write_roles": ["staff"],
            }
        }

Entities without a field_permissions() method are handled transparently — no masking or validation is applied.

Wrapping a repository

from fractal_repositories.utils.field_permissions_repository import (
    FieldPermissionsRepository,
    OnWriteConflict,
)

inner_repo = OrderRepository()  # any Repository implementation
repo = FieldPermissionsRepository(inner_repo)

Read masking

Pass roles to find_one() or find(). Fields the caller is not allowed to read are set to None. Pass roles=None to skip masking (e.g. for internal service calls):

# Staff sees everything
order = repo.find_one(Specification.parse(id="1"), roles=["staff"])
# order.internal_notes == "confidential"

# Regular user gets the field blanked
order = repo.find_one(Specification.parse(id="1"), roles=["customer"])
# order.internal_notes is None

# Internal call — no masking
order = repo.find_one(Specification.parse(id="1"))
# order.internal_notes == "confidential"

Write validation

Pass roles to add() or update(). By default (OnWriteConflict.RAISE), a PermissionError is raised if the caller submits a non-default value for a field they are not allowed to write:

# Allowed — internal_notes is at its default value
repo.add(Order(id="1", amount=99.0), roles=["customer"])

# Raises PermissionError — customer cannot set internal_notes
repo.add(Order(id="1", amount=99.0, internal_notes="hack"), roles=["customer"])

on_write_conflict="preserve" — PUT-style APIs

When callers send back the full entity (e.g. a REST PUT), they will have received secured fields set to None. Submitting None against a staff-set value would raise under the default mode. Use OnWriteConflict.PRESERVE to silently restore the stored value instead, so the rest of the update proceeds normally:

repo = FieldPermissionsRepository(inner_repo, on_write_conflict=OnWriteConflict.PRESERVE)

# Staff sets a note
repo.update(Order(id="1", amount=99.0, internal_notes="VIP customer"), roles=["staff"])

# Customer updates the amount — internal_notes is silently preserved, not raised
repo.update(Order(id="1", amount=120.0, internal_notes=None), roles=["customer"])
# stored: internal_notes == "VIP customer"

Custom exception class

Replace PermissionError with your own exception by passing forbidden_exception_class:

from myapp.exceptions import ForbiddenError

repo = FieldPermissionsRepository(inner_repo, forbidden_exception_class=ForbiddenError)

MemoizedRepository

MemoizedRepository wraps any repository and caches find_one() results in-process, so repeated lookups for the same specification never hit the underlying store more than once. Mutations are cache-aware: update refreshes the cached entry in place, remove_one and delete evict only the affected entity, and add leaves the cache untouched (the new entity is populated lazily on the next find_one).

find(), count(), and is_healthy() always delegate directly to the inner repository without caching.

Constructor

from fractal_repositories.utils.memoized_repository import MemoizedRepository

inner_repo = ItemRepository()  # any Repository implementation
repo = MemoizedRepository(inner_repo)

Caching behaviour

spec = Specification.parse(id="42")

item = repo.find_one(spec)   # hits the inner repository
item = repo.find_one(spec)   # served from in-process cache — inner not called again

item = repo.get("42")        # get() calls find_one() internally, so also cached

Cache invalidation

Each write operation is targeted — only the affected entity's cache entry changes:

repo.add(Item(id="1", name="Alice"))          # no cache effect; populated lazily on next find_one
repo.update(Item(id="1", name="Bob"))         # updates _cache["1"] in place; other entries untouched
repo.remove_one(Specification.parse(id="1"))  # evicts _cache["1"] only
repo.delete("1")                              # evicts _cache["1"] only (calls remove_one)

Each MemoizedRepository instance maintains its own independent cache, so wrapping the same inner repo in two separate MemoizedRepository objects gives two separate caches.

CachedRepository

CachedRepository implements a write-through dual-layer cache. It takes a main_repository (e.g. Mongo or Postgres) and a cache_repository (e.g. InMemory) and keeps them in sync automatically. Reads are always served from the fast cache; writes go to both layers simultaneously.

Constructor

from fractal_repositories.utils.cached_repository import CachedRepository

repo = CachedRepository(
    main_repository=mongo_repo,   # any Repository implementation
    cache_repository=memory_repo, # any Repository implementation
)

Write-through behaviour

Every mutation is applied to both the main repository and the cache:

repo.add(item)              # written to main_repository and cache_repository
repo.update(item)           # updated in main_repository and cache_repository
repo.remove_one(spec)       # removed from main_repository and cache_repository

Read-from-cache behaviour

find_one() and find() are served exclusively from the cache, so they never touch the main repository after the cache has been populated:

item  = repo.find_one(spec)   # reads from cache_repository only
items = list(repo.find())     # reads from cache_repository only
n     = repo.count()          # reads from main_repository (authoritative count)

Populating the cache

The cache starts empty. Call one of the two sync helpers before your application begins serving reads:

• load_cache() — one-way populate. Copies every entity from the main repository into the cache. Use this on a fresh, empty cache (calling it on a non-empty cache will attempt to add entities that already exist).
• reload_cache() — full sync. Upserts every entity from the main repository into the cache, then removes any cache entries whose ids are no longer present in the main repository. Safe to call repeatedly for periodic refresh.

# Startup: populate the cache once
repo.load_cache()

# Periodic refresh: keep the cache in sync with the source of truth
repo.reload_cache()

Example

from fractal_repositories.utils.cached_repository import CachedRepository

# inner_main_repo  — e.g. a PostgresProductRepository
# inner_cache_repo — e.g. an InMemoryProductRepository

repo = CachedRepository(
    main_repository=inner_main_repo,
    cache_repository=inner_cache_repo,
)

# Populate the cache on application startup
repo.load_cache()

# From here on, find_one/find are served from memory
product = repo.find_one(Specification.parse(id="sku-42"))

MigratingRepository

MigratingRepository switches a repository to a new store without migrating it first. It reads from and writes to primary, the new store. An entity that primary does not have yet is looked up in source, the old store, copied to primary under the same id, and returned. source is only ever read.

from fractal_repositories.utils.migrating_repository import MigratingRepository

users = MigratingRepository(
    primary=mongo_users,       # the new store
    source=firestore_users,    # the old one, still written to by other apps
    refresh_after=15 * 60,     # re-read a copied entity from the old store at most every 15 min
)

users.find_one(Specification.parse(email="alice@example.com"))  # copied on first use
users.backfill()                                                # copy the rest, when it suits

• refresh_after keeps copies current while the old store is still where changes are made. An entity is re-read from source once that many seconds have passed since its last check in this process, and updated in primary when it changed. An entity source does not have is left alone, so what is created in primary is its own from the start. Leave it at None to copy each entity once.
• same(a, b) decides whether a copy changed (default: equal asdict()). Pass one that ignores bookkeeping fields set when an entity is built, or every refresh writes.
• Only a narrowed, first-page find falls back to source; an unfiltered one never copies the whole store as a side effect of listing it.
• source failing is logged and never fails a read of something primary already has.
• count() counts primary.

Wrapped in MemoizedRepository, a refresh is only seen when the memoized entry is read again from the inner repository; turn memoization off for entities that change in the old store.

DistributedReadRepository

DistributedReadRepository is a read-only data-federation layer. It takes a main_repository and a list of DistributedRelation objects. For every entity returned by the main repository, it queries each related repository and merges the results into the entity before yielding it. There are no add, update, or remove_one methods — this class is intentionally read-only.

DistributedRelation

from fractal_repositories.utils.distributed_read_repository import DistributedRelation

relation = DistributedRelation(
    repository=other_repo,              # ReadRepository to query for each main entity
    specification_func=lambda e: ...,   # callable: main entity → Specification
    aggregate_target_field="",          # "" = flat merge (default), or a field name for aggregation
)

Constructor

from fractal_repositories.utils.distributed_read_repository import DistributedReadRepository

repo = DistributedReadRepository(
    main_repository=main_repo,
    other_repositories=[relation1, relation2],
)

count() returns the sum of counts across all repositories. is_healthy() checks all repositories.

Flat-merge example (field enrichment)

When aggregate_target_field is "" (the default), non-empty, non-id fields from the related entity are merged field-by-field into the main entity. This is useful when a second repository holds supplementary data for the same entity type.

from fractal_specifications.generic.specification import Specification
from fractal_repositories.utils.distributed_read_repository import (
    DistributedReadRepository,
    DistributedRelation,
)

# product_repo  — main store: id, name, price
# metadata_repo — secondary store: id, description, tags

repo = DistributedReadRepository(
    main_repository=product_repo,
    other_repositories=[
        DistributedRelation(
            repository=metadata_repo,
            specification_func=lambda p: Specification.parse(id=p.id),
            # aggregate_target_field="" — fields are merged directly onto the product
        )
    ],
)

# Each yielded Product has name, price AND description, tags populated
product = repo.find_one(Specification.parse(id="sku-42"))

Aggregation example (one-to-many)

When aggregate_target_field is set to a field name, every matching entity from the related repository is appended to that list field on the main entity. The field must already exist on the entity (typically defined as an empty list default).

from fractal_repositories.utils.distributed_read_repository import (
    DistributedReadRepository,
    DistributedRelation,
)

# order_repo    — main store: Order(id, customer_id, total, line_items=[])
# line_item_repo — secondary store: LineItem(id, order_id, sku, qty)

repo = DistributedReadRepository(
    main_repository=order_repo,
    other_repositories=[
        DistributedRelation(
            repository=line_item_repo,
            specification_func=lambda order: Specification.parse(order_id=order.id),
            aggregate_target_field="line_items",  # collect into Order.line_items
        )
    ],
)

# Each yielded Order has its line_items list populated
order = repo.find_one(Specification.parse(id="order-99"))
# order.line_items == [LineItem(...), LineItem(...), ...]

Contrib

Fractal Repositories comes with ready to use adapter mixins for:

• Django
• SQLAlchemy
• PostgreSQL
• MongoDB
• DuckDB
• Google Firestore
• Google Cloud Storage

Examples

DuckDB (In-process Analytical Database)

DuckDB is an in-process SQL OLAP database management system, perfect for analytical workloads and embedded analytics.

from dataclasses import dataclass
from fractal_repositories.core.entity import Entity
from fractal_repositories.core.repositories import Repository
from fractal_repositories.contrib.duckdb import DuckDBRepositoryMixin
from fractal_specifications.generic.specification import Specification


@dataclass
class User(Entity):
    id: str
    name: str
    email: str


class UserRepository(Repository[User], DuckDBRepositoryMixin[User]):
    entity = User


# In-memory database (great for testing)
repo = UserRepository(database=":memory:", table="users")

# Or file-based database for persistence
# repo = UserRepository(database="analytics.db", table="users")

# Add entities
user = User(id="1", name="Alice", email="alice@example.com")
repo.add(user)

# Query with specifications
found_user = repo.find_one(Specification.parse(id="1"))

# Find all users
all_users = list(repo.find())

# Count users
user_count = repo.count()

# Update
user.email = "newemail@example.com"
repo.update(user)

# Remove
repo.remove_one(Specification.parse(id="1"))

MongoDB

from fractal_repositories.contrib.mongo import MongoRepositoryMixin


class MongoUserRepository(UserRepository, MongoRepositoryMixin[User]):
    pass


repo = MongoUserRepository(
    host="localhost",
    port="27017",
    username="user",
    password="pass",
    database="mydb",
    collection="users"
)

Native datetimes

By default, datetimes are stored as the ISO strings Entity.asdict() produces. Set native_datetimes to store them as BSON dates instead: they sort and compare correctly across UTC offsets, and work with date indexes, TTL indexes and date aggregations.

class MongoUserRepository(UserRepository, MongoRepositoryMixin[User]):
    native_datetimes = True  # or pass native_datetimes=True to the constructor

• Datetimes are stored in UTC and truncated to the millisecond, the precision of a BSON date. Filters are truncated the same way, so an equality filter on a microsecond value still matches what was stored.
• Reads return timezone-aware UTC datetimes, even from a client created without tz_aware=True.
• date fields stay YYYY-MM-DD strings, which already sort correctly.

Migrating existing data. A collection written without native_datetimes holds strings that datetime filters no longer see. With auto_migrate (on by default) the repository converts them the first time it is used: every datetime field of the entity, including fields of nested models, is found through its type hints and converted in batches. Completion is recorded in the fractal_migrations collection, so later starts only look that up.

The migration is idempotent and safe to run from several processes at once: each write only applies while the field still holds the string it read. Strings that do not parse as a datetime, and datetimes inside lists, are left as they are and logged.

To migrate from a deploy step instead, for instance for a collection large enough that its first scan should not delay a request, turn auto_migrate off and run it yourself:

from fractal_repositories.contrib.mongo.migrations import migrate_datetimes

repo = MongoUserRepository(..., native_datetimes=True, auto_migrate=False)
result = migrate_datetimes(repo)                # MigrationResult(converted=..., unparseable=..., ...)
migrate_datetimes(repo, reverse=True)           # back to strings, e.g. to downgrade

Anything that reads the raw documents outside this library (exports, BI tools, other services) sees dates instead of strings once a collection is migrated.

Google Firestore

from fractal_repositories.contrib.gcp.firestore import FirestoreRepositoryMixin


class FirestoreUserRepository(UserRepository, FirestoreRepositoryMixin[User]):
    pass


repo = FirestoreUserRepository(
    collection="users",
    service_account_path="/path/to/service-account.json"
)

PostgreSQL

from fractal_repositories.contrib.postgresql import PostgresRepositoryMixin


class PostgresUserRepository(UserRepository, PostgresRepositoryMixin[User]):
    pass


repo = PostgresUserRepository(
    postgres_host="localhost",
    postgres_port="5432",
    postgres_db="mydb",
    postgres_user="user",
    postgres_password="pass",
    table="users"
)

Connections come from a pool shared by every repository with the same connection settings in the process. They are opened on demand, kept for reuse, and always returned, also when a call raises. find reads its rows before yielding them, so an iterator that is not read to the end holds no connection. Optional settings:

PostgresRepositoryMixin.close_all_pools() closes them, for a shutdown hook.

Django

from fractal_repositories.contrib.django import DjangoModelRepositoryMixin


class DjangoUserRepository(UserRepository, DjangoModelRepositoryMixin[User]):
    pass


# Assuming you have a Django model
from myapp.models import UserModel

repo = DjangoUserRepository(django_model=UserModel)

SQLAlchemy

from fractal_repositories.contrib.sqlalchemy.mixins import SqlAlchemyRepositoryMixin


class SqlAlchemyUserRepository(UserRepository, SqlAlchemyRepositoryMixin[User]):
    pass


# Use with your SQLAlchemy session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///example.db")
Session = sessionmaker(bind=engine)
session = Session()

repo = SqlAlchemyUserRepository(session=session, model=YourSqlAlchemyModel)

File (JSON Lines)

FileRepositoryMixin persists each entity as one JSON object per line in <root_dir>/db/<RepositoryName>.jsonl. It is built to survive corrupt data and torn writes:

• Resilient reads — the reader skips empty lines, null bytes, and JSON that does not deserialize into an entity (logging a warning for each) instead of crashing. Genuine deserialization bugs propagate loudly rather than silently dropping rows.
• Crash-safe appends — add appends a single line, so a process killed mid-write produces at most one malformed trailing line, which the reader skips.
• Atomic rewrites — update and remove_one rewrite the file via a temp file plus os.replace, so a crash leaves either the old complete file or the new one, never a half-rewritten (truncated) table.

This is designed for a single writer process — there is no file locking, so concurrent writers can clobber each other on the rewrite path.

from fractal_repositories.mixins.file_repository_mixin import FileRepositoryMixin


class FileUserRepository(UserRepository, FileRepositoryMixin[User]):
    pass


repo = FileUserRepository(root_dir="/var/lib/myapp")

SQLite (self-contained, no SQLAlchemy)

SqliteRepositoryMixin is a drop-in replacement for FileRepositoryMixin that stores the data in a SQLite file at <root_dir>/db/<RepositoryName>.sqlite instead of a JSON Lines file. It uses only the standard-library sqlite3 module — no SQLAlchemy and no third-party dependencies.

It is fully transparent: swap the base class and change nothing else. The constructor signature (root_dir=...) and all query results are identical to the file mixin.

• No schema to manage — each entity is stored as a JSON document in a single generic table whose shape never changes: CREATE TABLE "<RepositoryName>" (id TEXT PRIMARY KEY, data TEXT NOT NULL). All entity structure lives inside the data column, so there is nothing to migrate when the entity changes.
• Auto-adapting to entity changes — old rows upgrade transparently on read. Removed/renamed fields in stored rows are ignored (Entity.from_dict keeps only known fields). A row that can no longer be built into the current entity (for example after a new required field with no default is added) is skipped with a warning rather than crashing the read — give new fields a default and old rows keep working.
• SQLite is the sole source of truth — every read hits the database; nothing is cached in memory between calls. The id-keyed paths (get / find_one / delete by id) use the indexed primary key, and count() with no specification is answered by SELECT COUNT(*).
• Specifications are pushed down into SQL — SqliteSpecificationBuilder translates a specification into a WHERE clause over the JSON data column (json_extract(data, '$.field') = ?), so non-matching rows are filtered in SQLite's C layer and never deserialized. Pushdown is applied only when it is exactly equivalent to evaluating the specification in Python — i.e. for JSON-native scalar values (str / int / float / bool / None). Specifications whose value is a type that asdict stores as a string (Decimal, date, datetime, UUID, Enum) or that use operators SQLite cannot faithfully express (Contains, regex, Not) fall back to Python evaluation. Pushdown only ever changes performance, never the result — the output is always identical to the other repositories. (One deliberate refinement: a range comparison against a null field excludes that row in SQL, where the in-memory path would raise TypeError.) Ordering and pagination are applied in Python over the matched rows, because Decimal (a fixed-point string) and other serialized types do not sort naturally in SQL.
• Crash-safe and concurrent — SQLite provides ACID atomicity, crash-safety, and concurrent-writer locking, so unlike the file mixin there is no single-writer caveat and no torn-write handling to worry about.

from fractal_repositories.mixins.sqlite_repository_mixin import SqliteRepositoryMixin


class SqliteUserRepository(UserRepository, SqliteRepositoryMixin[User]):
    pass


repo = SqliteUserRepository(root_dir="/var/lib/myapp")