Skip to content

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

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 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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 property

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 property

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

Persists operation state via backend.

Source code in src/fennflow/uow/core.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
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() async

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 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
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 dataclass

Extracts info from Unit of Work.

Source code in src/fennflow/uow/inspector.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@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::

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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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::

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

    )
Source code in src/fennflow/core/configs/base.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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 
 8
 9
10
11
12
13
14
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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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::

connector = ConnectorFactory.from_config(S3ConnectorConfig(...))
Source code in src/fennflow/connectors/_factory.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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) staticmethod

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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@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::

class UOW(UnitOfWork):
    config = ConfigDict(
        connector=InMemoryConnectorConfig(),
    )
Source code in src/fennflow/connectors/in_memory/core.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
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 
4
5
6
7
8
9
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::

class UOW(UnitOfWork):
    config = ConfigDict(
        connector=S3ConnectorConfig(...),
    )
Source code in src/fennflow/connectors/s3/core.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
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 access key ID.
aws_secret_access_key str | None

AWS secret access key.
endpoint_url str | None

Custom endpoint URL for S3-compatible storage.
aiobotocore_config AioConfig | None

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()
Source code in src/fennflow/connectors/s3/config.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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) async

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:

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 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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) async

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

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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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) async

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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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) async

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

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 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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) async

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

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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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) async

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

async with UOW() as uow:
    await uow.user_files.get_prefix("some_prefix")
Source code in src/fennflow/repositories/get_prefix.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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) async

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

response = await uow.user_files.at("user1/").get("file.txt")
if response:
    file = response[0]
Source code in src/fennflow/repositories/get.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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) async

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

 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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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) async

Puts file into storage.

Example::

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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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::

class UOW(UnitOfWork):
    user_files = RepoField(UserFiles, namespace="user-files")
Source code in src/fennflow/repositories/fields/base.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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::

class UOW(UnitOfWork):
    user_files = S3RepoField(UserFiles, bucket_name="my-bucket")
Source code in src/fennflow/repositories/fields/s3.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 property

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.

 required

Returns:

Type Description
Self

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/
Source code in src/fennflow/repositories/at.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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

Duration of the audio in seconds, if known.
Source code in src/fennflow/files/media/audio_content.py 
 4
 5
 6
 7
 8
 9
10
11
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 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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) classmethod

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.

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

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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@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 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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) classmethod

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:

from fennflow.files import MediaType
content = ContentFactory.from_bytes(MediaType.TEXT_PLAIN, b"Hello, World!")
Source code in src/fennflow/files/factory.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@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) staticmethod

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

url = ContentFactory.from_url("https://example.com/file.txt")
Source code in src/fennflow/files/factory.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@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 
4
5
class DocumentContent(BaseBinary):
    """Media content representing a document."""

ImageContent

Bases: BaseBinary

Media content representing an image file.

Attributes:

Name Type Description
height int | None

Height of the image in pixels, if known.
width int | None

Width of the image in pixels, if known.
Source code in src/fennflow/files/media/image_content.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
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::

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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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::

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 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
 4
 5
 6
 7
 8
 9
10
11
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

Duration of the video in seconds.
height int | None

Height of the video in pixels.
width int | None

Width of the video in pixels.
Source code in src/fennflow/files/media/video_content.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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 
 7
 8
 9
10
11
12
13
14
15
16
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 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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 
18
19
20
21
22
23
24
25
26
27
28
29
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::

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 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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) async

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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@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,
                )
            )