Skip to content

Scoped Repository Permissions

FennFlow's mixin architecture lets each repository expose only the operations it should allow. Instead of a single god-repository with full access everywhere, you compose exactly the capabilities each part of your system needs — nothing more.

Example: Main storage with full access, backup storage with safe write-only access

import asyncio

from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import SqlalchemyBackendConfig
from fennflow.connectors import S3ConnectorConfig
from fennflow.files import BinaryContent
from fennflow.repositories import (
    CreateRepository,
    DeleteRepository,
    GetRepository,
    PutRepository,
    S3RepoField,
)


class CRUDRepository(
    CreateRepository,
    DeleteRepository,
    GetRepository,
    PutRepository,
):
    pass


class SafeWriteReadRepository(
    CreateRepository,
    GetRepository,
):
    pass


class AppUOW(UnitOfWork):
    files = S3RepoField(CRUDRepository, bucket_name="main-bucket")
    backups = S3RepoField(SafeWriteReadRepository, bucket_name="backup-bucket")
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(),
        connector=S3ConnectorConfig(
            endpoint_url="https://s3.amazonaws.com",
            aws_access_key_id="aws-key",
            aws_secret_access_key="aws-secret",
        ),
    )


async def main():
    report_file = BinaryContent.from_local_path("report.txt")

    async with AppUOW() as uow:
        # Full access on main storage
        await uow.files.at("reports/").put(report_file)
        await uow.files.delete("reports/old_report.pdf")

        # Safe write on backup — create raises if file already exists,
        # delete and overwrite are simply not available
        await uow.backups.at("reports/").create(report_file)


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

SafeWriteReadRepository has no put and no delete — not restricted at runtime, just never composed in. Misuse is a AttributeError at development time, not a bug in production.