fennflow 0.2.0

Atomic-like Agnostic Object Storage Framework, the Pydantic way

Atomic-like Agnostic Object Storage Framework, the Pydantic way

---

Documentation: 📖 Docs

---

FennFlow is a Python s3 framework designed to help you quickly, confidently, and painlessly manipulate files in your object storage implementing SSOT pattern and Saga compensation flow.

Why use FennFlow?

Working with aiobotocore often feels like handling raw bytes and dicts. FennFlow wraps S3 operations into a high-level Unit of Work pattern, providing:

• Atomic-like multistep operations — if something fails, previous actions are automatically compensated (Saga Pattern).
• Clean Architecture — treat S3 as proper repositories using mixins (PutRepository, GetRepository, etc.).
• Pydantic-powered models — work with TextContent, JsonContent, ImageContent and others instead of raw bytes.

Supported Connectors

Connector	Description	Documentation
AWS S3 (default)	s3 compatible object storage via aiobotocore	📖 Docs
In-Memory	great for and tests and development	📖 Docs

Supported Backends

FennFlow uses backend as a source of truth for your file storage. No matter what your file storage contains, backend ensures your data is consistent.

Backend	Description	Documentation
In-Memory	great for and tests, development	📖 Docs
SQLAlchemy (default)	persistent metadata backend, great for all environments	📖 Docs

Backend Comparison

	Raw aiobotocore	SQLAlchemy (default)
Consistency	🔴 None No link between files and metadata	✅ High Persistent across restarts
Compensation	🔴 None Orphaned files on failure	✅ High Automatic within session
Reliability	🔴 Low Failures leave storage in unknown state	✅ High Consistent state guaranteed across restarts
Latency	✅ Lowest Pure S3 network overhead only	🟡 Low/middle DB overhead
Infrastructure	✅ None	✅ None SQLite by default
Memory usage	✅ None	✅ Minimal Metadata persisted to disk, not held in-process

Quick Start

Here's a minimal example of FennFlow:

import asyncio

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


# 1. Define your repository with mixins
class CrudRepository(
    PutRepository,
    DeleteRepository,
    GetRepository,
    ListRepository,
    ):
    pass


# 2. Set up your Unit of Work
class UOW(UnitOfWork):
    my_files = S3RepoField(CrudRepository, bucket_name="my_files")
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(),
        connector=S3ConnectorConfig(),
        )


async def main():
    text_file = TextContent.from_content("Hello, world!")
    json_file = JsonContent.from_content([1, 2, 3])
    binary_file = BinaryContent(data=b"some bytes", media_type="text/plain")

    async with UOW() as uow:
        await uow.my_files.at("folder1").put(
            text_file,
            json_file,
            binary_file,
            )

        paths = await uow.my_files.at("folder1").list()
        print(paths)  # ListResponse[Filepath, ...]

        files = await uow.my_files.get(*paths)
        print(files)  # MediaResponse[TextContent, JsonContent, BinaryContent]


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

(This example is complete, it can be run “as is”, assuming you’ve installed the fennflow package)

Next Steps

To try FennFlow for yourself, clone it and follow the instructions in the examples.

Read the docs to learn more about working with FennFlow.

Read the API Reference to understand FennFlow’s interface.

Learn how to utilize llms with FennFlow.