# FennFlow

> Atomic-like Agnostic Object Storage Framework, the Pydantic way

FennFlow is a Python atomic-like agnostic object storage framework 
providing atomic-like file operations
with Saga-pattern compensation, built on top of aiobotocore with
Pydantic-powered content models and a Unit of Work abstraction.

# Introduction

### Atomic-like Agnostic Object Storage Framework, the Pydantic way

______________________________________________________________________

**Documentation**: [📖 Docs](https://alex-fir-it.github.io/FennFlow/)

______________________________________________________________________

### *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

| Connector        | Description                                  | Documentation                                                                        |
| ---------------- | -------------------------------------------- | ------------------------------------------------------------------------------------ |
| AWS S3 (default) | s3 compatible object storage via aiobotocore | [📖 Docs](https://alex-fir-it.github.io/core_concepts/connectors/#s3connector)       |
| In-Memory        | great for and tests and development          | [📖 Docs](https://alex-fir-it.github.io/core_concepts/connectors/#inmemoryconnector) |

## 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                                                                      |
| -------------------- | ------------------------------------------------------- | ---------------------------------------------------------------------------------- |
| SQLAlchemy (default) | persistent metadata backend, great for all environments | [📖 Docs](https://alex-fir-it.github.io/core_concepts/backends/#sqlalchemybackend) |
| In-Memory            | great for and tests, development                        | [📖 Docs](https://alex-fir-it.github.io/core_concepts/backends/#inmemorybackend)   |

### 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:

```python3
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](https://github.com/Alex-FIR-IT/FennFlow) and follow the instructions in the [examples](https://alex-fir-it.github.io/examples/setup/index.md).

Read the [docs](https://alex-fir-it.github.io/core_concepts/uow/index.md) to learn more about FennFlow.

Read the [API Reference](https://alex-fir-it.github.io/api/index.md) to understand FennFlow’s interface.

Learn how to utilize [llms](https://alex-fir-it.github.io/llms/index.md) with FennFlow.

# Installation

FennFlow is available on PyPI as [fennflow](https://pypi.org/project/fennflow/) so installation is as simple as:

```bash
pip install fennflow
```

```bash
uv add fennflow
```

(Requires Python 3.10+)

This installs the `fennflow` package, core dependencies, and libraries required to use FennFlow.

# Getting Help

If you need help getting started with FennFlow or with advanced usage, the following sources may be useful.

## GitHub Issues

The [FennFlow GitHub Issues](https://github.com/Alex-FIR-IT/FennFlow/issues) are a great place to ask questions and give us feedback.

# Overview

FennFlow encapsulates all the logic of the Saga and SSOT (single source of truth) approaches, wrapping all file operations in Pydantic models.

S3 has no concept of transactions. A multi-step operation that fails halfway leaves storage in an unknown state. FennFlow addresses this with two complementary patterns: SSOT ensures the backend always presents a consistent view of what exists, and Saga ensures any partially executed sequence is fully compensated on failure. Together they guarantee that from the end user's perspective, an operation either happened completely or not at all.

## SSOT: backend as a single source of truth

File storage (S3, etc.) is treated as a dumb byte store — it has no concept of sessions, pending operations, or consistency guarantees. FennFlow introduces a **backend** as a metadata layer that sits above the file storage and owns the authoritative view of what exists.

Before any read or write reaches the file storage, the backend is consulted first:

- `GET` operation only proceed to the connector if the backend has a record of the file. If the backend doesn't know about it, the file is considered non-existent — even if it physically lives in S3.
- `PUT` and `DELETE` operations are registered in the backend as `PENDING` before the connector is touched.

This separation means the backend can always answer "what exists right now?" consistently, regardless of what the file storage contains at any given moment — including files left behind by crashed sessions or external processes.

## Saga: how operations reach consistency

FennFlow does not use distributed transactions or the Outbox pattern. The Outbox pattern requires a separate process to relay messages, which adds infrastructure dependency. Instead, FennFlow uses a Saga-like flow driven entirely by the backend.

Saga is a pattern for managing multi-step operations without distributed transactions. Each step is executed independently and paired with a compensation action that can undo it. If any step fails, compensations are run in reverse order to restore consistency.

### Operation lifecycle

Every write operation goes through these steps:

```text
register as PENDING in backend
        │
        ▼
execute against file storage
        │
   ┌────┴────┐
commit      rollback
   │            │
mark DONE   compensate in reverse order
```

**Commit path** (`uow.commit()`):

1. Fetch all `PENDING` operations from the backend.
1. Mark each one as successful in the backend and flush.
1. Finalize each operation (cleanup of temporary resources).

**Rollback path** (`uow.rollback()`):

1. Fetch all `PENDING` operations from the backend in reverse order.
1. For each operation, run its compensation against the file storage.
1. Mark compensated operations and flush.
1. Finalize.

Compensation logic is operation-specific. For instance:

- `PUT` compensation — deletes the uploaded file.
- `DELETE` compensation — restores the file from its temporary copy (see below).

### How write operation compensation works

Write operations that physically modify the file storage need a way to undo their effect on rollback. FennFlow handles this by preserving enough state before executing the operation so that compensation can fully reverse it.

A `DELETE`, for example, does not immediately remove the file. Instead, on execute, the file is **copied to a temporary path** inside the same storage:

```text
fennflow/tmp/session_{session_id}/operation_{operation_id}/{original_path}
```

The original file is then deleted. If rollback is needed, the compensation step copies the file back from `fennflow/tmp/` to its original path. If commit succeeds, the `fennflow/tmp/` file is removed during finalize.

Temporary files live in the file storage itself, not in memory. This pattern applies to all write operations that require compensation.

### Session isolation

A file involved in any pending operation — whether CREATE, PUT, or DELETE — is invisible to all other sessions until that operation is committed. This is a conservative visibility model: FennFlow errs on the side of hiding a file rather than exposing potentially inconsistent state to concurrent sessions.
# Core Concepts

# Unit of Work

`UnitOfWork` is the main entry point of FennFlow. Every file operation must go through it. It owns the session, coordinates the backend and connector, and enforces the Saga compensation flow.

```python
async with UOW() as uow:
    await uow.user_files.at("user1/").put(file)
```

## Defining a UoW

Subclass `UnitOfWork`, declare your repositories as class-level fields, and set a `config`:

```python
from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import SqlalchemyBackendConfig
from fennflow.connectors import S3ConnectorConfig
from fennflow.repositories import CreateRepository, GetRepository, S3RepoField


class MyRepo(CreateRepository, GetRepository):
    pass


class UOW(UnitOfWork):
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(),
        connector=S3ConnectorConfig(...),
        )
    user_files = S3RepoField(MyRepo, bucket_name="my-bucket")
```

All fields declared on the class become repository instances scoped to the session on first access.

## Lifecycle

Opening a UoW with `async with` does three things in order:

1. Opens the backend and connector concurrently.
1. Runs the reconciler according to the ReconcileConfig (see [Reconciler](https://alex-fir-it.github.io/core_concepts/reconciler/index.md)).
1. Returns the UoW instance ready for operations.

On exit, the UoW either commits or rolls back, then closes both backend and connector.

## auto_commit

By default `auto_commit=True`. The UoW commits if the `async with` block exits cleanly, and rolls back if an exception is raised.

Set `auto_commit=False` to control commit manually:

```python
async with UOW(auto_commit=False) as uow:
    await uow.user_files.at("user1/").put(file)
    await uow.commit()
```

# Connectors

A connector performs the **actual I/O** against your file storage. Where the backend tracks what should exist, the connector is what physically puts, gets, deletes, and lists objects.

Connectors are not called directly by user code. The UoW and repositories route all operations through the connector internally.

## S3Connector

Backed by `aiobotocore`. Compatible with AWS S3 and any S3-compatible storage (MinIO, Selectel, etc.).

```python
from fennflow.connectors import S3ConnectorConfig


class UOW(UnitOfWork):
    config = ConfigDict(
        connector=S3ConnectorConfig()
```

If you omit credentials, the standard AWS credential chain is used (environment variables, `~/.aws/credentials`, IAM roles, etc.). For a quick start you can declare the following env vars in your .env file:

```ini
AWS_ACCESS_KEY_ID = ...
AWS_SECRET_ACCESS_KEY = ...
AWS_DEFAULT_REGION = ...
AWS_ENDPOINT_URL = ...
```

`S3ConnectorConfig.aiobotocore_config` accepts an `AioConfig` instance for advanced client tuning (timeouts, retries, connection pool size, etc.).

## InMemoryConnector

Stores files in a class-level dictionary within the process. Intended for tests and local development.

```python
from fennflow.connectors import InMemoryConnectorConfig


class UOW(UnitOfWork):
    config = ConfigDict(
        connector=InMemoryConnectorConfig(),
        )
```

Call `InMemoryConnector.drop_all()` between tests to reset connector state.

## Default config

If `config` is omitted entirely from a UoW subclass, backend and connector default to Sqlalchemy and S3Connector accordingly.

## Implementing a custom connector

Subclass `AbstractConnector` and register it in `connector_registry`.

For available and planned connectors, see the [roadmap](https://github.com/users/Alex-FIR-IT/projects/2/views/2). If you need a connector that isn't there, [open an issue](https://github.com/Alex-FIR-IT/FennFlow/issues).

# Backends

A backend stores **operation metadata** — which files exist, their current status, and which session owns a pending operation. It is the source of truth for FennFlow: if a file is not in the backend, it is invisible to the UoW regardless of what the file storage contains.

Backends do not store file data. That is the connector's responsibility.

## Why a separate backend?

Raw file storage (S3, etc.) has no concept of pending operations, sessions, or compensation. The backend gives FennFlow a consistent view of state that it can query and update independently of the storage. This is what makes the Saga flow possible: the backend is always consulted before any storage operation, and its state is what commit and rollback act on.

## SQLAlchemyBackend

The default backend. Stores metadata in a database via SQLAlchemy. Defaults to a local SQLite file (`fennflow.db`) with no external infrastructure required.

```python
from fennflow.backends import SqlalchemyBackendConfig


class UOW(UnitOfWork):
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(),
        )
```

`SqlalchemyBackendConfig` accepts the following fields:

- `database_url` — any SQLAlchemy-compatible async URL. Defaults to `sqlite+aiosqlite:///fennflow.db`. Requires `aiosqlite` to be installed. Pass an explicit URL to use a different database.
- `db_schema` — schema name for FennFlow's tables. Defaults to `"fennflow"`.
- `table_name` — table name for metadata records. Defaults to `"metadata"`.
- `scope` — label to isolate backend state. Defaults to `"default"`. Use different scopes when running multiple UoW classes in the same process that point to different storage instances, to prevent their metadata from merging.

```python
from fennflow.backends import SqlalchemyBackendConfig


class AWSUOW(UnitOfWork):
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(scope="aws"),
        )


class MinIOUOW(UnitOfWork):
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(scope="minio"),
        )
```

## InMemoryBackend

`InMemoryBackendConfig` is a subclass of `SqlalchemyBackendConfig` with the `database_url` pointing to an in-memory SQLite database:

```python
from fennflow.backends import InMemoryBackendConfig


class TestUOW(AppUOW):
    config = ConfigDict(
        backend=InMemoryBackendConfig(),
        )
```

## Implementing a custom backend

Subclass `AbstractBackend`, create a factory and register it in `backend_registry`.

For available and planned backends, see the [roadmap](https://github.com/users/Alex-FIR-IT/projects/2/views/2).

If you need a backend that isn't there, [open an issue](https://github.com/Alex-FIR-IT/FennFlow/issues).

# Repositories

Repositories are the interface through which user code interacts with file storage. They are declared as fields on a UoW subclass and expose typed, Saga-tracked operations.

## Mixins

FennFlow uses a mixin pattern. You compose a repository class from the capabilities you need:

```python
from fennflow.repositories import (
    PutRepository,
    GetRepository,
    DeleteRepository,
    ListRepository,
    )


class CrudRepository(
    PutRepository,
    DeleteRepository,
    GetRepository,
    ListRepository,
    ):
    pass
```

Available mixins:

### Base Mixins

| Mixin                            | Operation                            | Participates in Saga | Backend    | Connector | Supported By                  |
| -------------------------------- | ------------------------------------ | -------------------- | ---------- | --------- | ----------------------------- |
| `GetRepository`                  | Download one or more files           | No (read-only)       | read       | read      | S3Connector InMemoryConnector |
| `ListRepository`                 | List files by prefix with pagination | No (read-only)       | read       | —         | S3Connector InMemoryConnector |
| `GeneratePresignedUrlRepository` | Generate presigned URL               | No (read-only)       | read       | read      | S3Connector                   |
| `PutRepository`                  | Upsert one or more files             | Yes                  | read+write | write     | S3Connector InMemoryConnector |
| `CreateRepository`               | Upload one or more files             | Yes                  | read+write | write     | S3Connector InMemoryConnector |
| `DeleteRepository`               | Delete a file                        | Yes                  | read+write | write     | S3Connector InMemoryConnector |

### Prefix Mixins

| Mixin                                  | Operation                                  | Participates in Saga | Backend    | Connector | Supported By                  |
| -------------------------------------- | ------------------------------------------ | -------------------- | ---------- | --------- | ----------------------------- |
| `GetPrefixRepository`                  | Download a whole prefix                    | No (read-only)       | read       | read      | S3Connector InMemoryConnector |
| `GeneratePresignedUrlPrefixRepository` | Generate presigned URLs for a whole prefix | No (read-only)       | read       | read      | S3Connector                   |
| `DeletePrefixRepository`               | Delete a whole prefix                      | Yes                  | read+write | write     | S3Connector InMemoryConnector |

Read-only operations (such as `GetRepository`) consult the backend before touching storage. If the backend has no record of a file, no network request is made.

## RepoField and S3RepoField

Attach a repository class to a UoW using a field descriptor:

```python
from fennflow.repositories import RepoField, S3RepoField


class UOW(UnitOfWork):
    # generic — namespace maps to whatever the connector uses as a bucket/prefix
    user_files = RepoField(CrudRepository, namespace="user-files")

    # S3-specific alias — bucket_name is just a named alias for namespace
    user_files = S3RepoField(CrudRepository, bucket_name="user-files")
```

`S3RepoField` is a convenience wrapper around `RepoField`. Functionally identical — it exists to make intent explicit when working with S3.

The field is lazily initialized: the repository instance is created on first attribute access within a session.

## Path scoping with `.at()`

All repository mixins inherit `.at()` from `AtRepository`. It returns a new repository instance scoped to the given path, without modifying the original:

```python
async with UOW() as uow:
    # scope to a subfolder
    storage = uow.user_files.at("user1/docs")
    await storage.put(file)

    # chain further
    await uow.user_files.at("user1").at("docs").put(file)
```

`.at()` calls are additive and composable. The path is normalized internally. The original field (`uow.user_files`) is unchanged after `.at()`.

`.cwd` returns the current normalized path of a scoped repository instance.

# Files

FennFlow wraps raw bytes in typed content models instead of passing `bytes` and `dict` directly. Every file you put into or get from storage is represented as one of these models.

All **binary** content types inherit from `BaseBinary`, which itself inherits from `BaseContent` (a Pydantic model).

## Content types

| Class             | Media type         | Notes                                                                   |
| ----------------- | ------------------ | ----------------------------------------------------------------------- |
| `BaseBinary`      | any                | Base class. Use when no specific type fits                              |
| `BinaryContent`   | any                | Extends `BaseBinary` with "from_local_path" classmethod                 |
| `TextContent`     | `text/plain`       | Stores text as UTF-8 bytes internally. `.content` returns `str`         |
| `JsonContent`     | `application/json` | Stores JSON as UTF-8 bytes. `.content` returns the parsed Python object |
| `ImageContent`    | `image/*`          | Extends `BaseBinary` with optional `width` and `height` fields          |
| `AudioContent`    | `audio/*`          | Extends `BaseBinary` with optional `duration` field                     |
| `VideoContent`    | `video/*`          | Extends `BaseBinary` with optional `duration`, `width`, `height` fields |
| `DocumentContent` | any document type  | Thin subclass of `BaseBinary`                                           |

There is also a model representing url files:

| Class        | Media type | Notes                                                 |
| ------------ | ---------- | ----------------------------------------------------- |
| `UrlContent` | any        | Stores a URL string instead of bytes. `data` is `str` |

## Creating content

`TextContent` and `JsonContent` expose a `from_content()` classmethod as the primary constructor:

```python
from fennflow.files import (
    BinaryContent,
    ContentFactory,
    ImageContent,
    JsonContent,
    MediaType,
    TextContent,
    )

text = TextContent.from_content("Hello, world!")
json_file = JsonContent.from_content({"key": "value"})
json_list = JsonContent.from_content([1, 2, 3])
from_local_path_binary = BinaryContent.from_local_path("my_file.txt")

# Manul initialization requires explicit media_type or filename
binary = BinaryContent(
    data=b"...",
    media_type=MediaType.APPLICATION_OCTET_STREAM,
    )

# Optional metadata fields
image = ImageContent(
    data=img_bytes,
    media_type=MediaType.IMAGE_PNG,
    width=800,
    height=600,
    )

# ContentFactory can be used to get specific class of content
file: TextContent = (
    ContentFactory.from_bytes(
        media_type=MediaType.TEXT_PLAIN,
        data=file_bytes,
        **metadata,
        ),
    )
```

## Filename and media type resolution

Both `filename` and `media_type` are optional at construction time. FennFlow resolves them:

- If only `media_type` is given: filename is generated as the SHA-256 hash of the data, with the extension guessed from the media type.
- If only `filename` is given: media type is guessed from the extension via `mimetypes`.
- If both are omitted: raises `FileNameAndMediaTypeBothNoneException`.
- If the filename has no extension: the extension is guessed from the media type and appended.

A warning is logged if the file extension and media type do not agree.

## Extra metadata

Any keyword arguments not matching declared fields are collected into `extra_metadata: dict[str, str]`. This metadata is forwarded to the connector (e.g. stored as S3 object metadata).

## ContentFactory

When FennFlow retrieves a file from storage, it reconstructs the appropriate content type using `ContentFactory.from_bytes()`. The factory resolves the class from a registry by MIME type match falling back to `BaseBinary` for unknown types.

You can register custom content types in `content_registry` to have them returned automatically on `get`.

# Path Templates

Raw S3 paths scattered across a codebase are a maintenance problem. A path like `credentials/user_id_123/passport` written inline in five different places is five places to update when the layout changes, and five places to introduce a typo.

Path templates solve this by centralizing path construction logic in a single, typed, testable place — without adding any fancy magic concepts.

## The Problem

`.at()` accepts a plain string, which means path construction is entirely the caller's responsibility:

```python
async with AppUOW() as uow:
    await uow.credentials.at(f"credentials/user_id_{user_id}/passport").put(passport)
```

This works, but it pushes path knowledge into call sites. Any refactor of your storage layout requires hunting down every string.

## PathTemplate Protocol

FennFlow provides a minimal protocol for path templates:

```python
from fennflow.path_template import PathTemplate
```

```python
class PathTemplate(Protocol):
    def render(
            self
            ) -> str: ...
```

`.at()` accepts either a `str` or any object that satisfies `PathTemplate` — i.e. any object with a `render() -> str` method. No base class to inherit, no registration required.

## Defining Templates

Define your templates as plain dataclasses in your application code and then use them directly with .at():

```python
import asyncio
from dataclasses import dataclass

from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import InMemoryBackendConfig
from fennflow.connectors import InMemoryConnectorConfig
from fennflow.repositories import GetRepository, S3RepoField


class AppUOW(UnitOfWork):
    avatars = S3RepoField(GetRepository, bucket_name="avatars")
    credentials = S3RepoField(GetRepository, bucket_name="credentials")

    config = ConfigDict(
        backend=InMemoryBackendConfig(),
        connector=InMemoryConnectorConfig(),
    )


@dataclass(slots=True)
class AvatarPath:
    user_id: int

    def render(self) -> str:
        return f"avatars/user_{self.user_id}"


@dataclass(slots=True)
class PassportPath:
    user_id: int

    def render(self) -> str:
        return f"credentials/user_{self.user_id}/passport"


@dataclass(slots=True)
class DriverLicensePath:
    user_id: int

    def render(self) -> str:
        return f"credentials/user_{self.user_id}/driver_license"


async def main():
    user_id = 535

    async with AppUOW() as uow:
        avatar_storage = uow.avatars.at(AvatarPath(user_id=user_id))
        assert avatar_storage.cwd == "avatars/user_535/"

        passport_storage = uow.credentials.at(PassportPath(user_id=user_id))
        assert passport_storage.cwd == "credentials/user_535/passport/"

        driver_license_storage = uow.credentials.at(DriverLicensePath(user_id=user_id))
        assert driver_license_storage.cwd == "credentials/user_535/driver_license/"


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

The IDE resolves `AvatarPath(user_id=user_id)` as a typed constructor. Mypy checks the field types. Autocomplete works on all template fields. If `user_id` changes type or a new required field is added, the type checker catches every broken call site immediately.

## Multi-Parameter Templates

Templates are dataclasses — they can hold as many fields as needed:

```python
@dataclass(slots=True)
class ReportPath:
    org_id: int
    year: int
    month: int

    def render(
            self
            ) -> str:
        return f"reports/org_{self.org_id}/{self.year}/{self.month:02d}"
```

```python
await uow.reports.at(ReportPath(org_id=1, year=2025, month=6)).put(report)
```

## Chaining Templates

Since `.at()` is chainable and templates are just another form of path segment, templates and plain strings can be mixed and chained freely:

```python
@dataclass(slots=True)
class UserPrefix:
    user_id: int

    def render(
            self
            ) -> str:
        return f"users/user_{self.user_id}"


@dataclass(slots=True)
class YearMonth:
    year: int
    month: int

    def render(
            self
            ) -> str:
        return f"{self.year}/{self.month:02d}"
```

```python
async with AppUOW() as uow:
    # two templates chained
    await uow.files.at(UserPrefix(user_id=1)).at(YearMonth(year=2025, month=6)).put(report)

    # template chained with a plain string
    await uow.files.at(UserPrefix(user_id=1)).at("archive").put(backup)
```

# Reconciler

The reconciler re-syncs the backend with what the file storage actually contains. Without reconciliation, the backend would report no files exist even though the storage is full — making everything invisible to `GetRepository`, `ListRepository`, etc.

## When it runs

Reconciliation is configured via `ReconcileConfig` inside `ConfigDict`. These are default settings:

```python
from fennflow.reconciler import ReconcileConfig, ReconcileFrequencyEnum, ReconcileStrategyEnum


class UOW(UnitOfWork):
    config = ConfigDict(
        reconcile=ReconcileConfig(
            frequency=ReconcileFrequencyEnum.ON_START_APP,
            strategy=ReconcileStrategyEnum.FILL_IF_EMPTY,
            batch_size=1000,
            ),
        )
```

**Frequency options:**

| Value              | Behavior                                     |
| ------------------ | -------------------------------------------- |
| `ON_START_APP`     | Runs once per process lifetime per UoW class |
| `ON_SESSION_START` | Runs on every `async with UOW()` entry       |
| `NEVER`            | Disabled                                     |

**Strategy options:**

| Value            | Behavior                                                                                                           |
| ---------------- | ------------------------------------------------------------------------------------------------------------------ |
| `FILL_IF_EMPTY`  | Only reconciles if the backend has no records at all                                                               |
| `INSERT_MISSING` | Inserts records for files found in storage that the backend doesn't know about, leaving existing records untouched |
| `REPLACE`        | Replaces all backend state with what storage contains                                                              |

## What the reconciler does

On trigger, the reconciler iterates all `RepoField` instances declared on the UoW class. For each one, it paginates through the file storage (via the connector's `list_objects`) and inserts the found paths into the backend as `UPLOADED` records, according to the chosen conflict strategy.

> More info here - [Core Concepts — Reconciler](https://alex-fir-it.github.io/advanced_features/reconciler/index.md).
# Advanced Features

# Reconciler — Manual Usage

> New to the reconciler? Start with [Core Concepts — Reconciler](https://alex-fir-it.github.io/core_concepts/reconciler/index.md) first.

By default, reconciliation is triggered automatically on UoW entry according to `ReconcileConfig`. In some cases you may want to trigger it manually — for example, to force a `REPLACE` sync mid-session after an external process has modified the storage directly, or to reconcile on demand without changing the UoW's default config.

```python
import asyncio

from fennflow.reconciler import Reconciler, ReconcileStrategyEnum
from fennflow.uow import UowInspector


async def main():
    async with UOW() as uow:
        uow_inspector = UowInspector(uow=uow)
        reconciler = Reconciler(
            uow_fields=uow_inspector.get_repo_fields(),
            connector=uow.connector,
            backend=uow.backend,
            )
        await reconciler.reconcile(
            session_id=uow._session_id,
            batch_size=500,
            strategy=ReconcileStrategyEnum.REPLACE,
            backend_scope=uow.config["backend"].scope
            )


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

`UowInspector` introspects the UoW class and yields all declared `RepoField` instances. `Reconciler` then paginates through each field's namespace in the file storage and syncs the backend according to the chosen strategy.

Note that manual reconciliation runs inside an open session. Any `PENDING` operations from the current session are not affected — reconciliation only touches `UPLOADED` records in the persistent backend storage, not the session-local operation buffer.
# Examples

# Examples

Here we include some examples of how to use FennFlow and what it can do.

## Usage

These examples are distributed with `fennflow` so you can run them by cloning the [fennflow repo](https://github.com/Alex-FIR-IT/FennFlow).

# Working with File Content

FennFlow wraps raw bytes in typed content models. Instead of juggling `bytes` and `dict`, you work with `TextContent`, `JsonContent`, `BinaryContent` and others — each with typed accessors for their data.

## Example: Creating, uploading and retrieving typed files

```python
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 (
    CreateRepository,
    GetRepository,
    ListRepository,
    S3RepoField,
)


class ReadWriteRepository(CreateRepository, GetRepository, ListRepository):
    pass


class AppUOW(UnitOfWork):
    files = S3RepoField(ReadWriteRepository, bucket_name="my-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():
    # Three ways to create file content
    text_file = TextContent.from_content("Hello, world!")
    json_file = JsonContent.from_content({"user": "alice", "score": 42})

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

    async with AppUOW() as uow:
        await uow.files.at("folder/").create(text_file, json_file, binary_file)

        paths = await uow.files.at("folder/").list()
        response = await uow.files.get(*paths)

        print(response.texts[0].content)  # "Hello, world!"
        print(response.texts[1].filename)  # "my_file.txt"

        print(response.filter(JsonContent)[0].content)  # {"user": "alice", "score": 42}
        print(
            response.filter_by_media_type(MediaType.APPLICATION_OCTET_STREAM)[0].data
        )  # b"some bytes"


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

`TextContent.content` returns a `str`, `JsonContent.content` returns the parsed Python object — no manual `decode()` or `json.loads()` needed.

MediaResponse uses `ContentFactory` to resolve the correct class on retrieval automatically based on the stored MIME type, so `.texts`, `.filter()`, `.filter_by_media_type()` and other typed accessors on `MediaResponse` always return the right types.

# Using Multiple Connectors

FennFlow makes it trivial to work with multiple object storages simultaneously. Each `UnitOfWork` owns its own connector and backend scope, so you can compose them freely in the same coroutine.

## Example: Migrating files from AWS S3 to MinIO

```python
import asyncio

from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import SqlalchemyBackendConfig
from fennflow.connectors import S3ConnectorConfig
from fennflow.repositories import (
    GetRepository,
    ListRepository,
    PutRepository,
    S3RepoField,
)


class ReadRepository(GetRepository, ListRepository):
    pass


class WriteRepository(PutRepository):
    pass


class AWSS3UOW(UnitOfWork):
    files = S3RepoField(ReadRepository, bucket_name="my-bucket")
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(scope="aws"),
        connector=S3ConnectorConfig(
            endpoint_url="https://s3.amazonaws.com",
            aws_access_key_id="aws-key",
            aws_secret_access_key="aws-secret",
        ),
    )


class MinIOUOW(UnitOfWork):
    files = S3RepoField(WriteRepository, bucket_name="my-bucket")
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(scope="minio"),
        connector=S3ConnectorConfig(
            endpoint_url="https://minio.example.com",
            aws_access_key_id="minio-key",
            aws_secret_access_key="minio-secret",
        ),
    )


async def main():
    async with AWSS3UOW() as aws, MinIOUOW() as minio:
        paths = await aws.files.at("folder/").list()

        for path in paths:
            response = await aws.files.get(path)
            await minio.files.at("folder/").put(*response)


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

The two UoWs are fully independent — different credentials, different endpoints, different backend scopes. If anything fails mid-transfer, each UoW rolls back its own pending operations independently.

# Saga Compensation in Action

FennFlow's Unit of Work automatically rolls back all pending operations if something goes wrong. Combined with `auto_commit=False`, you get full control over when a transaction is committed — making it trivial to tie file storage to the result of an external service call.

## Example: Upload avatar and verify with external face verification API

```python
import asyncio

import httpx

from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import SqlalchemyBackendConfig
from fennflow.connectors import S3ConnectorConfig
from fennflow.files import BinaryContent, ImageContent
from fennflow.repositories import (
    CreateRepository,
    GeneratePresignedUrlRepository,
    S3RepoField,
)


class AvatarRepository(CreateRepository, GeneratePresignedUrlRepository):
    pass


class AppUOW(UnitOfWork):
    avatars = S3RepoField(AvatarRepository, bucket_name="avatars")
    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 process_avatar(user_id: str, avatar: ImageContent) -> None:
    async with AppUOW(auto_commit=False) as uow:
        await uow.avatars.at(user_id).create(avatar)

        presigned_url = await uow.avatars.generate_presigned_url(avatar.filename)

        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://verify.example.com/face",
                json={"url": presigned_url},
            )

        if response.is_success:
            await uow.commit()
        # if verification failed or an exception was raised,
        # the UoW rolls back on exit — avatar is removed from S3 automatically


async def main():
    avatar = BinaryContent.from_local_path("avatar.jpg")
    await process_avatar(user_id="user-123", avatar=avatar)


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

No manual cleanup code. No try/finally. If the verification API rejects the image or raises an exception, the UoW exits without a commit and the uploaded avatar is automatically compensated out of S3. You can use auto_commit=True and call uow.rollback on failure as well.

# 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

```python
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.

# Testing with FennFlow

FennFlow is designed to be testable without mocks. Swap `S3ConnectorConfig` for `InMemoryConnectorConfig` and point the backend at an in-memory SQLite database — your tests run with zero infrastructure and zero mocks.

## Example: Testing user registration with avatar upload

```python
import pytest
import pytest_asyncio

from fennflow import ConfigDict, UnitOfWork
from fennflow.backends import InMemoryBackendConfig, SqlalchemyBackendConfig
from fennflow.connectors import InMemoryConnectorConfig, S3ConnectorConfig
from fennflow.files import ImageContent
from fennflow.repositories import CreateRepository, GetRepository, S3RepoField


class AvatarRepository(CreateRepository, GetRepository):
    pass


class AppUOW(UnitOfWork):
    avatars = S3RepoField(AvatarRepository, bucket_name="avatars")
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(),
        connector=S3ConnectorConfig(),
    )


class TestUOW(AppUOW):
    config = ConfigDict(
        backend=InMemoryBackendConfig(),
        connector=InMemoryConnectorConfig(),
    )


class DbClient:
    async def create_user(self, user_id: str, avatar_path: str) -> None: ...


async def register_user(
    user_id: str,
    avatar: ImageContent,
    uow: AppUOW,
    db: DbClient,
) -> None:
    async with uow:
        await uow.avatars.at(user_id).create(avatar)
        await db.create_user(user_id=user_id, avatar_path=avatar.filename)


@pytest_asyncio.fixture
def db():
    return DbClient()


@pytest.mark.asyncio
async def test_register_user_saves_avatar(db):
    avatar = ImageContent(
        data=b"fake-image-bytes",
        media_type="image/jpeg",
        filename="avatar.jpg",
    )

    await register_user(user_id="user-123", avatar=avatar, uow=TestUOW(), db=db)

    async with TestUOW() as uow:
        response = await uow.avatars.at("user-123").get(avatar.filename)
        assert response.images[0].content.filename == "avatar.jpg"
```

`TestUOW` inherits everything from `AppUOW` and overrides only the config — same repositories, same business logic, different infrastructure. No mocks, no patching, no Docker.
# API Reference

# API

## `UnitOfWork`

Unit of Work (UOW) — the main entry point of FennFlow.

It coordinates file operations by managing:

- backend (operation metadata storage)
- connector (actual storage, e.g. S3)
- execution and compensation logic (Saga pattern)

**Example**::

```text
class UOW(UnitOfWork):
    config = ConfigDict(
        backend=PostgresBackendConfig(...),
        connector=S3ConnectorConfig(...),
    )
    user_files = S3RepoField(UserFiles, bucket_name="bucket_name")
    # or
    # user_files = RepoField(UserFiles, namespace="bucket_name")

async with UOW() as uow:
    await uow.user_files.at("user1/").put(file)
```

**Behavior**:

- By default, `auto_commit=True`: commits all operations if the context exits successfully
- If an exception occurs or `auto_commit=False`: triggers rollback with compensation logic

Important:

- Users should NOT interact with backend or connector directly
- All operations must go through UOW
- Rollback applies compensation in reverse order (Saga pattern)

Source code in `src/fennflow/uow/core.py`

```python
class UnitOfWork:
    """Unit of Work (UOW) — the main entry point of FennFlow.

    It coordinates file operations by managing:
    - backend (operation metadata storage)
    - connector (actual storage, e.g. S3)
    - execution and compensation logic (Saga pattern)

    **Example**::

        class UOW(UnitOfWork):
            config = ConfigDict(
                backend=PostgresBackendConfig(...),
                connector=S3ConnectorConfig(...),
            )
            user_files = S3RepoField(UserFiles, bucket_name="bucket_name")
            # or
            # user_files = RepoField(UserFiles, namespace="bucket_name")

        async with UOW() as uow:
            await uow.user_files.at("user1/").put(file)

    **Behavior**:
    - By default, `auto_commit=True`:
        commits all operations if the context exits successfully
    - If an exception occurs or `auto_commit=False`:
        triggers rollback with compensation logic

    Important:
    - Users should NOT interact with backend or connector directly
    - All operations must go through UOW
    - Rollback applies compensation in reverse order (Saga pattern)

    """

    config: ConfigDict | None = None

    def __init__(
        self,
        auto_commit: bool = True,
    ):
        self._auto_commit = auto_commit
        self._session_id = uuid.uuid4()
        self._resolved_config = ConfigResolver.resolve_config(self.config)

        self._backend = BackendFactory.from_config(config=self._resolved_config.backend)
        self._connector = ConnectorFactory.from_config(
            config=self._resolved_config.connector
        )
        self._operation_executor = OperationExecutor(
            connector=self.connector,
        )

    @property
    def backend(self) -> BackendOrchestrator:
        """Direct access to the backend for read-only inspection.

        Warning: mutating backend state directly bypasses Saga guarantees.
        Use UoW methods for all write operations.
        """
        return self._backend

    @property
    def connector(self) -> AbstractConnector:
        """Direct access to the connector.

        Warning: operations performed directly on the connector
        are not tracked by the backend.
        Therefore, they will not be compensated by uow.
        """
        return self._connector

    async def __aenter__(
        self,
    ):
        try:
            await asyncio.gather(
                self.connector.open(),
                self.backend.open(),
            )

            await ReconcileOrchestrator().reconcile_if_needed(uow=self)

            return self

        except Exception:
            await self._cleanup()
            raise

    async def __aexit__(
        self,
        exc_type,
        exc,
        tb,
    ):
        try:
            if exc_type is not None or not self._auto_commit:
                await self.rollback()
            elif self._auto_commit:
                await self.commit()
        finally:
            await self._cleanup()

    async def _finalize_operation(self, operation: OperationRecord) -> None:
        try:
            await self._operation_executor.finalize(operation)
        except Exception:
            logger.warning(
                "Finalization failed.",
                extra={
                    "operation_id": operation.record.operation_id,
                    "session_id": operation.record.session_id,
                },
                exc_info=True,
            )

    async def _finalize_operations(self, operations: Iterable[OperationRecord]) -> None:
        await asyncio.gather(
            *(self._finalize_operation(op) for op in operations),
            return_exceptions=True,
        )

    async def commit(
        self,
    ) -> None:
        """Persists operation state via backend."""
        operations = self.backend.session_buffer.get_all()

        if operations:
            for operation in operations:
                operation.record.mark_done()

            with suppress(Exception):
                await self._finalize_operations(operations)
        await self.backend.commit()

    async def rollback(
        self,
    ) -> None:
        """Performs rollback for saga flow.

        Runs compensation for all pending operations
        and then rolls back backend state.
        """
        operations = self.backend.session_buffer.get_all()
        finalize_operations = []
        for operation in reversed(tuple(operations)):
            try:
                await self._operation_executor.compensate(operation)
            except Exception as e:
                logger.exception(
                    "Compensation failed.",
                    extra={
                        "operation_id": operation.record.operation_id,
                        "session_id": operation.record.session_id,
                    },
                )
                operation.record.mark_compensation_failed(error=str(e))

            else:
                finalize_operations.append(operation)

        with suppress(Exception):
            await self._finalize_operations(finalize_operations)
        await self.backend.commit()

    async def _cleanup(self) -> None:
        await asyncio.gather(
            self.connector.close(),
            self.backend.close(),
            return_exceptions=True,
        )
```

### `backend`

Direct access to the backend for read-only inspection.

Warning: mutating backend state directly bypasses Saga guarantees. Use UoW methods for all write operations.

### `connector`

Direct access to the connector.

Warning: operations performed directly on the connector are not tracked by the backend. Therefore, they will not be compensated by uow.

### `commit()`

Persists operation state via backend.

Source code in `src/fennflow/uow/core.py`

```python
async def commit(
    self,
) -> None:
    """Persists operation state via backend."""
    operations = self.backend.session_buffer.get_all()

    if operations:
        for operation in operations:
            operation.record.mark_done()

        with suppress(Exception):
            await self._finalize_operations(operations)
    await self.backend.commit()
```

### `rollback()`

Performs rollback for saga flow.

Runs compensation for all pending operations and then rolls back backend state.

Source code in `src/fennflow/uow/core.py`

```python
async def rollback(
    self,
) -> None:
    """Performs rollback for saga flow.

    Runs compensation for all pending operations
    and then rolls back backend state.
    """
    operations = self.backend.session_buffer.get_all()
    finalize_operations = []
    for operation in reversed(tuple(operations)):
        try:
            await self._operation_executor.compensate(operation)
        except Exception as e:
            logger.exception(
                "Compensation failed.",
                extra={
                    "operation_id": operation.record.operation_id,
                    "session_id": operation.record.session_id,
                },
            )
            operation.record.mark_compensation_failed(error=str(e))

        else:
            finalize_operations.append(operation)

    with suppress(Exception):
        await self._finalize_operations(finalize_operations)
    await self.backend.commit()
```

## `UowInspector`

Extracts info from Unit of Work.

Source code in `src/fennflow/uow/inspector.py`

```python
@dataclass(slots=True, frozen=True)
class UowInspector:
    """Extracts info from Unit of Work."""

    uow: UnitOfWork

    def get_repo_fields(self) -> Generator[RepoField, None, None]:
        """Yields repo fields.

        Example::

            class UOW(UnitOfWork):
                images = S3RepoField(...)
                videos: S3RepoField(...)

            extractor = UowInspector(uow=uow)
            extractor.get_repo_fields() # yield images, then videos fields
        """
        for field in vars(type(self.uow)).values():
            if isinstance(field, RepoField):
                yield field
```

### `get_repo_fields()`

Yields repo fields.

Example::

```text
class UOW(UnitOfWork):
    images = S3RepoField(...)
    videos: S3RepoField(...)

extractor = UowInspector(uow=uow)
extractor.get_repo_fields() # yield images, then videos fields
```

Source code in `src/fennflow/uow/inspector.py`

```python
def get_repo_fields(self) -> Generator[RepoField, None, None]:
    """Yields repo fields.

    Example::

        class UOW(UnitOfWork):
            images = S3RepoField(...)
            videos: S3RepoField(...)

        extractor = UowInspector(uow=uow)
        extractor.get_repo_fields() # yield images, then videos fields
    """
    for field in vars(type(self.uow)).values():
        if isinstance(field, RepoField):
            yield field
```

## `ConfigDict`

Bases: `TypedDict`

Configuration for a UnitOfWork instance.

All fields are optional — if not provided, defaults are used.

Attributes:

| Name        | Type              | Description                                                            |
| ----------- | ----------------- | ---------------------------------------------------------------------- |
| `backend`   | `BackendConfig`   | Configuration for the metadata backend (e.g. SqlalchemyBackendConfig). |
| `connector` | `ConnectorConfig` | Configuration for the storage connector (e.g. S3ConnectorConfig).      |

Example::

```text
class UOW(UnitOfWork):
    config = ConfigDict(
        backend=SqlalchemyBackendConfig(),
        connector=S3ConnectorConfig(),

    )
```

Source code in `src/fennflow/core/configs/base.py`

```python
class ConfigDict(TypedDict, total=False):
    """Configuration for a UnitOfWork instance.

    All fields are optional — if not provided, defaults are used.

    Attributes:
        backend: Configuration for the metadata backend
            (e.g. ``SqlalchemyBackendConfig``).
        connector: Configuration for the storage connector (e.g. ``S3ConnectorConfig``).

    Example::

        class UOW(UnitOfWork):
            config = ConfigDict(
                backend=SqlalchemyBackendConfig(),
                connector=S3ConnectorConfig(),

            )
    """

    backend: BackendConfig
    connector: ConnectorConfig
    reconcile: ReconcileConfig
```

## `InMemoryBackendConfig`

Bases: `SqlalchemyBackendConfig`

Configuration for the in-memory SQLite backend.

Source code in `src/fennflow/backends/in_memory/config.py`

```python
class InMemoryBackendConfig(SqlalchemyBackendConfig):
    """Configuration for the in-memory SQLite backend."""

    database_url: str = Field(
        default="sqlite+aiosqlite:///file:memdb1?mode=memory",
        description="Database URL for the in-memory SQLite database",
    )
```

## `SqlalchemyBackendConfig`

Bases: `AbstractBackendConfig`

Configuration for the SqlAlchemy backend.

No configuration is required — the in-memory backend is zero-dependency.

Source code in `src/fennflow/backends/sqlalchemy/config.py`

```python
class SqlalchemyBackendConfig(AbstractBackendConfig):
    """Configuration for the SqlAlchemy backend.

    No configuration is required — the in-memory backend
    is zero-dependency.
    """

    database_url: str = Field(
        default_factory=database_url_factory,
        description="Database URL for the FennFlow database",
    )
    db_schema: str = Field(
        default="fennflow",
        description="Schema for the FennFlow database",
    )
    table_name: str = Field(
        default="metadata",
        description="Table name for the FennFlow metadata records",
    )
```

## `ConnectorFactory`

Factory for creating connector instances from config objects.

Resolves the appropriate connector class from `connector_registry` based on the config class name.

Example::

```text
connector = ConnectorFactory.from_config(S3ConnectorConfig(...))
```

Source code in `src/fennflow/connectors/_factory.py`

```python
class ConnectorFactory:
    """Factory for creating connector instances from config objects.

    Resolves the appropriate connector class from ``connector_registry``
    based on the config class name.

    Example::

        connector = ConnectorFactory.from_config(S3ConnectorConfig(...))
    """

    @staticmethod
    def from_config(config: ConnectorConfig) -> AbstractConnector:
        """Create a connector instance from a config object.

        Args:
            config: The connector configuration instance.

        Returns:
            An initialized connector instance.

        Raises:
            KeyError: If no connector is registered for the config type.
        """
        connector_cls = connector_registry.get(config.__class__.__name__)
        if not connector_cls:
            raise KeyError(f"Unknown connector for : {type(config)=}")

        return connector_cls(config=config)
```

### `from_config(config)`

Create a connector instance from a config object.

Parameters:

| Name     | Type              | Description                           | Default    |
| -------- | ----------------- | ------------------------------------- | ---------- |
| `config` | `ConnectorConfig` | The connector configuration instance. | *required* |

Returns:

| Type                | Description                        |
| ------------------- | ---------------------------------- |
| `AbstractConnector` | An initialized connector instance. |

Raises:

| Type       | Description                                        |
| ---------- | -------------------------------------------------- |
| `KeyError` | If no connector is registered for the config type. |

Source code in `src/fennflow/connectors/_factory.py`

```python
@staticmethod
def from_config(config: ConnectorConfig) -> AbstractConnector:
    """Create a connector instance from a config object.

    Args:
        config: The connector configuration instance.

    Returns:
        An initialized connector instance.

    Raises:
        KeyError: If no connector is registered for the config type.
    """
    connector_cls = connector_registry.get(config.__class__.__name__)
    if not connector_cls:
        raise KeyError(f"Unknown connector for : {type(config)=}")

    return connector_cls(config=config)
```

## `InMemoryConnector`

Bases: `AbstractConnector`

In-memory connector for file storage, primarily used for testing.

Stores files in a class-level dictionary shared across all instances,

Use `drop_all()` between tests to reset state.

Example::

```text
class UOW(UnitOfWork):
    config = ConfigDict(
        connector=InMemoryConnectorConfig(),
    )
```

Source code in `src/fennflow/connectors/in_memory/core.py`

```python
class InMemoryConnector(AbstractConnector):
    """In-memory connector for file storage, primarily used for testing.

    Stores files in a class-level dictionary shared across all instances,

    Use ``drop_all()`` between tests to reset state.

    Example::

        class UOW(UnitOfWork):
            config = ConfigDict(
                connector=InMemoryConnectorConfig(),
            )
    """

    _storage: defaultdict[Namespace, dict[StoragePath, BinaryMedia]] | None = None

    async def open(self) -> Self:
        return self

    async def close(self):
        pass

    def __init__(self, config: InMemoryConnectorConfig):
        self._config = config

        if self.__class__._storage is None:
            self.__class__._storage = defaultdict(dict)

    @property
    def storage(
        self,
    ) -> defaultdict[Namespace, dict[StoragePath, BinaryMedia]]:
        if self.__class__._storage is None:
            raise RuntimeError(
                "Cannot get in-memory storage. InMemoryConnector is not initialized.",
            )
        return self.__class__._storage

    async def put(
        self,
        file: BinaryMedia,
        repo_extra: RepoExtra,
        connector_extra: ConnectorExtra = OMIT,  # noqa: ARG002
    ) -> ConnectorRawResponse[None]:
        namespace = repo_extra["namespace"]
        self.storage[namespace][file.storage_path] = file
        logger.debug(f"{file=} uploaded to {namespace=}")

        return ConnectorRawResponse(raw_response=None)

    async def get(
        self,
        storage_path: StoragePath,
        repo_extra: RepoExtra,
        connector_extra: ConnectorExtra = OMIT,  # noqa: ARG002
    ) -> MediaResponse[None]:

        if storage_path not in self.storage[repo_extra["namespace"]]:
            return MediaResponse()

        file = self.storage[repo_extra["namespace"]][storage_path]

        content_response = ContentResponse(
            raw_response=None,
            content=ContentFactory.from_bytes(
                media_type=file.media_type,
                data=file.data,
                **file.get_metadata(),
            ),
        )

        return MediaResponse.from_content_response(response=content_response)

    async def delete(
        self,
        storage_path: StoragePath,
        repo_extra: RepoExtra,
        connector_extra: ConnectorExtra = OMIT,  # noqa: ARG002
    ) -> ConnectorRawResponse[None]:
        self.storage[repo_extra["namespace"]].pop(storage_path, None)

        return ConnectorRawResponse(raw_response=None)

    @reraise_with(NoSuchKeyException(), catch=KeyError)
    async def copy_object(
        self,
        repo_extra: RepoExtra,
        from_storage_path: StoragePath,
        to_storage_path: StoragePath,
        to_namespace: Namespace,
        connector_extra: ConnectorExtra = OMIT,  # noqa: ARG002
    ) -> ConnectorRawResponse[None]:
        file = self.storage[repo_extra["namespace"]][from_storage_path]
        self.storage[to_namespace][to_storage_path] = file

        return ConnectorRawResponse(raw_response=None)

    @classmethod
    def drop_all(cls) -> None:
        cls._storage = defaultdict(dict)

    async def list_objects(
        self,
        prefix: str,
        repo_extra: RepoExtraType,
        limit: int = 1000,
        continuation_token: Omittable[str] | None = OMIT,
        connector_extra: ConnectorExtra = OMIT,  # noqa: ARG002
    ) -> ConnectorListResponse[None]:
        filtered_storage_paths = []
        all_storage_paths = sorted(self.storage[repo_extra["namespace"]])

        if continuation_token:
            index = bisect.bisect_right(all_storage_paths, continuation_token)
        else:
            index = 0

        for storage_path in islice(all_storage_paths, index, None):
            if 0 >= limit:
                continuation_token = storage_path
                break

            if storage_path.startswith(prefix):
                filtered_storage_paths.append(storage_path)
                limit -= 1
        else:
            continuation_token = None

        return ConnectorListResponse(
            storage_paths=filtered_storage_paths,
            continuation_token=continuation_token,
            raw_response=None,
        )

    async def generate_presigned_url(
        self,
        storage_path: str,  # noqa: ARG002
        repo_extra: RepoExtraType,  # noqa: ARG002
        expires_in: Omittable[int] = OMIT,  # noqa: ARG002
        connector_extra: Omittable[Any] = OMIT,  # noqa: ARG002
    ) -> NoReturn:
        raise ConnectorCapabilityException("generate_presigned_url")
```

## `InMemoryConnectorConfig`

Bases: `AbstractConnectorConfig`

Configuration for the in-memory connector.

No configuration is required — the in-memory connector is zero-dependency and is intended for testing and development only.

Source code in `src/fennflow/connectors/in_memory/config.py`

```python
class InMemoryConnectorConfig(AbstractConnectorConfig):
    """Configuration for the in-memory connector.

    No configuration is required — the in-memory connector
    is zero-dependency and is intended for testing and development only.
    """
```

## `S3Connector`

Bases: `AbstractConnector[S3Extra]`

Connector for AWS S3-compatible object storage via aiobotocore.

Use `S3ConnectorConfig` to configure credentials, region, etc.

Example::

```text
class UOW(UnitOfWork):
    config = ConfigDict(
        connector=S3ConnectorConfig(...),
    )
```

Source code in `src/fennflow/connectors/s3/core.py`

```python
class S3Connector(AbstractConnector[S3Extra]):
    """Connector for AWS S3-compatible object storage via aiobotocore.

    Use ``S3ConnectorConfig`` to configure credentials, region, etc.

    Example::

        class UOW(UnitOfWork):
            config = ConfigDict(
                connector=S3ConnectorConfig(...),
            )
    """

    _aio_session: AioSession | None = None

    def __init__(
        self,
        config: S3ConnectorConfig,
    ):
        self._config = config
        self._client: S3Client | None = None

    @property
    def aio_session(self) -> AioSession:
        if self.__class__._aio_session is None:
            raise RuntimeError("AioSession is not initialized.")

        return self.__class__._aio_session

    async def open(
        self,
    ) -> Self:
        if self.__class__._aio_session is None:
            self.__class__._aio_session = get_session()

        self._client = await S3Client(
            config=self._config,
            session=self.aio_session,
        ).open()
        return self

    async def close(
        self,
    ):
        if self._client:
            await self._client.close()
            self._client = None

    @property
    def s3client(
        self,
    ) -> S3Client:
        if self._client is None:
            raise RuntimeError("S3Connector client is not initialized.")
        return self._client

    async def put(
        self,
        file: BinaryMedia,
        repo_extra: S3Extra,
        connector_extra: ConnectorExtra = OMIT,
    ) -> ConnectorRawResponse[PutObjectOutputTypeDef]:
        extra: dict[str, Any] = {}

        if is_given(connector_extra):
            extra.update(connector_extra)

        bucket_name = repo_extra["namespace"]
        response = await self.s3client.client.put_object(
            Bucket=bucket_name,
            Key=file.storage_path,
            Body=file.data,
            ContentType=file.media_type,
            Metadata=file.get_metadata(),
            **extra,
        )
        logger.debug(f"{file=} uploaded to {bucket_name=}")

        return ConnectorRawResponse(raw_response=response)

    async def get(
        self,
        storage_path: StoragePath,
        repo_extra: S3Extra,
        connector_extra: ConnectorExtra = OMIT,
    ) -> MediaResponse[GetObjectOutputTypeDef]:
        extra: dict[str, Any] = {}

        if is_given(connector_extra):
            extra.update(connector_extra)

        try:
            response = await self.s3client.client.get_object(
                Bucket=repo_extra["namespace"],
                Key=storage_path,
                **extra,
            )
        except ClientError as _client_error:
            if _client_error.response["Error"]["Code"] == "NoSuchKey":
                return MediaResponse()
            raise

        async with response["Body"] as stream:
            content_response = ContentResponse(
                content=ContentFactory.from_bytes(
                    media_type=response["ContentType"],
                    data=await stream.read(),
                    **response.get("Metadata", {}),
                ),
                raw_response=response,
            )

        return MediaResponse.from_content_response(content_response)

    async def delete(
        self,
        storage_path: StoragePath,
        repo_extra: S3Extra,
        connector_extra: ConnectorExtra = OMIT,
    ) -> ConnectorRawResponse[DeleteObjectOutputTypeDef]:
        extra: dict[str, Any] = {}

        if is_given(connector_extra):
            extra.update(connector_extra)

        bucket_name = repo_extra["namespace"]
        response = await self.s3client.client.delete_object(
            Bucket=bucket_name,
            Key=storage_path,
            **extra,
        )
        logger.debug(f"file with {storage_path=} deleted from {bucket_name=}")

        return ConnectorRawResponse(raw_response=response)

    @reraise_with(
        NoSuchKeyException(),
        catch=lambda e: (
            isinstance(e, ClientError) and e.response["Error"]["Code"] == "NoSuchKey"
        ),
    )
    async def copy_object(
        self,
        repo_extra: S3Extra,
        from_storage_path: StoragePath,
        to_storage_path: StoragePath,
        to_namespace: Namespace,
        connector_extra: ConnectorExtra = OMIT,
    ) -> ConnectorRawResponse[CopyObjectOutputTypeDef]:
        extra: dict[str, Any] = {}

        if is_given(connector_extra):
            extra.update(connector_extra)

        bucket_name = repo_extra["namespace"]

        response = await self.s3client.client.copy_object(
            CopySource={"Bucket": bucket_name, "Key": from_storage_path},
            Bucket=to_namespace,
            Key=to_storage_path,
            **extra,
        )
        logger.debug(
            f"file from {bucket_name=} with {from_storage_path=} "
            f"copied to {to_namespace=}"
        )

        return ConnectorRawResponse(raw_response=response)

    async def list_objects(
        self,
        prefix: str,
        repo_extra: RepoExtraType,
        limit: int = 1000,
        continuation_token: Omittable[str] | None = OMIT,
        connector_extra: ConnectorExtra = OMIT,
    ) -> ConnectorListResponse[ListObjectsV2OutputTypeDef]:
        extra: dict[str, Any] = {}

        if is_given(connector_extra):
            extra.update(connector_extra)

        if continuation_token:
            extra["ContinuationToken"] = continuation_token

        response = await self.s3client.client.list_objects_v2(
            Bucket=repo_extra["namespace"],
            Prefix=prefix,
            MaxKeys=limit,
            **extra,
        )

        return ConnectorListResponse(
            storage_paths=tuple(obj["Key"] for obj in response.get("Contents", [])),
            continuation_token=response.get("NextContinuationToken"),
            raw_response=response,
        )

    async def generate_presigned_url(
        self,
        storage_path: str,
        repo_extra: RepoExtraType,
        expires_in: Omittable[int] = OMIT,
        connector_extra: ConnectorExtra = OMIT,
    ) -> str:
        params: dict[str, Any] = {
            "ClientMethod": "get_object",
            "Params": {"Bucket": repo_extra["namespace"], "Key": storage_path},
        }

        if is_given(expires_in):
            params["ExpiresIn"] = expires_in

        if is_given(connector_extra):
            params.update(connector_extra)

        url = await self.s3client.client.generate_presigned_url(**params)

        return url
```

## `S3ConnectorConfig`

Bases: `AbstractConnectorConfig`

Configuration for the S3 connector.

Credentials can be provided explicitly via this config or through any method supported by the AWS credential chain (environment variables, `~/.aws/credentials`, IAM roles, etc.). See the `AWS documentation <https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html>`\_ for the full list of supported options.

Attributes:

| Name                    | Type        | Description |
| ----------------------- | ----------- | ----------- |
| `aws_access_key_id`     | \`str       | None\`      |
| `aws_secret_access_key` | \`str       | None\`      |
| `endpoint_url`          | \`str       | None\`      |
| `aiobotocore_config`    | \`AioConfig | None\`      |

Example::

```text
# explicit credentials
class UOW(UnitOfWork):
    user_files = UserFiles
    config = ConfigDict(
                connector=S3ConnectorConfig(
                        aws_access_key_id="key",
                        aws_secret_access_key="secret",
                        )
                        )


# rely on AWS credential chain
S3ConnectorConfig()
```

Source code in `src/fennflow/connectors/s3/config.py`

```python
class S3ConnectorConfig(AbstractConnectorConfig):
    """Configuration for the S3 connector.

    Credentials can be provided explicitly via this config or through any method
    supported by the AWS credential chain
    (environment variables, ``~/.aws/credentials``, IAM roles, etc.).
    See the `AWS documentation
    <https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html>`_
    for the full list of supported options.

    Attributes:
        aws_access_key_id: AWS access key ID.
        aws_secret_access_key: AWS secret access key.
        endpoint_url: Custom endpoint URL for S3-compatible storage.
        aiobotocore_config: Advanced aiobotocore client configuration.

    Example::

        # explicit credentials
        class UOW(UnitOfWork):
            user_files = UserFiles
            config = ConfigDict(
                        connector=S3ConnectorConfig(
                                aws_access_key_id="key",
                                aws_secret_access_key="secret",
                                )
                                )


        # rely on AWS credential chain
        S3ConnectorConfig()
    """

    model_config = AbstractConnectorConfig.model_config | ConfigDict(
        arbitrary_types_allowed=True
    )
    aws_access_key_id: str | None = None
    aws_secret_access_key: str | None = None
    endpoint_url: str | None = None
    aiobotocore_config: AioConfig | None = None
```

## `CreateRepository`

Bases: `AtRepository`, `ValidateDuplicatesMixin`

Repository for uploading (creating) files in the storage.

This repository implements the "create" operation, which uploads new files to the configured storage (e.g. S3) within the current Unit of Work.

**Behavior**:

- Each file is registered in the backend as a pending operation
- Files are uploaded via the connector
- Backend commit is executed on uow.commit

Source code in `src/fennflow/repositories/create.py`

```python
class CreateRepository(
    AtRepository,
    ValidateDuplicatesMixin,
):
    """Repository for uploading (creating) files in the storage.

    This repository implements the "create" operation, which uploads new files
    to the configured storage (e.g. S3) within the current Unit of Work.

    **Behavior**:

    - Each file is registered in the backend as a pending operation
    - Files are uploaded via the connector
    - Backend commit is executed on uow.commit

    """

    async def create(
        self,
        *files: BinaryMedia,
        connector_extra: ConnectorExtra = OMIT,
    ) -> list[ConnectorRawResponse[Any]]:
        """Puts file if it doesn't exist in the backend.

        **Example**::

            file1 = TextContent.from_content("This is the first file.")
            await uow.user_files.at("user1/").create(file1)

        Raises:
            RecordAlreadyExistsException:
                If a file with the same path already exists in a backend
            FilepathsCollisionError:
                If files with the same filepath are passed
        """
        self.validate_duplicates_from_files(files)
        tasks = []
        operations = []
        for file in files:
            file._storage_prefix = self.cwd

            record = await self._uow._backend.backend_engine.execute(
                GetVisibleQuerySpec(
                    scope=self._uow._resolved_config.backend.scope,
                    namespace=self.repo_extra["namespace"],
                    storage_path=file.storage_path,
                    session_id=self._uow._session_id,
                )
            )

            if record:
                raise RecordAlreadyExistsException(
                    storage_path=record.storage_path,
                )

            operation = OperationRecord.from_uow(
                uow=self._uow,
                operation_type=OperationTypeEnum.CREATE,
                storage_path=file.storage_path,
                context=CreateContext(file=file),
                repo_extra=self.repo_extra,
            )

            await self._uow.backend.insert(
                operation,
                on_conflict=OnConflictDoEnum.REPLACE,
            )
            tasks.append(
                self._uow._operation_executor.execute(
                    operation,
                    connector_extra=connector_extra,
                ),
            )
            operations.append(operation)

        await self._uow.backend.flush(operations=operations)
        return await asyncio.gather(*tasks)
```

### `create(*files, connector_extra=OMIT)`

Puts file if it doesn't exist in the backend.

**Example**::

```text
file1 = TextContent.from_content("This is the first file.")
await uow.user_files.at("user1/").create(file1)
```

Raises:

| Type                           | Description                                              |
| ------------------------------ | -------------------------------------------------------- |
| `RecordAlreadyExistsException` | If a file with the same path already exists in a backend |
| `FilepathsCollisionError`      | If files with the same filepath are passed               |

Source code in `src/fennflow/repositories/create.py`

```python
async def create(
    self,
    *files: BinaryMedia,
    connector_extra: ConnectorExtra = OMIT,
) -> list[ConnectorRawResponse[Any]]:
    """Puts file if it doesn't exist in the backend.

    **Example**::

        file1 = TextContent.from_content("This is the first file.")
        await uow.user_files.at("user1/").create(file1)

    Raises:
        RecordAlreadyExistsException:
            If a file with the same path already exists in a backend
        FilepathsCollisionError:
            If files with the same filepath are passed
    """
    self.validate_duplicates_from_files(files)
    tasks = []
    operations = []
    for file in files:
        file._storage_prefix = self.cwd

        record = await self._uow._backend.backend_engine.execute(
            GetVisibleQuerySpec(
                scope=self._uow._resolved_config.backend.scope,
                namespace=self.repo_extra["namespace"],
                storage_path=file.storage_path,
                session_id=self._uow._session_id,
            )
        )

        if record:
            raise RecordAlreadyExistsException(
                storage_path=record.storage_path,
            )

        operation = OperationRecord.from_uow(
            uow=self._uow,
            operation_type=OperationTypeEnum.CREATE,
            storage_path=file.storage_path,
            context=CreateContext(file=file),
            repo_extra=self.repo_extra,
        )

        await self._uow.backend.insert(
            operation,
            on_conflict=OnConflictDoEnum.REPLACE,
        )
        tasks.append(
            self._uow._operation_executor.execute(
                operation,
                connector_extra=connector_extra,
            ),
        )
        operations.append(operation)

    await self._uow.backend.flush(operations=operations)
    return await asyncio.gather(*tasks)
```

## `DeletePrefixRepository`

Bases: `AtRepository`

Repository mixin for deleting files from storage by prefix.

Implements Saga-based deletion with automatic compensation on failure.

Source code in `src/fennflow/repositories/delete_prefix.py`

```python
class DeletePrefixRepository(AtRepository):
    """Repository mixin for deleting files from storage by prefix.

    Implements Saga-based deletion with automatic compensation on failure.
    """

    async def delete_prefix(
        self,
        prefix: str,
        *,
        connector_extra: ConnectorExtra = OMIT,
    ) -> DeleteResponse:
        """Delete files from storage by prefix.

        Args:
            prefix: Prefix files relative to the current directory.
            connector_extra: Additional kwargs forwarded to the connector.

        Returns:
            DeleteResponse containing a result per path in the same order
            as it was listed.
            Each element is a ConnectorRawResponse.

        **Example**::

            async with UOW() as uow:
                await uow.user_files.delete_prefix("") # deletes all files in the bucket
        """
        storage_paths: list[StoragePath] = await paths.get_all_paths(
            repository=self,
            prefix=prefix,
        )

        delete_repository = DeleteRepository(
            uow=self._uow,
            path="",
            repo_extra=self.repo_extra,
        )
        return await delete_repository.delete(
            *storage_paths,
            connector_extra=connector_extra,
        )
```

### `delete_prefix(prefix, *, connector_extra=OMIT)`

Delete files from storage by prefix.

Parameters:

| Name              | Type             | Description                                     | Default    |
| ----------------- | ---------------- | ----------------------------------------------- | ---------- |
| `prefix`          | `str`            | Prefix files relative to the current directory. | *required* |
| `connector_extra` | `ConnectorExtra` | Additional kwargs forwarded to the connector.   | `OMIT`     |

Returns:

| Type             | Description                                                   |
| ---------------- | ------------------------------------------------------------- |
| `DeleteResponse` | DeleteResponse containing a result per path in the same order |
| `DeleteResponse` | as it was listed.                                             |
| `DeleteResponse` | Each element is a ConnectorRawResponse.                       |

**Example**::

```text
async with UOW() as uow:
    await uow.user_files.delete_prefix("") # deletes all files in the bucket
```

Source code in `src/fennflow/repositories/delete_prefix.py`

```python
async def delete_prefix(
    self,
    prefix: str,
    *,
    connector_extra: ConnectorExtra = OMIT,
) -> DeleteResponse:
    """Delete files from storage by prefix.

    Args:
        prefix: Prefix files relative to the current directory.
        connector_extra: Additional kwargs forwarded to the connector.

    Returns:
        DeleteResponse containing a result per path in the same order
        as it was listed.
        Each element is a ConnectorRawResponse.

    **Example**::

        async with UOW() as uow:
            await uow.user_files.delete_prefix("") # deletes all files in the bucket
    """
    storage_paths: list[StoragePath] = await paths.get_all_paths(
        repository=self,
        prefix=prefix,
    )

    delete_repository = DeleteRepository(
        uow=self._uow,
        path="",
        repo_extra=self.repo_extra,
    )
    return await delete_repository.delete(
        *storage_paths,
        connector_extra=connector_extra,
    )
```

## `DeleteRepository`

Bases: `AtRepository`

Repository mixin for deleting files from storage.

Implements Saga-based deletion with automatic compensation on failure.

Source code in `src/fennflow/repositories/delete.py`

```python
class DeleteRepository(AtRepository):
    """Repository mixin for deleting files from storage.

    Implements Saga-based deletion with automatic compensation on failure.
    """

    async def delete(
        self,
        *paths: str,
        connector_extra: ConnectorExtra = OMIT,
    ) -> DeleteResponse:
        """Delete files from storage.

        Args:
            paths: Paths to the files relative to the current directory.
            connector_extra: Additional kwargs forwarded to the connector.

        Returns:
            DeleteResponse containing a result per path in the same order as input.
            Each element is a ConnectorRawResponse if the file was deleted,
            or None if the path did not exist in the backend.

        """
        tasks = []
        task_indexes = []
        operations = []
        results: DeleteResponse = [None] * len(paths)

        for task_index, path in enumerate(paths):
            storage_path = self._join_path(path)
            record = await visible_record.get_visible_record(
                repostory=self,
                storage_path=storage_path,
            )

            if record is None:
                continue

            operation = self.__create_operation(record=record)
            await self.__insert_into_buffer(operation=operation)

            tasks.append(self.__construct_executor_task(operation, connector_extra))
            task_indexes.append(task_index)
            operations.append(operation)

        await self._uow.backend.flush(operations=operations)
        return await gather_tasks.gather_tasks(
            tasks=tasks,
            task_indexes=task_indexes,
            results=results,
        )

    def __get_context(self, record: Record) -> DeleteContext:
        return DeleteContext(
            to_storage_path=record.generate_tmp_path(),
            to_namespace=self.repo_extra["namespace"],
        )

    def __create_operation(self, record: Record) -> OperationRecord:
        return OperationRecord.from_uow(
            uow=self._uow,
            operation_type=OperationTypeEnum.DELETE,
            storage_path=record.storage_path,
            context=self.__get_context(record=record),
            repo_extra=self.repo_extra,
        )

    def __construct_executor_task(
        self,
        operation: OperationRecord,
        connector_extra: ConnectorExtra = OMIT,
    ) -> Coroutine[Any, Any, Any]:
        return self._uow._operation_executor.execute(
            operation,
            connector_extra=connector_extra,
        )

    async def __insert_into_buffer(self, operation: OperationRecord) -> None:
        await self._uow.backend.insert(
            operation,
            on_conflict=OnConflictDoEnum.REPLACE,
        )
```

### `delete(*paths, connector_extra=OMIT)`

Delete files from storage.

Parameters:

| Name              | Type             | Description                                           | Default |
| ----------------- | ---------------- | ----------------------------------------------------- | ------- |
| `paths`           | `str`            | Paths to the files relative to the current directory. | `()`    |
| `connector_extra` | `ConnectorExtra` | Additional kwargs forwarded to the connector.         | `OMIT`  |

Returns:

| Type             | Description                                                             |
| ---------------- | ----------------------------------------------------------------------- |
| `DeleteResponse` | DeleteResponse containing a result per path in the same order as input. |
| `DeleteResponse` | Each element is a ConnectorRawResponse if the file was deleted,         |
| `DeleteResponse` | or None if the path did not exist in the backend.                       |

Source code in `src/fennflow/repositories/delete.py`

```python
async def delete(
    self,
    *paths: str,
    connector_extra: ConnectorExtra = OMIT,
) -> DeleteResponse:
    """Delete files from storage.

    Args:
        paths: Paths to the files relative to the current directory.
        connector_extra: Additional kwargs forwarded to the connector.

    Returns:
        DeleteResponse containing a result per path in the same order as input.
        Each element is a ConnectorRawResponse if the file was deleted,
        or None if the path did not exist in the backend.

    """
    tasks = []
    task_indexes = []
    operations = []
    results: DeleteResponse = [None] * len(paths)

    for task_index, path in enumerate(paths):
        storage_path = self._join_path(path)
        record = await visible_record.get_visible_record(
            repostory=self,
            storage_path=storage_path,
        )

        if record is None:
            continue

        operation = self.__create_operation(record=record)
        await self.__insert_into_buffer(operation=operation)

        tasks.append(self.__construct_executor_task(operation, connector_extra))
        task_indexes.append(task_index)
        operations.append(operation)

    await self._uow.backend.flush(operations=operations)
    return await gather_tasks.gather_tasks(
        tasks=tasks,
        task_indexes=task_indexes,
        results=results,
    )
```

## `GeneratePresignedUrlPrefixRepository`

Bases: `AtRepository`

Repository mixin for generating presigned urls by prefix.

Combines ListRepository and GeneratePresignedUrlRepository.

Source code in `src/fennflow/repositories/generate_presigned_url_prefix.py`

```python
class GeneratePresignedUrlPrefixRepository(AtRepository):
    """Repository mixin for generating presigned urls by prefix.

    Combines ListRepository and GeneratePresignedUrlRepository.
    """

    async def generate_presigned_url_prefix(
        self,
        prefix: str,
        *,
        expires_in: Omittable[int] = OMIT,
        connector_extra: ConnectorExtra = OMIT,
    ) -> PresignedUrlResponse:
        """Generate a presigned URLs for the given prefix.

        Does not interact with the backend or file storage.
        No Saga compensation is applied.

        Args:
            prefix: Prefix files relative to the current directory.
            expires_in: Expiry duration in seconds. When omitted, the
                connector's default is used.
            connector_extra: Additional kwargs forwarded to the connector.

        Returns:
            PresignedUrlResponse containing the generated URLs.

        Raises:
            ConnectorCapabilityException: If the configured connector
                does not support presigned URL generation.

        **Example**::

            async with UOW() as uow:
                response = await uow.files.generate_presigned_url("user1/"
                    expires_in=600,
                )
                print(response.results) # list[str | None]
                print(list(response.urls)) # list[str]
        """
        storage_paths: list[StoragePath] = await paths.get_all_paths(
            repository=self,
            prefix=prefix,
        )

        generate_presigned_url_repo = GeneratePresignedUrlRepository(
            uow=self._uow,
            path="",
            repo_extra=self.repo_extra,
        )
        return await generate_presigned_url_repo.generate_presigned_url(
            *storage_paths,
            expires_in=expires_in,
            connector_extra=connector_extra,
        )
```

### `generate_presigned_url_prefix(prefix, *, expires_in=OMIT, connector_extra=OMIT)`

Generate a presigned URLs for the given prefix.

Does not interact with the backend or file storage. No Saga compensation is applied.

Parameters:

| Name              | Type             | Description                                                                | Default    |
| ----------------- | ---------------- | -------------------------------------------------------------------------- | ---------- |
| `prefix`          | `str`            | Prefix files relative to the current directory.                            | *required* |
| `expires_in`      | `Omittable[int]` | Expiry duration in seconds. When omitted, the connector's default is used. | `OMIT`     |
| `connector_extra` | `ConnectorExtra` | Additional kwargs forwarded to the connector.                              | `OMIT`     |

Returns:

| Type                   | Description                                         |
| ---------------------- | --------------------------------------------------- |
| `PresignedUrlResponse` | PresignedUrlResponse containing the generated URLs. |

Raises:

| Type                           | Description                                                            |
| ------------------------------ | ---------------------------------------------------------------------- |
| `ConnectorCapabilityException` | If the configured connector does not support presigned URL generation. |

**Example**::

```text
async with UOW() as uow:
    response = await uow.files.generate_presigned_url("user1/"
        expires_in=600,
    )
    print(response.results) # list[str | None]
    print(list(response.urls)) # list[str]
```

Source code in `src/fennflow/repositories/generate_presigned_url_prefix.py`

```python
async def generate_presigned_url_prefix(
    self,
    prefix: str,
    *,
    expires_in: Omittable[int] = OMIT,
    connector_extra: ConnectorExtra = OMIT,
) -> PresignedUrlResponse:
    """Generate a presigned URLs for the given prefix.

    Does not interact with the backend or file storage.
    No Saga compensation is applied.

    Args:
        prefix: Prefix files relative to the current directory.
        expires_in: Expiry duration in seconds. When omitted, the
            connector's default is used.
        connector_extra: Additional kwargs forwarded to the connector.

    Returns:
        PresignedUrlResponse containing the generated URLs.

    Raises:
        ConnectorCapabilityException: If the configured connector
            does not support presigned URL generation.

    **Example**::

        async with UOW() as uow:
            response = await uow.files.generate_presigned_url("user1/"
                expires_in=600,
            )
            print(response.results) # list[str | None]
            print(list(response.urls)) # list[str]
    """
    storage_paths: list[StoragePath] = await paths.get_all_paths(
        repository=self,
        prefix=prefix,
    )

    generate_presigned_url_repo = GeneratePresignedUrlRepository(
        uow=self._uow,
        path="",
        repo_extra=self.repo_extra,
    )
    return await generate_presigned_url_repo.generate_presigned_url(
        *storage_paths,
        expires_in=expires_in,
        connector_extra=connector_extra,
    )
```

## `GeneratePresignedUrlRepository`

Bases: `AtRepository`

Repository mixin for generating presigned URLs.

Provides access to connector-level presigned URL generation without touching the backend or file storage.

Note

This mixin is only supported by connectors that implement presigned URL generation (e.g. S3Connector). Using it with InMemoryConnector or LocalConnector will raise ConnectorCapabilityException.

Source code in `src/fennflow/repositories/generate_presigned_url.py`

```python
class GeneratePresignedUrlRepository(AtRepository):
    """Repository mixin for generating presigned URLs.

    Provides access to connector-level presigned URL generation
    without touching the backend or file storage.

    Note:
        This mixin is only supported by connectors that implement
        presigned URL generation (e.g. S3Connector).
        Using it with InMemoryConnector or LocalConnector will raise
        ConnectorCapabilityException.
    """

    async def generate_presigned_url(
        self,
        *storage_paths: str,
        expires_in: Omittable[int] = OMIT,
        connector_extra: Omittable[Any] = OMIT,
    ) -> PresignedUrlResponse:
        """Generate a presigned URL for the given path.

        Does not interact with the backend or file storage.
        No Saga compensation is applied.

        Args:
            storage_paths: Paths to the files relative to the current directory.
            expires_in: Expiry duration in seconds. When omitted, the
                connector's default is used.
            connector_extra: Additional connector-specific parameters
                passed directly to the connector.

        Returns:
            PresignedUrlResponse containing the generated URLs.

        Raises:
            ConnectorCapabilityException: If the configured connector
                does not support presigned URL generation.

        **Example**::

            async with UOW() as uow:
                response = await uow.files.at("user1/").generate_presigned_url(
                    "report.pdf",
                    "NonExistingFile",
                    expires_in=600,
                )
                print(response.results) # [<Link>, None]
                print(list(response.urls)) # [<Link>]
        """
        tasks = []
        task_indexes = []
        results: list[None | str] = [None] * len(storage_paths)

        for task_idx, relative_path in enumerate(storage_paths):
            storage_path = self._join_path(relative_path)

            record = await visible_record.get_visible_record(
                repostory=self,
                storage_path=storage_path,
            )

            if record is None:
                continue

            tasks.append(
                self._uow.connector.generate_presigned_url(
                    storage_path=storage_path,
                    expires_in=expires_in,
                    repo_extra=self.repo_extra,
                    connector_extra=connector_extra,
                )
            )
            task_indexes.append(task_idx)

        return await self.__construct_response(
            tasks=tasks,
            task_indexes=task_indexes,
            results=results,
        )

    async def __construct_response(
        self,
        tasks: list[Coroutine[Any, Any, str]],
        task_indexes: list[int],
        results: list[None | str],
    ) -> PresignedUrlResponse:
        results = await gather_tasks.gather_tasks(
            tasks=tasks,
            task_indexes=task_indexes,
            results=results,
        )
        return PresignedUrlResponse(results=results)
```

### `generate_presigned_url(*storage_paths, expires_in=OMIT, connector_extra=OMIT)`

Generate a presigned URL for the given path.

Does not interact with the backend or file storage. No Saga compensation is applied.

Parameters:

| Name              | Type             | Description                                                                | Default |
| ----------------- | ---------------- | -------------------------------------------------------------------------- | ------- |
| `storage_paths`   | `str`            | Paths to the files relative to the current directory.                      | `()`    |
| `expires_in`      | `Omittable[int]` | Expiry duration in seconds. When omitted, the connector's default is used. | `OMIT`  |
| `connector_extra` | `Omittable[Any]` | Additional connector-specific parameters passed directly to the connector. | `OMIT`  |

Returns:

| Type                   | Description                                         |
| ---------------------- | --------------------------------------------------- |
| `PresignedUrlResponse` | PresignedUrlResponse containing the generated URLs. |

Raises:

| Type                           | Description                                                            |
| ------------------------------ | ---------------------------------------------------------------------- |
| `ConnectorCapabilityException` | If the configured connector does not support presigned URL generation. |

**Example**::

```text
async with UOW() as uow:
    response = await uow.files.at("user1/").generate_presigned_url(
        "report.pdf",
        "NonExistingFile",
        expires_in=600,
    )
    print(response.results) # [<Link>, None]
    print(list(response.urls)) # [<Link>]
```

Source code in `src/fennflow/repositories/generate_presigned_url.py`

```python
async def generate_presigned_url(
    self,
    *storage_paths: str,
    expires_in: Omittable[int] = OMIT,
    connector_extra: Omittable[Any] = OMIT,
) -> PresignedUrlResponse:
    """Generate a presigned URL for the given path.

    Does not interact with the backend or file storage.
    No Saga compensation is applied.

    Args:
        storage_paths: Paths to the files relative to the current directory.
        expires_in: Expiry duration in seconds. When omitted, the
            connector's default is used.
        connector_extra: Additional connector-specific parameters
            passed directly to the connector.

    Returns:
        PresignedUrlResponse containing the generated URLs.

    Raises:
        ConnectorCapabilityException: If the configured connector
            does not support presigned URL generation.

    **Example**::

        async with UOW() as uow:
            response = await uow.files.at("user1/").generate_presigned_url(
                "report.pdf",
                "NonExistingFile",
                expires_in=600,
            )
            print(response.results) # [<Link>, None]
            print(list(response.urls)) # [<Link>]
    """
    tasks = []
    task_indexes = []
    results: list[None | str] = [None] * len(storage_paths)

    for task_idx, relative_path in enumerate(storage_paths):
        storage_path = self._join_path(relative_path)

        record = await visible_record.get_visible_record(
            repostory=self,
            storage_path=storage_path,
        )

        if record is None:
            continue

        tasks.append(
            self._uow.connector.generate_presigned_url(
                storage_path=storage_path,
                expires_in=expires_in,
                repo_extra=self.repo_extra,
                connector_extra=connector_extra,
            )
        )
        task_indexes.append(task_idx)

    return await self.__construct_response(
        tasks=tasks,
        task_indexes=task_indexes,
        results=results,
    )
```

## `GetPrefixRepository`

Bases: `AtRepository`

Repository mixin for getting files from storage by prefix.

Combines ListRepository and GetRepository.

Source code in `src/fennflow/repositories/get_prefix.py`

```python
class GetPrefixRepository(AtRepository):
    """Repository mixin for getting files from storage by prefix.

    Combines ListRepository and GetRepository.
    """

    async def get_prefix(
        self,
        prefix: str,
        *,
        connector_extra: ConnectorExtra = OMIT,
    ) -> MediaResponse[Any]:
        """Get files from storage by prefix.

        Args:
            prefix: Prefix files relative to the current directory.
            connector_extra: Additional kwargs forwarded to the connector.

        Returns:
            MediaResponse

        **Example**::

            async with UOW() as uow:
                await uow.user_files.get_prefix("some_prefix")
        """
        storage_paths: list[StoragePath] = await paths.get_all_paths(
            repository=self,
            prefix=prefix,
        )

        get_repository = GetRepository(
            uow=self._uow,
            path="",
            repo_extra=self.repo_extra,
        )
        return await get_repository.get(*storage_paths, connector_extra=connector_extra)
```

### `get_prefix(prefix, *, connector_extra=OMIT)`

Get files from storage by prefix.

Parameters:

| Name              | Type             | Description                                     | Default    |
| ----------------- | ---------------- | ----------------------------------------------- | ---------- |
| `prefix`          | `str`            | Prefix files relative to the current directory. | *required* |
| `connector_extra` | `ConnectorExtra` | Additional kwargs forwarded to the connector.   | `OMIT`     |

Returns:

| Type                 | Description   |
| -------------------- | ------------- |
| `MediaResponse[Any]` | MediaResponse |

**Example**::

```text
async with UOW() as uow:
    await uow.user_files.get_prefix("some_prefix")
```

Source code in `src/fennflow/repositories/get_prefix.py`

```python
async def get_prefix(
    self,
    prefix: str,
    *,
    connector_extra: ConnectorExtra = OMIT,
) -> MediaResponse[Any]:
    """Get files from storage by prefix.

    Args:
        prefix: Prefix files relative to the current directory.
        connector_extra: Additional kwargs forwarded to the connector.

    Returns:
        MediaResponse

    **Example**::

        async with UOW() as uow:
            await uow.user_files.get_prefix("some_prefix")
    """
    storage_paths: list[StoragePath] = await paths.get_all_paths(
        repository=self,
        prefix=prefix,
    )

    get_repository = GetRepository(
        uow=self._uow,
        path="",
        repo_extra=self.repo_extra,
    )
    return await get_repository.get(*storage_paths, connector_extra=connector_extra)
```

## `GetRepository`

Bases: `AtRepository`

Repository for retrieving a file from storage within the current scope.

This method returns a `MediaResponse` object containing the requested file, if it exists according to the backend (source of truth).

**Behavior**:

- The backend is treated as the source of truth
- If the file is not present in the backend, the storage is NOT queried
- If the file exists in the backend, it is fetched from the storage via the connector

**Notes**:

- This method is read-only and does not participate in transaction flows (no saga)
- No network request is made if the backend does not contain the file
- Storage and backend may become inconsistent (e.g. after restart with InMemoryBackend); in such cases, use a reconcile mechanism to resync state. ()

Source code in `src/fennflow/repositories/get.py`

```python
class GetRepository(AtRepository):
    """Repository for retrieving a file from storage within the current scope.

    This method returns a `MediaResponse` object containing the requested file,
    if it exists according to the backend (source of truth).

    **Behavior**:

    - The backend is treated as the source of truth
    - If the file is not present in the backend, the storage is NOT queried
    - If the file exists in the backend,
    it is fetched from the storage via the connector

    **Notes**:

    - This method is read-only and does not participate in transaction flows (no saga)
    - No network request is made if the backend does not contain the file
    - Storage and backend may become inconsistent
      (e.g. after restart with InMemoryBackend);
      in such cases, use a reconcile mechanism to resync state.
      (<add_link_here on wiki>)

    """

    async def get(
        self,
        *paths: str,
        connector_extra: ConnectorExtra = OMIT,
    ) -> MediaResponse[Any]:
        """Retrieve a file from storage within the current scope.

        Args:
            *paths (str):
                Relative file's paths within the scoped repository

            connector_extra:
                Additional connector-specific parameters passed directly
                to the connector (e.g. S3 `get_object` arguments)

        Returns:
            MediaResponse:
                - `MediaResponse(media=(...))` if the file exists
                - `MediaResponse()` (empty) if the file does not exist

        **Example**::

            response = await uow.user_files.at("user1/").get("file.txt")
            if response:
                file = response[0]

        """
        tasks = []
        for path in paths:
            storage_path = self._join_path(path)
            operation = await self._uow._backend.backend_engine.execute(
                GetVisibleQuerySpec(
                    scope=self._uow._resolved_config.backend.scope,
                    namespace=self.repo_extra["namespace"],
                    storage_path=storage_path,
                    session_id=self._uow._session_id,
                )
            )

            if operation:
                tasks.append(
                    self._uow.connector.get(
                        storage_path=storage_path,
                        repo_extra=self.repo_extra,
                        connector_extra=connector_extra,
                    )
                )

        if tasks:
            results = await asyncio.gather(*tasks)
            return MediaResponse.join(results)

        return MediaResponse()
```

### `get(*paths, connector_extra=OMIT)`

Retrieve a file from storage within the current scope.

Parameters:

| Name              | Type             | Description                                                                                              | Default |
| ----------------- | ---------------- | -------------------------------------------------------------------------------------------------------- | ------- |
| `*paths`          | `str`            | Relative file's paths within the scoped repository                                                       | `()`    |
| `connector_extra` | `ConnectorExtra` | Additional connector-specific parameters passed directly to the connector (e.g. S3 get_object arguments) | `OMIT`  |

Returns:

| Name            | Type                 | Description                                                                                      |
| --------------- | -------------------- | ------------------------------------------------------------------------------------------------ |
| `MediaResponse` | `MediaResponse[Any]` | MediaResponse(media=(...)) if the file exists MediaResponse() (empty) if the file does not exist |

**Example**::

```text
response = await uow.user_files.at("user1/").get("file.txt")
if response:
    file = response[0]
```

Source code in `src/fennflow/repositories/get.py`

```python
async def get(
    self,
    *paths: str,
    connector_extra: ConnectorExtra = OMIT,
) -> MediaResponse[Any]:
    """Retrieve a file from storage within the current scope.

    Args:
        *paths (str):
            Relative file's paths within the scoped repository

        connector_extra:
            Additional connector-specific parameters passed directly
            to the connector (e.g. S3 `get_object` arguments)

    Returns:
        MediaResponse:
            - `MediaResponse(media=(...))` if the file exists
            - `MediaResponse()` (empty) if the file does not exist

    **Example**::

        response = await uow.user_files.at("user1/").get("file.txt")
        if response:
            file = response[0]

    """
    tasks = []
    for path in paths:
        storage_path = self._join_path(path)
        operation = await self._uow._backend.backend_engine.execute(
            GetVisibleQuerySpec(
                scope=self._uow._resolved_config.backend.scope,
                namespace=self.repo_extra["namespace"],
                storage_path=storage_path,
                session_id=self._uow._session_id,
            )
        )

        if operation:
            tasks.append(
                self._uow.connector.get(
                    storage_path=storage_path,
                    repo_extra=self.repo_extra,
                    connector_extra=connector_extra,
                )
            )

    if tasks:
        results = await asyncio.gather(*tasks)
        return MediaResponse.join(results)

    return MediaResponse()
```

## `ListRepository`

Bases: `AtRepository`

Repository for retrieving a files from storage within the current scope.

Source code in `src/fennflow/repositories/list.py`

```python
class ListRepository(AtRepository):
    """Repository for retrieving a files from storage within the current scope."""

    async def list(
        self,
        prefix: str = "",
        continuation_token: Omittable[str] = OMIT,
        limit: int = 1000,
    ) -> ListResponse:
        """Uploads files under the current path, optionally filtered by prefix.

        Files are visible if they are uploaded (committed) or pending within the
        current session. Pending files from other sessions are not returned.

        Args:
            prefix: Sub-path to filter results. Appended to the current ``at()``
                path. Defaults to ``""`` (list everything under the current path).
            continuation_token: Opaque token returned by a previous call to
                continue paginating.
            limit: Maximum number of storage_paths to return. Defaults to ``1000``.

        Returns:
            ListResponse: A container of storage_paths matching the query. Includes
                a ``continuation_token`` if more results are available, otherwise
                ``None``.

        **Example**::

             async with UOW() as uow:
                 await uow.files.at("folder1/").put(file1, file2, file3)
                 page = await uow.files.at("folder1/").list(limit=2)

                 next_page = await uow.files.at("folder1/").list(
                     limit=2,
                     continuation_token=page.continuation_token,
                 )
        """
        storage_prefix = self._join_path(prefix)

        record_page = await self._uow.backend.backend_engine.execute(
            SelectVisibleQuerySpec(
                scope=self._uow._resolved_config.backend.scope,
                namespace=self.repo_extra["namespace"],
                prefix=storage_prefix,
                continuation_token=continuation_token,
                limit=limit,
                session_id=self._uow._session_id,
            )
        )

        return ListResponse(
            storage_paths=tuple(record.storage_path for record in record_page),
            continuation_token=record_page.continuation_token,
        )
```

### `list(prefix='', continuation_token=OMIT, limit=1000)`

Uploads files under the current path, optionally filtered by prefix.

Files are visible if they are uploaded (committed) or pending within the current session. Pending files from other sessions are not returned.

Parameters:

| Name                 | Type             | Description                                                                                                             | Default |
| -------------------- | ---------------- | ----------------------------------------------------------------------------------------------------------------------- | ------- |
| `prefix`             | `str`            | Sub-path to filter results. Appended to the current at() path. Defaults to "" (list everything under the current path). | `''`    |
| `continuation_token` | `Omittable[str]` | Opaque token returned by a previous call to continue paginating.                                                        | `OMIT`  |
| `limit`              | `int`            | Maximum number of storage_paths to return. Defaults to 1000.                                                            | `1000`  |

Returns:

| Name           | Type           | Description                                                                                                                   |
| -------------- | -------------- | ----------------------------------------------------------------------------------------------------------------------------- |
| `ListResponse` | `ListResponse` | A container of storage_paths matching the query. Includes a continuation_token if more results are available, otherwise None. |

**Example**::

```text
 async with UOW() as uow:
     await uow.files.at("folder1/").put(file1, file2, file3)
     page = await uow.files.at("folder1/").list(limit=2)

     next_page = await uow.files.at("folder1/").list(
         limit=2,
         continuation_token=page.continuation_token,
     )
```

Source code in `src/fennflow/repositories/list.py`

```python
async def list(
    self,
    prefix: str = "",
    continuation_token: Omittable[str] = OMIT,
    limit: int = 1000,
) -> ListResponse:
    """Uploads files under the current path, optionally filtered by prefix.

    Files are visible if they are uploaded (committed) or pending within the
    current session. Pending files from other sessions are not returned.

    Args:
        prefix: Sub-path to filter results. Appended to the current ``at()``
            path. Defaults to ``""`` (list everything under the current path).
        continuation_token: Opaque token returned by a previous call to
            continue paginating.
        limit: Maximum number of storage_paths to return. Defaults to ``1000``.

    Returns:
        ListResponse: A container of storage_paths matching the query. Includes
            a ``continuation_token`` if more results are available, otherwise
            ``None``.

    **Example**::

         async with UOW() as uow:
             await uow.files.at("folder1/").put(file1, file2, file3)
             page = await uow.files.at("folder1/").list(limit=2)

             next_page = await uow.files.at("folder1/").list(
                 limit=2,
                 continuation_token=page.continuation_token,
             )
    """
    storage_prefix = self._join_path(prefix)

    record_page = await self._uow.backend.backend_engine.execute(
        SelectVisibleQuerySpec(
            scope=self._uow._resolved_config.backend.scope,
            namespace=self.repo_extra["namespace"],
            prefix=storage_prefix,
            continuation_token=continuation_token,
            limit=limit,
            session_id=self._uow._session_id,
        )
    )

    return ListResponse(
        storage_paths=tuple(record.storage_path for record in record_page),
        continuation_token=record_page.continuation_token,
    )
```

## `PutRepository`

Bases: `AtRepository`, `ValidateDuplicatesMixin`

Repository for upserting files in the storage.

This repository implements the "put" operation, which uploads new files to the configured storage (e.g. S3) within the current Unit of Work.

**Behavior**:

- Each file is registered in the backend as a pending operation
- Files are uploaded via the connector
- Backend commit is executed on uow.commit

Source code in `src/fennflow/repositories/put.py`

```python
class PutRepository(AtRepository, ValidateDuplicatesMixin):
    """Repository for upserting files in the storage.

    This repository implements the "put" operation, which uploads new files
    to the configured storage (e.g. S3) within the current Unit of Work.

    **Behavior**:

    - Each file is registered in the backend as a pending operation
    - Files are uploaded via the connector
    - Backend commit is executed on uow.commit

    """

    async def put(
        self,
        *files: BinaryMedia,
        connector_extra: ConnectorExtra = OMIT,
    ) -> list[ConnectorRawResponse[Any]]:
        """Puts file into storage.

        **Example**::

            file1 = TextContent.from_content("This is the first file.")
            await uow.user_files.at("user1/").put(file1)

        Raises:
            FilepathsCollisionError:
                If files with the same filepath are passed


        """
        self.validate_duplicates_from_files(files)
        tasks = []
        operations = []
        for file in files:
            file._storage_prefix = self.cwd

            operation = await self._uow.backend.get(
                scope=self._uow._resolved_config.backend.scope,
                namespace=self.repo_extra["namespace"],
                storage_path=file.storage_path,
            )

            operation = OperationRecord.from_uow(
                uow=self._uow,
                operation_type=OperationTypeEnum.PUT,
                storage_path=file.storage_path,
                context=self.__get_context(operation, file),
                repo_extra=self.repo_extra,
            )

            await self._uow.backend.insert(
                operation,
                on_conflict=OnConflictDoEnum.REPLACE,
            )
            tasks.append(
                self._uow._operation_executor.execute(
                    operation,
                    connector_extra=connector_extra,
                ),
            )
            operations.append(operation)
        await self._uow.backend.flush(operations=operations)
        return await asyncio.gather(*tasks)

    def __get_context(
        self,
        operation: OperationRecord[PutContext] | None,
        file: BinaryMedia,
    ) -> PutContext:
        if (
            operation
            and operation.record.is_pending
            and operation.record.session_id == self._uow._session_id
            and operation.record.is_put_type
        ):
            tmp_path = operation.require_context().tmp_path
            ctx = PutContext(
                file=file,
                tmp_path=tmp_path,
            )
        else:
            ctx = PutContext(
                file=file,
            )

        return ctx
```

### `put(*files, connector_extra=OMIT)`

Puts file into storage.

**Example**::

```text
file1 = TextContent.from_content("This is the first file.")
await uow.user_files.at("user1/").put(file1)
```

Raises:

| Type                      | Description                                |
| ------------------------- | ------------------------------------------ |
| `FilepathsCollisionError` | If files with the same filepath are passed |

Source code in `src/fennflow/repositories/put.py`

```python
async def put(
    self,
    *files: BinaryMedia,
    connector_extra: ConnectorExtra = OMIT,
) -> list[ConnectorRawResponse[Any]]:
    """Puts file into storage.

    **Example**::

        file1 = TextContent.from_content("This is the first file.")
        await uow.user_files.at("user1/").put(file1)

    Raises:
        FilepathsCollisionError:
            If files with the same filepath are passed


    """
    self.validate_duplicates_from_files(files)
    tasks = []
    operations = []
    for file in files:
        file._storage_prefix = self.cwd

        operation = await self._uow.backend.get(
            scope=self._uow._resolved_config.backend.scope,
            namespace=self.repo_extra["namespace"],
            storage_path=file.storage_path,
        )

        operation = OperationRecord.from_uow(
            uow=self._uow,
            operation_type=OperationTypeEnum.PUT,
            storage_path=file.storage_path,
            context=self.__get_context(operation, file),
            repo_extra=self.repo_extra,
        )

        await self._uow.backend.insert(
            operation,
            on_conflict=OnConflictDoEnum.REPLACE,
        )
        tasks.append(
            self._uow._operation_executor.execute(
                operation,
                connector_extra=connector_extra,
            ),
        )
        operations.append(operation)
    await self._uow.backend.flush(operations=operations)
    return await asyncio.gather(*tasks)
```

## `RepoField`

Bases: `Generic[RepoType]`

A descriptor that lazily initializes a repository instance on a UnitOfWork.

Parameters:

| Name        | Type             | Description                                                      | Default    |
| ----------- | ---------------- | ---------------------------------------------------------------- | ---------- |
| `repo_cls`  | `type[RepoType]` | The repository class to instantiate.                             | *required* |
| `namespace` | `Namespace`      | The storage namespace (e.g. S3 bucket name) for this repository. | *required* |

Example::

```text
class UOW(UnitOfWork):
    user_files = RepoField(UserFiles, namespace="user-files")
```

Source code in `src/fennflow/repositories/fields/base.py`

```python
class RepoField(Generic[RepoType]):
    """A descriptor that lazily initializes a repository instance on a UnitOfWork.

    Args:
        repo_cls: The repository class to instantiate.
        namespace: The storage namespace (e.g. S3 bucket name) for this repository.

    Example::

        class UOW(UnitOfWork):
            user_files = RepoField(UserFiles, namespace="user-files")

    """

    def __init__(
        self,
        repo_cls: type[RepoType],
        *,
        namespace: Namespace,
    ):
        self.repo_cls = repo_cls
        self.repo_extra: RepoExtra = {"namespace": namespace}

    def __set_name__(
        self,
        owner,
        name,
    ):
        self.name = name

    @overload
    def __get__(
        self,
        instance: None,
        owner,
    ) -> RepoField[RepoType]: ...

    @overload
    def __get__(
        self,
        instance,
        owner,
    ) -> RepoType: ...

    def __get__(
        self,
        instance,
        owner,
    ):
        if instance is None:
            return self

        repo = instance.__dict__.get(self.name)
        if repo is None:
            repo = self.repo_cls(
                uow=instance,
                path="/",
                repo_extra=self.repo_extra,
            )
            instance.__dict__[self.name] = repo

        return repo
```

## `S3RepoField(repo_cls, bucket_name)`

Create a RepoField configured for S3 storage.

Parameters:

| Name          | Type             | Description                          | Default    |
| ------------- | ---------------- | ------------------------------------ | ---------- |
| `repo_cls`    | `type[RepoType]` | The repository class to instantiate. | *required* |
| `bucket_name` | `BucketName`     | alias for RepoField.namespace.       | *required* |

Returns:

| Type                  | Description                                                 |
| --------------------- | ----------------------------------------------------------- |
| `RepoField[RepoType]` | A configured RepoField bound to the given repository class. |

**Example**::

```text
class UOW(UnitOfWork):
    user_files = S3RepoField(UserFiles, bucket_name="my-bucket")
```

Source code in `src/fennflow/repositories/fields/s3.py`

```python
def S3RepoField(
    repo_cls: type[RepoType],
    bucket_name: BucketName,
) -> RepoField[RepoType]:
    """Create a RepoField configured for S3 storage.

    Args:
        repo_cls: The repository class to instantiate.
        bucket_name: alias for RepoField.namespace.

    Returns:
        A configured RepoField bound to the given repository class.

    **Example**::

        class UOW(UnitOfWork):
            user_files = S3RepoField(UserFiles, bucket_name="my-bucket")

    """
    return RepoField(
        repo_cls,
        namespace=bucket_name,
    )
```

## `AtRepository`

Bases: `BaseRepository`

Repository mixin that adds path navigation capabilities.

Allows scoping operations to a specific path within the namespace using the `at()` method.

Source code in `src/fennflow/repositories/at.py`

```python
class AtRepository(BaseRepository):
    """Repository mixin that adds path navigation capabilities.

    Allows scoping operations to a specific path within the namespace
    using the ``at()`` method.

    """

    def _join_path(self, *paths: str) -> StoragePath:
        """Join the current path with one or more path segments.

        Args:
            *paths: Path segments to append to the current path.

        Returns:
            The joined and normalized path string.
        """
        return Path.join_path(self._path, *paths)

    def at(self, path: str | PathTemplate) -> Self:
        """Return a new repository instance scoped to the given path.

        Args:
            path: Path to scope the repository to, relative to the current path.

        Returns:
            A new repository instance with the updated path.

        Example::

            storage = uow.user_files.at("user1/")
            print(storage.cwd) # "user1/"

        Example::

            @dataclass(slots=True)
            class YearMonth:
                year: int
                month: int

                def render(self) -> str:
                    return f"{self.year}/{self.month:02d}"

            storage = uow.user_files.at("user1/").at(YearMonth(year=2026, month=1))
            print(storage.cwd) # user1/2026/01/

        """
        if isinstance(path, str):
            resolved = path
        elif isinstance(path, PathTemplate):
            resolved = path.render()
        else:
            raise TypeError(
                f"at() expects a str or PathTemplate, got {type(path).__name__!r}. "
                "Define a render() -> str method on your class "
                "to use it as a path template."
            )

        new_path = self._join_path(resolved)

        return self.__class__(self._uow, new_path, repo_extra=self.repo_extra)

    def _cd_root(self) -> Self:
        """Return a new repository instance scoped to the root path."""
        return self.__class__(
            self._uow,
            "",
            repo_extra=self.repo_extra,
        )

    @property
    def cwd(self) -> str:
        """Return the current working path.

        Returns:
            The normalized current path string.
        """
        return Path.normalize_folder(self._path)
```

### `cwd`

Return the current working path.

Returns:

| Type  | Description                         |
| ----- | ----------------------------------- |
| `str` | The normalized current path string. |

### `at(path)`

Return a new repository instance scoped to the given path.

Parameters:

| Name   | Type  | Description    | Default                                                        |
| ------ | ----- | -------------- | -------------------------------------------------------------- |
| `path` | \`str | PathTemplate\` | Path to scope the repository to, relative to the current path. |

Returns:

| Type   | Description                                      |
| ------ | ------------------------------------------------ |
| `Self` | A new repository instance with the updated path. |

Example::

```text
storage = uow.user_files.at("user1/")
print(storage.cwd) # "user1/"
```

Example::

```text
@dataclass(slots=True)
class YearMonth:
    year: int
    month: int

    def render(self) -> str:
        return f"{self.year}/{self.month:02d}"

storage = uow.user_files.at("user1/").at(YearMonth(year=2026, month=1))
print(storage.cwd) # user1/2026/01/
```

Source code in `src/fennflow/repositories/at.py`

```python
def at(self, path: str | PathTemplate) -> Self:
    """Return a new repository instance scoped to the given path.

    Args:
        path: Path to scope the repository to, relative to the current path.

    Returns:
        A new repository instance with the updated path.

    Example::

        storage = uow.user_files.at("user1/")
        print(storage.cwd) # "user1/"

    Example::

        @dataclass(slots=True)
        class YearMonth:
            year: int
            month: int

            def render(self) -> str:
                return f"{self.year}/{self.month:02d}"

        storage = uow.user_files.at("user1/").at(YearMonth(year=2026, month=1))
        print(storage.cwd) # user1/2026/01/

    """
    if isinstance(path, str):
        resolved = path
    elif isinstance(path, PathTemplate):
        resolved = path.render()
    else:
        raise TypeError(
            f"at() expects a str or PathTemplate, got {type(path).__name__!r}. "
            "Define a render() -> str method on your class "
            "to use it as a path template."
        )

    new_path = self._join_path(resolved)

    return self.__class__(self._uow, new_path, repo_extra=self.repo_extra)
```

## `AudioContent`

Bases: `BaseBinary`

Media content representing an audio file.

Attributes:

| Name       | Type  | Description |
| ---------- | ----- | ----------- |
| `duration` | \`int | None\`      |

Source code in `src/fennflow/files/media/audio_content.py`

```python
class AudioContent(BaseBinary):
    """Media content representing an audio file.

    Attributes:
        duration: Duration of the audio in seconds, if known.
    """

    duration: int | None = None
```

## `BaseBinary`

Bases: `BaseContent`

Base class for binary content types.

Attributes:

| Name   | Type    | Description        |
| ------ | ------- | ------------------ |
| `data` | `bytes` | raw file's content |

Source code in `src/fennflow/files/media/base_binary.py`

```python
class BaseBinary(BaseContent):
    """Base class for binary content types.

    Attributes:
        data: raw file's content
    """

    data: bytes = Field(repr=False)

    @property
    def data_size_mb(self) -> float:
        size_bytes = len(self.data)
        size_mb = round(size_bytes / (1024 * 1024), 2)
        return size_mb

    @field_serializer("data")
    def ser_data(self, v: bytes, _info):
        return base64.b64encode(v).decode("ascii")

    @field_validator("data", mode="before")
    @classmethod
    def de_data(cls, v):
        if isinstance(v, str):
            return base64.b64decode(v)
        return v
```

## `BinaryContent`

Bases: `BaseBinary`

Class for arbitrary binary data.

Source code in `src/fennflow/files/media/binary_content.py`

```python
class BinaryContent(BaseBinary):
    """Class for arbitrary binary data."""

    @classmethod
    def from_local_path(
        cls,
        path: str | Path,
        media_type: Omittable[MediaTypes] = OMIT,
        **kwargs: Any,
    ) -> Self:
        """Create a ``BinaryContent`` instance from the local filesystem.

        The MIME type is guessed from the file extension when ``media_type`` is not
        provided. The filename is taken from the final component of ``path`` and can
        be overridden via ``kwargs``.

        Unlike ``ContentFactory.from_local_path``, this method always returns an
        instance of the class it is called on — it never resolves a richer subtype
        from the content registry.

        Args:
            path: Absolute or relative path to the file, as a ``str`` or
                ``pathlib.Path``.
            media_type: MIME type to assign to the content. When omitted, the type
                is guessed from the file extension via ``MimeTypeGuesser``.
            **kwargs: Additional fields forwarded to the content model (e.g.
                ``filename`` to override the name derived from ``path``,
                or arbitrary keys stored in ``extra_metadata``).

        Returns:
            An instance of ``BinaryContent`` (or the subclass this method is called
            on) containing the file's raw bytes and resolved metadata.

        Raises:
            FileNotFoundError: If ``path`` does not point to an existing file.
            ExtensionCannotBeGuessed: If ``media_type`` is omitted and the extension
                cannot be mapped to a known MIME type.

        Example::

            from fennflow.files import MediaType
            from fennflow.files import BinaryContent

            # MIME type guessed from extension
            content = BinaryContent.from_local_path("report.pdf")

            # Explicit MIME type
            content = BinaryContent.from_local_path(
                "dump.bin",
                media_type=MediaType.APPLICATION_OCTET_STREAM,
            )

            # Override filename and attach extra metadata
            content = BinaryContent.from_local_path(
                "/tmp/upload_xyz",
                media_type=MediaType.IMAGE_PNG,
                filename="avatar.png",
                uploaded_by="alice",
            )


        Notes:
            This method is not inherited by other binary content classes
            such as TextContent, ImageContent, etc. Because this method returns
            an instance of the class it is called on, inheritance would allow
            incorrect usage — for example, obtaining an ImageContent instance
            from "text.txt". To get a specific content type, use
            ContentFactory.from_local_path or the class constructor directly
            (e.g. ImageContent(...)).
        """
        filename = Path(path).name

        if not is_given(media_type):
            media_type = MimeTypeGuesser.guess_type(filename=filename)

        kwargs.setdefault("filename", filename)

        with open(path, "rb") as file:
            return cls(
                media_type=media_type,
                data=file.read(),
                **kwargs,
            )
```

### `from_local_path(path, media_type=OMIT, **kwargs)`

Create a `BinaryContent` instance from the local filesystem.

The MIME type is guessed from the file extension when `media_type` is not provided. The filename is taken from the final component of `path` and can be overridden via `kwargs`.

Unlike `ContentFactory.from_local_path`, this method always returns an instance of the class it is called on — it never resolves a richer subtype from the content registry.

Parameters:

| Name         | Type                    | Description                                                                                                                                          | Default                                                          |
| ------------ | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------- |
| `path`       | \`str                   | Path\`                                                                                                                                               | Absolute or relative path to the file, as a str or pathlib.Path. |
| `media_type` | `Omittable[MediaTypes]` | MIME type to assign to the content. When omitted, the type is guessed from the file extension via MimeTypeGuesser.                                   | `OMIT`                                                           |
| `**kwargs`   | `Any`                   | Additional fields forwarded to the content model (e.g. filename to override the name derived from path, or arbitrary keys stored in extra_metadata). | `{}`                                                             |

Returns:

| Type   | Description                                                         |
| ------ | ------------------------------------------------------------------- |
| `Self` | An instance of BinaryContent (or the subclass this method is called |
| `Self` | on) containing the file's raw bytes and resolved metadata.          |

Raises:

| Type                       | Description                                                                       |
| -------------------------- | --------------------------------------------------------------------------------- |
| `FileNotFoundError`        | If path does not point to an existing file.                                       |
| `ExtensionCannotBeGuessed` | If media_type is omitted and the extension cannot be mapped to a known MIME type. |

Example::

```text
from fennflow.files import MediaType
from fennflow.files import BinaryContent

# MIME type guessed from extension
content = BinaryContent.from_local_path("report.pdf")

# Explicit MIME type
content = BinaryContent.from_local_path(
    "dump.bin",
    media_type=MediaType.APPLICATION_OCTET_STREAM,
)

# Override filename and attach extra metadata
content = BinaryContent.from_local_path(
    "/tmp/upload_xyz",
    media_type=MediaType.IMAGE_PNG,
    filename="avatar.png",
    uploaded_by="alice",
)
```

Notes

This method is not inherited by other binary content classes such as TextContent, ImageContent, etc. Because this method returns an instance of the class it is called on, inheritance would allow incorrect usage — for example, obtaining an ImageContent instance from "text.txt". To get a specific content type, use ContentFactory.from_local_path or the class constructor directly (e.g. ImageContent(...)).

Source code in `src/fennflow/files/media/binary_content.py`

```python
@classmethod
def from_local_path(
    cls,
    path: str | Path,
    media_type: Omittable[MediaTypes] = OMIT,
    **kwargs: Any,
) -> Self:
    """Create a ``BinaryContent`` instance from the local filesystem.

    The MIME type is guessed from the file extension when ``media_type`` is not
    provided. The filename is taken from the final component of ``path`` and can
    be overridden via ``kwargs``.

    Unlike ``ContentFactory.from_local_path``, this method always returns an
    instance of the class it is called on — it never resolves a richer subtype
    from the content registry.

    Args:
        path: Absolute or relative path to the file, as a ``str`` or
            ``pathlib.Path``.
        media_type: MIME type to assign to the content. When omitted, the type
            is guessed from the file extension via ``MimeTypeGuesser``.
        **kwargs: Additional fields forwarded to the content model (e.g.
            ``filename`` to override the name derived from ``path``,
            or arbitrary keys stored in ``extra_metadata``).

    Returns:
        An instance of ``BinaryContent`` (or the subclass this method is called
        on) containing the file's raw bytes and resolved metadata.

    Raises:
        FileNotFoundError: If ``path`` does not point to an existing file.
        ExtensionCannotBeGuessed: If ``media_type`` is omitted and the extension
            cannot be mapped to a known MIME type.

    Example::

        from fennflow.files import MediaType
        from fennflow.files import BinaryContent

        # MIME type guessed from extension
        content = BinaryContent.from_local_path("report.pdf")

        # Explicit MIME type
        content = BinaryContent.from_local_path(
            "dump.bin",
            media_type=MediaType.APPLICATION_OCTET_STREAM,
        )

        # Override filename and attach extra metadata
        content = BinaryContent.from_local_path(
            "/tmp/upload_xyz",
            media_type=MediaType.IMAGE_PNG,
            filename="avatar.png",
            uploaded_by="alice",
        )


    Notes:
        This method is not inherited by other binary content classes
        such as TextContent, ImageContent, etc. Because this method returns
        an instance of the class it is called on, inheritance would allow
        incorrect usage — for example, obtaining an ImageContent instance
        from "text.txt". To get a specific content type, use
        ContentFactory.from_local_path or the class constructor directly
        (e.g. ImageContent(...)).
    """
    filename = Path(path).name

    if not is_given(media_type):
        media_type = MimeTypeGuesser.guess_type(filename=filename)

    kwargs.setdefault("filename", filename)

    with open(path, "rb") as file:
        return cls(
            media_type=media_type,
            data=file.read(),
            **kwargs,
        )
```

## `ContentFactory`

Factory for creating media content instances from raw data.

Resolves the appropriate content class from the registry based on MIME type, falling back to `BaseBinary` for unknown types.

Source code in `src/fennflow/files/factory.py`

```python
class ContentFactory:
    """Factory for creating media content instances from raw data.

    Resolves the appropriate content class from the registry based on
    MIME type, falling back to ``BaseBinary`` for unknown types.

    """

    @staticmethod
    @cache
    def _get_prefixes() -> list[str]:
        """Return registry prefixes sorted by length descending for match resolution."""
        return sorted(
            (p for p in content_registry if p.endswith("/")),
            key=len,
            reverse=True,
        )

    @classmethod
    def from_bytes(
        cls,
        media_type: MediaTypes,
        data: bytes,
        **kwargs: Any,
    ) -> BinaryMedia:
        """Create a media content instance from raw bytes.

        Resolves the content class from the registry by exact MIME type match,
        then by prefix match, falling back to ``BaseBinary`` if no match is found.

        Args:
            media_type: The MIME type of the content (e.g. ``"text/plain"``).
            data: The raw bytes to wrap.
            **kwargs: Additional fields passed to the content model.

        Returns:
            A media content instance appropriate for the given MIME type.

        Raises:
            ValueError: If the resolved content class fails validation.

        **Example**:

            from fennflow.files import MediaType
            content = ContentFactory.from_bytes(MediaType.TEXT_PLAIN, b"Hello, World!")
        """
        payload = {
            "media_type": media_type,
            "data": data,
            **kwargs,
        }

        if media_type in content_registry:
            content_cls = content_registry[media_type]
        else:
            for prefix in cls._get_prefixes():
                if media_type.startswith(prefix):
                    content_cls = content_registry[prefix]
                    break
            else:
                content_cls = BaseBinary

        try:
            return content_cls.model_validate(payload)
        except ValidationError as exc:
            raise ValueError(
                f"Failed to validate {content_cls.__name__=} for {media_type=}"
            ) from exc

    @staticmethod
    def from_url(
        url: str,
        media_type: MediaTypes = MediaType.APPLICATION_OCTET_STREAM,
        **kwargs: Any,
    ) -> UrlContent:
        """Create a ``UrlContent`` instance from a URL string.

        Args:
            url: The URL string to wrap.
            media_type: The MIME type of the resource.
                Defaults to ``"application/octet-stream"``.
            **kwargs: Additional fields passed to the content model.

        Returns:
            A ``UrlContent`` instance wrapping the given URL.

        Raises:
            ValueError: If the resolved content class fails validation.

        **Example**::

            url = ContentFactory.from_url("https://example.com/file.txt")
        """
        if "filename" not in kwargs:
            kwargs["filename"] = FilenameGenerator.generate_from_url(url)

        payload = {
            "data": url,
            "media_type": media_type,
            **kwargs,
        }

        try:
            return UrlContent.model_validate(payload)
        except ValidationError as exc:
            raise ValueError(f"Failed to create UrlContent for {url=}") from exc

    @classmethod
    def from_local_path(
        cls,
        path: str | Path,
        media_type: Omittable[MediaTypes] = OMIT,
        **kwargs: Any,
    ) -> BinaryMedia:
        filename = Path(path).name

        if not is_given(media_type):
            media_type = MimeTypeGuesser.guess_type(filename=filename)

        kwargs.setdefault("filename", filename)

        with open(path, "rb") as file:
            return cls.from_bytes(
                data=file.read(),
                media_type=media_type,
                **kwargs,
            )
```

### `from_bytes(media_type, data, **kwargs)`

Create a media content instance from raw bytes.

Resolves the content class from the registry by exact MIME type match, then by prefix match, falling back to `BaseBinary` if no match is found.

Parameters:

| Name         | Type         | Description                                       | Default    |
| ------------ | ------------ | ------------------------------------------------- | ---------- |
| `media_type` | `MediaTypes` | The MIME type of the content (e.g. "text/plain"). | *required* |
| `data`       | `bytes`      | The raw bytes to wrap.                            | *required* |
| `**kwargs`   | `Any`        | Additional fields passed to the content model.    | `{}`       |

Returns:

| Type          | Description                                                   |
| ------------- | ------------------------------------------------------------- |
| `BinaryMedia` | A media content instance appropriate for the given MIME type. |

Raises:

| Type         | Description                                     |
| ------------ | ----------------------------------------------- |
| `ValueError` | If the resolved content class fails validation. |

**Example**:

```text
from fennflow.files import MediaType
content = ContentFactory.from_bytes(MediaType.TEXT_PLAIN, b"Hello, World!")
```

Source code in `src/fennflow/files/factory.py`

```python
@classmethod
def from_bytes(
    cls,
    media_type: MediaTypes,
    data: bytes,
    **kwargs: Any,
) -> BinaryMedia:
    """Create a media content instance from raw bytes.

    Resolves the content class from the registry by exact MIME type match,
    then by prefix match, falling back to ``BaseBinary`` if no match is found.

    Args:
        media_type: The MIME type of the content (e.g. ``"text/plain"``).
        data: The raw bytes to wrap.
        **kwargs: Additional fields passed to the content model.

    Returns:
        A media content instance appropriate for the given MIME type.

    Raises:
        ValueError: If the resolved content class fails validation.

    **Example**:

        from fennflow.files import MediaType
        content = ContentFactory.from_bytes(MediaType.TEXT_PLAIN, b"Hello, World!")
    """
    payload = {
        "media_type": media_type,
        "data": data,
        **kwargs,
    }

    if media_type in content_registry:
        content_cls = content_registry[media_type]
    else:
        for prefix in cls._get_prefixes():
            if media_type.startswith(prefix):
                content_cls = content_registry[prefix]
                break
        else:
            content_cls = BaseBinary

    try:
        return content_cls.model_validate(payload)
    except ValidationError as exc:
        raise ValueError(
            f"Failed to validate {content_cls.__name__=} for {media_type=}"
        ) from exc
```

### `from_url(url, media_type=MediaType.APPLICATION_OCTET_STREAM, **kwargs)`

Create a `UrlContent` instance from a URL string.

Parameters:

| Name         | Type         | Description                                                            | Default                    |
| ------------ | ------------ | ---------------------------------------------------------------------- | -------------------------- |
| `url`        | `str`        | The URL string to wrap.                                                | *required*                 |
| `media_type` | `MediaTypes` | The MIME type of the resource. Defaults to "application/octet-stream". | `APPLICATION_OCTET_STREAM` |
| `**kwargs`   | `Any`        | Additional fields passed to the content model.                         | `{}`                       |

Returns:

| Type         | Description                                   |
| ------------ | --------------------------------------------- |
| `UrlContent` | A UrlContent instance wrapping the given URL. |

Raises:

| Type         | Description                                     |
| ------------ | ----------------------------------------------- |
| `ValueError` | If the resolved content class fails validation. |

**Example**::

```text
url = ContentFactory.from_url("https://example.com/file.txt")
```

Source code in `src/fennflow/files/factory.py`

```python
@staticmethod
def from_url(
    url: str,
    media_type: MediaTypes = MediaType.APPLICATION_OCTET_STREAM,
    **kwargs: Any,
) -> UrlContent:
    """Create a ``UrlContent`` instance from a URL string.

    Args:
        url: The URL string to wrap.
        media_type: The MIME type of the resource.
            Defaults to ``"application/octet-stream"``.
        **kwargs: Additional fields passed to the content model.

    Returns:
        A ``UrlContent`` instance wrapping the given URL.

    Raises:
        ValueError: If the resolved content class fails validation.

    **Example**::

        url = ContentFactory.from_url("https://example.com/file.txt")
    """
    if "filename" not in kwargs:
        kwargs["filename"] = FilenameGenerator.generate_from_url(url)

    payload = {
        "data": url,
        "media_type": media_type,
        **kwargs,
    }

    try:
        return UrlContent.model_validate(payload)
    except ValidationError as exc:
        raise ValueError(f"Failed to create UrlContent for {url=}") from exc
```

## `DocumentContent`

Bases: `BaseBinary`

Media content representing a document.

Source code in `src/fennflow/files/media/document_content.py`

```python
class DocumentContent(BaseBinary):
    """Media content representing a document."""
```

## `ImageContent`

Bases: `BaseBinary`

Media content representing an image file.

Attributes:

| Name     | Type  | Description |
| -------- | ----- | ----------- |
| `height` | \`int | None\`      |
| `width`  | \`int | None\`      |

Source code in `src/fennflow/files/media/image_content.py`

```python
class ImageContent(BaseBinary):
    """Media content representing an image file.

    Attributes:
        height: Height of the image in pixels, if known.
        width: Width of the image in pixels, if known.
    """

    height: int | None = None
    width: int | None = None
```

## `JsonContent`

Bases: `BaseBinary`, `FromContentAbstract`, `ContentPropertyAbstract`

Media content representing a JSON file.

Stores JSON data as UTF-8 encoded bytes internally. Use `from_content()` to create from a Python object.

Attributes:

| Name       | Type  | Description                             |
| ---------- | ----- | --------------------------------------- |
| `encoding` | `str` | The text encoding. Defaults to "utf-8". |

Example::

```text
file = JsonContent.from_content({"key": "value"})
print(file.content) # {"key": "value"}
await uow.user_files.at("user1/").put(file)
```

Source code in `src/fennflow/files/media/json_content.py`

```python
class JsonContent(
    BaseBinary,
    FromContentAbstract,
    ContentPropertyAbstract,
):
    """Media content representing a JSON file.

    Stores JSON data as UTF-8 encoded bytes internally.
    Use ``from_content()`` to create from a Python object.

    Attributes:
        encoding: The text encoding. Defaults to ``"utf-8"``.

    Example::

        file = JsonContent.from_content({"key": "value"})
        print(file.content) # {"key": "value"}
        await uow.user_files.at("user1/").put(file)
    """

    encoding: str = "utf-8"

    @property
    def content(self) -> JsonValue:
        text = self.data.decode(self.encoding)
        return json.loads(text)

    @classmethod
    def from_content(
        cls,
        data: JsonValue,
        media_type: MediaTypes = MediaType.APPLICATION_JSON,
        encoding: str = "utf-8",
        filename: Omittable[str] = OMIT,
        ensure_ascii: bool = False,
        indent: int | str | None = None,
        **extra_json_dumps_kwargs,
    ) -> JsonContent:
        dumped_data = json.dumps(
            data,
            ensure_ascii=ensure_ascii,
            indent=indent,
            **extra_json_dumps_kwargs,
        )

        extra: dict[str, Any] = {}

        if is_given(filename):
            extra["filename"] = filename

        return cls(
            data=dumped_data.encode(encoding),
            media_type=media_type,
            encoding=encoding,
            **extra,
        )
```

## `MediaType`

Bases: `StrEnum`

Common MIME types for use with FennFlow content models.

Source code in `src/fennflow/files/enums.py`

```python
class MediaType(StrEnum):
    """Common MIME types for use with FennFlow content models."""

    # --- Text (→ TextContent) ---
    TEXT_PLAIN = "text/plain"
    TEXT_HTML = "text/html"
    TEXT_CSS = "text/css"
    TEXT_CSV = "text/csv"
    TEXT_XML = "text/xml"
    TEXT_MARKDOWN = "text/markdown"

    # --- Application / structured (→ JsonContent or DocumentContent) ---
    APPLICATION_JSON = "application/json"
    APPLICATION_XML = "application/xml"
    APPLICATION_PDF = "application/pdf"
    APPLICATION_ZIP = "application/zip"
    APPLICATION_GZIP = "application/gzip"
    APPLICATION_OCTET_STREAM = "application/octet-stream"

    # --- Images (→ ImageContent) ---
    IMAGE_JPEG = "image/jpeg"
    IMAGE_PNG = "image/png"
    IMAGE_GIF = "image/gif"
    IMAGE_WEBP = "image/webp"
    IMAGE_SVG = "image/svg+xml"
    IMAGE_TIFF = "image/tiff"
    IMAGE_BMP = "image/bmp"
    IMAGE_ICO = "image/x-icon"

    # --- Audio (→ AudioContent) ---
    AUDIO_MPEG = "audio/mpeg"
    AUDIO_OGG = "audio/ogg"
    AUDIO_WAV = "audio/wav"
    AUDIO_WEBM = "audio/webm"
    AUDIO_FLAC = "audio/flac"
    AUDIO_AAC = "audio/aac"

    # --- Video (→ VideoContent) ---
    VIDEO_MP4 = "video/mp4"
    VIDEO_WEBM = "video/webm"
    VIDEO_OGG = "video/ogg"
    VIDEO_QUICKTIME = "video/quicktime"
    VIDEO_AVI = "video/x-msvideo"
```

## `TextContent`

Bases: `BaseBinary`, `FromContentAbstract`, `ContentPropertyAbstract`

Media content representing a plain text file.

Stores text as UTF-8 encoded bytes internally. Use `from_content()` to create from a string.

Attributes:

| Name       | Type  | Description                             |
| ---------- | ----- | --------------------------------------- |
| `encoding` | `str` | The text encoding. Defaults to "utf-8". |

Example::

```text
file = TextContent.from_content("Hello, World!")
print(file.content) # "Hello, World!"
await uow.user_files.at("user1/").put(file)
```

Source code in `src/fennflow/files/media/text_content.py`

```python
class TextContent(
    BaseBinary,
    FromContentAbstract,
    ContentPropertyAbstract,
):
    """Media content representing a plain text file.

    Stores text as UTF-8 encoded bytes internally.
    Use ``from_content()`` to create from a string.

    Attributes:
        encoding: The text encoding. Defaults to ``"utf-8"``.

    Example::

        file = TextContent.from_content("Hello, World!")
        print(file.content) # "Hello, World!"
        await uow.user_files.at("user1/").put(file)
    """

    encoding: str = "utf-8"

    @property
    def content(self) -> str:
        try:
            return self.data.decode(self.encoding)
        except UnicodeDecodeError as e:
            raise ValueError(f"Cannot extract text from {self=}") from e

    @classmethod
    def from_content(
        cls,
        data: str,
        media_type: MediaTypes = MediaType.TEXT_PLAIN,
        encoding: str = "utf-8",
        filename: Omittable[str] = OMIT,
        **kwargs,  # noqa: ARG003
    ) -> Self:
        extra: dict[str, Any] = {}

        if is_given(filename):
            extra["filename"] = filename

        return cls(
            data=data.encode(encoding),
            media_type=media_type,
            encoding=encoding,
            **extra,
        )
```

## `UrlContent`

Bases: `BaseContent`

Media content representing a URL.

Attributes:

| Name   | Type  | Description     |
| ------ | ----- | --------------- |
| `data` | `str` | The URL string. |

Source code in `src/fennflow/files/media/url_content.py`

```python
class UrlContent(BaseContent):
    """Media content representing a URL.

    Attributes:
        data: The URL string.
    """

    data: str
```

## `VideoContent`

Bases: `BaseBinary`

Media content representing a video file.

Attributes:

| Name       | Type  | Description |
| ---------- | ----- | ----------- |
| `duration` | \`int | None\`      |
| `height`   | \`int | None\`      |
| `width`    | \`int | None\`      |

Source code in `src/fennflow/files/media/video_content.py`

```python
class VideoContent(BaseBinary):
    """Media content representing a video file.

    Attributes:
        duration: Duration of the video in seconds.
        height: Height of the video in pixels.
        width: Width of the video in pixels.
    """

    duration: int | None = None
    height: int | None = None
    width: int | None = None
```

## `ReconcileConfig`

Bases: `BasePydanticConfig`

Configuration for the reconciler.

Source code in `src/fennflow/reconciler/config.py`

```python
class ReconcileConfig(BasePydanticConfig):
    """Configuration for the reconciler."""

    frequency: (
        Literal["on_start_app", "on_session_start", "never"] | ReconcileFrequencyEnum
    ) = ReconcileFrequencyEnum.ON_START_APP
    strategy: (
        Literal["fill_if_empty", "replace", "insert_missing"] | ReconcileStrategyEnum
    ) = ReconcileStrategyEnum.FILL_IF_EMPTY
    batch_size: int = 1000
```

## `ReconcileFrequencyEnum`

Bases: `StrEnum`

Controls how often reconciliation is performed.

Attributes:

| Name               | Type | Description                                    |
| ------------------ | ---- | ---------------------------------------------- |
| `ON_START_APP`     |      | Reconcile once per process lifetime.           |
| `ON_SESSION_START` |      | Reconcile on every UnitOfWork.__aenter__ call. |
| `NEVER`            |      | Disable reconciliation.                        |

Source code in `src/fennflow/reconciler/enums.py`

```python
class ReconcileFrequencyEnum(StrEnum):
    """Controls how often reconciliation is performed.

    Attributes:
        ON_START_APP: Reconcile once per process lifetime.
        ON_SESSION_START: Reconcile on every ``UnitOfWork.__aenter__`` call.
        NEVER: Disable reconciliation.
    """

    ON_START_APP = "on_start_app"
    ON_SESSION_START = "on_session_start"
    NEVER = "never"
```

## `ReconcileStrategyEnum`

Bases: `StrEnum`

Defines how reconciliation updates existing data.

Attributes:

| Name             | Type | Description                                             |
| ---------------- | ---- | ------------------------------------------------------- |
| `FILL_IF_EMPTY`  |      | Reconcile only if the backend is empty.                 |
| `REPLACE`        |      | Reset all backend data before reconciling.              |
| `INSERT_MISSING` |      | Insert missing data while keeping existing data intact. |

Source code in `src/fennflow/reconciler/enums.py`

```python
class ReconcileStrategyEnum(StrEnum):
    """Defines how reconciliation updates existing data.

    Attributes:
        FILL_IF_EMPTY: Reconcile only if the backend is empty.
        REPLACE: Reset all backend data before reconciling.
        INSERT_MISSING: Insert missing data while keeping existing data intact.
    """

    FILL_IF_EMPTY = "fill_if_empty"
    REPLACE = "replace"
    INSERT_MISSING = "insert_missing"
```

## `Reconciler`

Synchronizes backend state with actual connector (storage) state.

On startup or session start, the backend may be out of sync with the real storage (e.g. on first connection with a persistent backend). `Reconciler` restores consistency by listing files from the connector and inserting them into the backend according to the chosen strategy. Called internally by `ReconcileOrchestrator` in UnitOfWork.**aenter**.

Notes

Reconciler does not perform garbage collection!

Example::

```text
import asyncio

from fennflow.reconciler import Reconciler, ReconcileStrategyEnum
from fennflow.uow import UowInspector


async def main():
    async with UOW() as uow:
        uow_inspector = UowInspector(uow=uow)
        reconcile = Reconciler(
            uow_fields=uow_inspector.get_repo_fields(),
            connector=uow.connector,
            backend=uow.backend,
            )
        await reconcile.reconcile(
            session_id=uow._session_id,
            batch_size=500,
            strategy=ReconcileStrategyEnum.REPLACE,
            backend_scope=uow.config["connector"].scope
            )

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

Source code in `src/fennflow/reconciler/core.py`

```python
class Reconciler:
    """Synchronizes backend state with actual connector (storage) state.

    On startup or session start, the backend may be out of sync with the
    real storage (e.g. on first connection with a persistent backend).
     ``Reconciler`` restores consistency
    by listing files from the connector and inserting them into the backend
    according to the chosen strategy.
    Called internally by ``ReconcileOrchestrator`` in UnitOfWork.__aenter__.

    Notes:
        Reconciler does not perform garbage collection!

    Example::

        import asyncio

        from fennflow.reconciler import Reconciler, ReconcileStrategyEnum
        from fennflow.uow import UowInspector


        async def main():
            async with UOW() as uow:
                uow_inspector = UowInspector(uow=uow)
                reconcile = Reconciler(
                    uow_fields=uow_inspector.get_repo_fields(),
                    connector=uow.connector,
                    backend=uow.backend,
                    )
                await reconcile.reconcile(
                    session_id=uow._session_id,
                    batch_size=500,
                    strategy=ReconcileStrategyEnum.REPLACE,
                    backend_scope=uow.config["connector"].scope
                    )

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

    def __init__(
        self,
        uow_fields: Iterable[RepoField],
        backend: BackendOrchestrator,
        connector: AbstractConnector,
    ) -> None:
        """Init method.

        Args:
            uow_fields: Repository field descriptors to reconcile.
                Each field provides the namespace and repo config needed to
                list objects from the connector.
            backend: The backend to sync state into.
            connector: The storage connector to read the source-of-truth from.
        """
        self.uow_fields = uow_fields
        self.backend = backend
        self.connector = connector

    @reraise_with(ReconcileFailedException())
    async def reconcile(
        self,
        session_id: UUID,
        strategy: ReconcileStrategyEnum,
        batch_size: int,
        backend_scope: BackendScope,
    ) -> None:
        """Reconcile all registered repository fields against the connector.

        Iterates over each ``RepoField``, lists its objects from the connector
        in pages, and inserts them into the backend. The conflict resolution
        behavior is determined by ``strategy``.

        Args:
            session_id: Session ID to stamp on inserted ``OperationRecord``s.
            strategy: Controls whether to skip reconciliation, overwrite existing
                records, or only insert missing ones. See ``ReconcileStrategyEnum``.
            batch_size: Number of objects to fetch per page from the connector.
            backend_scope: Scope to assign to inserted records in the backend.

        Raises:
            ReconcileFailedException: If any error occurs during reconciliation.
        """
        if not await self._should_reconcile(
            strategy=strategy,
            backend_scope=backend_scope,
        ):
            return

        for repo in self.uow_fields:
            on_conflict = reconcile_to_on_conflict_strategy[strategy]

            async for page in self._iter_pages(repo, batch_size=batch_size):
                await self.backend.backend_engine.execute(
                    InsertQuerySpec.from_operations(
                        operations=self._records_from_page(
                            session_id=session_id,
                            page=page,
                            repo_extra=repo.repo_extra,
                            backend_scope=backend_scope,
                        ),
                        on_conflict=on_conflict,
                    )
                )

    async def _should_reconcile(
        self,
        strategy: ReconcileStrategyEnum,
        backend_scope: BackendScope,
    ) -> bool:
        if strategy == ReconcileStrategyEnum.FILL_IF_EMPTY:
            return await self.backend.backend_engine.execute(
                IsEmptyQuerySpec(scope=backend_scope)
            )
        return True

    async def _iter_pages(
        self,
        repo: RepoField,
        batch_size: int,
    ) -> AsyncGenerator[ListResponse, None]:
        continuation_token = None

        while True:
            page = await self.connector.list_objects(
                prefix="",
                limit=batch_size,
                repo_extra=repo.repo_extra,
                continuation_token=continuation_token,
            )

            if not page.storage_paths:
                break

            yield page

            if page.continuation_token is None:
                break
            continuation_token = page.continuation_token

    @staticmethod
    def _records_from_page(
        session_id: UUID,
        page: ListResponse,
        repo_extra: RepoExtra,
        backend_scope: str,
    ) -> Generator[OperationRecord, None, None]:
        for storage_path in page:
            yield OperationRecord.create(
                session_id=session_id,
                storage_path=storage_path,
                operation_type=OperationTypeEnum.PUT,
                status=OperationStatusEnum.UPLOADED,
                repo_extra=repo_extra,
                scope=backend_scope,
            )

    @staticmethod
    def _get_repo_fields(uow: UnitOfWork) -> Generator[RepoField, None, None]:
        for field in vars(type(uow)).values():
            if isinstance(field, RepoField):
                yield field
```

### `__init__(uow_fields, backend, connector)`

Init method.

Parameters:

| Name         | Type                  | Description                                                                                                                             | Default    |
| ------------ | --------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `uow_fields` | `Iterable[RepoField]` | Repository field descriptors to reconcile. Each field provides the namespace and repo config needed to list objects from the connector. | *required* |
| `backend`    | `BackendOrchestrator` | The backend to sync state into.                                                                                                         | *required* |
| `connector`  | `AbstractConnector`   | The storage connector to read the source-of-truth from.                                                                                 | *required* |

Source code in `src/fennflow/reconciler/core.py`

```python
def __init__(
    self,
    uow_fields: Iterable[RepoField],
    backend: BackendOrchestrator,
    connector: AbstractConnector,
) -> None:
    """Init method.

    Args:
        uow_fields: Repository field descriptors to reconcile.
            Each field provides the namespace and repo config needed to
            list objects from the connector.
        backend: The backend to sync state into.
        connector: The storage connector to read the source-of-truth from.
    """
    self.uow_fields = uow_fields
    self.backend = backend
    self.connector = connector
```

### `reconcile(session_id, strategy, batch_size, backend_scope)`

Reconcile all registered repository fields against the connector.

Iterates over each `RepoField`, lists its objects from the connector in pages, and inserts them into the backend. The conflict resolution behavior is determined by `strategy`.

Parameters:

| Name            | Type                    | Description                                                                                                                  | Default    |
| --------------- | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------- | ---------- |
| `session_id`    | `UUID`                  | Session ID to stamp on inserted OperationRecords.                                                                            | *required* |
| `strategy`      | `ReconcileStrategyEnum` | Controls whether to skip reconciliation, overwrite existing records, or only insert missing ones. See ReconcileStrategyEnum. | *required* |
| `batch_size`    | `int`                   | Number of objects to fetch per page from the connector.                                                                      | *required* |
| `backend_scope` | `BackendScope`          | Scope to assign to inserted records in the backend.                                                                          | *required* |

Raises:

| Type                       | Description                                |
| -------------------------- | ------------------------------------------ |
| `ReconcileFailedException` | If any error occurs during reconciliation. |

Source code in `src/fennflow/reconciler/core.py`

```python
@reraise_with(ReconcileFailedException())
async def reconcile(
    self,
    session_id: UUID,
    strategy: ReconcileStrategyEnum,
    batch_size: int,
    backend_scope: BackendScope,
) -> None:
    """Reconcile all registered repository fields against the connector.

    Iterates over each ``RepoField``, lists its objects from the connector
    in pages, and inserts them into the backend. The conflict resolution
    behavior is determined by ``strategy``.

    Args:
        session_id: Session ID to stamp on inserted ``OperationRecord``s.
        strategy: Controls whether to skip reconciliation, overwrite existing
            records, or only insert missing ones. See ``ReconcileStrategyEnum``.
        batch_size: Number of objects to fetch per page from the connector.
        backend_scope: Scope to assign to inserted records in the backend.

    Raises:
        ReconcileFailedException: If any error occurs during reconciliation.
    """
    if not await self._should_reconcile(
        strategy=strategy,
        backend_scope=backend_scope,
    ):
        return

    for repo in self.uow_fields:
        on_conflict = reconcile_to_on_conflict_strategy[strategy]

        async for page in self._iter_pages(repo, batch_size=batch_size):
            await self.backend.backend_engine.execute(
                InsertQuerySpec.from_operations(
                    operations=self._records_from_page(
                        session_id=session_id,
                        page=page,
                        repo_extra=repo.repo_extra,
                        backend_scope=backend_scope,
                    ),
                    on_conflict=on_conflict,
                )
            )
```