fennflow 0.1.0

Atomic-like S3 Framework with Saga Pattern, the Pydantic way

Atomic-like S3 Framework, the Pydantic way

---

Documentation: docs/README.md

---

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 (PutRepository, 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	Quick start
In-Memory (default)	great for and tests and development	Quick start

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	Quick start

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())

Next Steps

Read the docs to learn more about working with FennFlow.