Atomic-like S3 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 Saga-like 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 (CreateRepository, GetRepository, etc.).
• Pydantic-powered models — work with TextContent, JsonContent, ImageContent and others instead of raw bytes.

Supported Connectors

Connector	Description	Documentation
AWS S3	s3 compatible object storage via aiobotocore	📖 Docs
In-Memory (default)	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	Documenration
In-Memory (default)	great for and tests, development and small projects	📖 Docs

Backend Comparison

	Raw aiobotocore	In-Memory
Consistency	🔴 None No link between files and metadata	🟡 Medium Consistent within process lifetime, lost on crash
Compensation	🔴 None Orphaned files on failure	🟡 Medium Automatic within process, orphaned files possible on crash
Reliability	🔴 Low Failures leave storage in unknown state	🟡 Medium Syncs with storage on restart, files uploaded during crash cannot be compensated
Latency	✅ Lowest Pure S3 network overhead only	✅ Lowest Minimal in-process overhead
Infrastructure	✅ None	✅ None
Memory usage	✅ None	🟡 Stores file metadata in-process

Quick Start

Here's a minimal example of FennFlow:

import asyncio

from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import InMemoryBackendConfig
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=InMemoryBackendConfig(),
        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

Read the docs to learn more about working with FennFlow.

Read the API Reference to understand FennFlow’s interface.