fennflow 0.8.1

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:

• SSOT — the backend is the single source of truth for your file storage. No matter what your file storage contains, the backend ensures a consistent view of what exists.
• Saga compensation flow — if something fails mid-operation, all previous actions are automatically compensated in reverse order, leaving storage in a consistent state.
• 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.
• Testability — swap S3ConnectorConfig for InMemoryConnectorConfig and point the backend at an in-memory SQLite database. Zero infrastructure, zero mocks.

Supported Connectors

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 Comparison

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, MediaType, 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])

    from_path_binary_file = BinaryContent.from_local_path("my_file.txt")
    binary_file = BinaryContent(data=b"some bytes", media_type=MediaType.TEXT_PLAIN)

    async with UOW() as uow:
        await uow.my_files.at("folder1").put(
            text_file,
            json_file,
            from_path_binary_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, TextContent, BaseBinary]


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.