"""File storage primitives for codegen-generated FastAPI projects.
This module's runtime dependency on ``boto3`` is gated behind the
``files`` extra. Install with::
pip install 'fsh-lib[files]'
# or: uv add 'fsh-lib[files]'
Importing this module without the extra raises ``ModuleNotFoundError``
on ``import boto3`` -- so the gate is honest rather than lazy:
either the dep is there and everything works, or it isn't and the
import surface fails fast.
A *file* is a binary blob (image, PDF, attachment) tracked by a
metadata row in the consumer's database and a corresponding object
in S3-compatible storage. This module ships three pieces:
* :class:`FileMixin` -- a codegen-database-compatible mixin supplying the
six storage columns every file row needs (``s3_key``,
``content_type``, ``size_bytes``, ``original_filename``,
``created_at``, ``uploaded_at``). Consumers subclass it on a
codegen-database model and add a PK plugin (typically
``UUIDV7PKPlugin``) for the ``id`` column.
* :class:`S3Storage` -- a small wrapper around ``boto3`` that
exposes the three operations a presigned-upload flow actually
needs: mint a presigned PUT URL, mint a presigned GET URL, delete
an object. The constructor takes explicit config so it's
testable; :func:`default_storage` builds one from ``FSH_S3_*``
env vars for the common case.
* Action functions -- :func:`request_upload`,
:func:`complete_upload`, :func:`download`, and :func:`delete_file`.
These plug into be's
:class:`~be.operations.action.Action` operation: the consumer
points ``resource.action`` entries at them directly (no
per-resource wrapper module). The :class:`FileMixin`-typed
parameters (instance for object actions, class for collection
actions) match any concrete subclass via the introspector's
supertype check, so the same four functions serve every file
resource.
"""
import datetime
import os
import re
import tempfile
import uuid
from dataclasses import dataclass, field
from functools import cached_property
from pathlib import Path
from typing import TYPE_CHECKING, Any
from urllib.parse import urlsplit
import boto3
from fastapi import FastAPI, HTTPException, Request, Response, status
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from sqlalchemy import (
BigInteger,
DateTime,
Text,
delete,
insert,
update,
)
from sqlalchemy.orm import Mapped, mapped_column
if TYPE_CHECKING:
from collections.abc import Callable
from sqlalchemy.ext.asyncio import AsyncSession
DEFAULT_PRESIGN_TTL = 900
"""Presigned URL lifetime in seconds (15 min).
Long enough for a browser to PUT a multi-megabyte file over a slow
connection; short enough that a leaked URL stops working before it
shows up in logs anyone reads.
"""
def _attachment_disposition(filename: str) -> str:
"""``Content-Disposition`` value that downloads as *filename*."""
base = filename.replace("\\", "/").rsplit("/", 1)[-1].replace('"', "")
return f'attachment; filename="{base}"'
def _sanitize_filename(filename: str) -> str:
"""Sanitized *filename* to suffix the storage key with."""
base = filename.replace("\\", "/").rsplit("/", 1)[-1]
return re.sub(r"[^A-Za-z0-9._-]+", "_", base)
[docs]
class StorageColumnsMixin:
"""Nullable storage columns for a file record: the base of FileMixin.
Contains the three core columns for tracking a stored file:
``s3_key``, ``content_type``, ``size_bytes``. All nullable so
async ops can create rows before the file is generated.
codegen-database's factory reuses this mixin's single shared
``mapped_column`` across every table it is mixed into, so it is safe
for one codegen-database model per process only. A second consumer
(e.g. the async-ops run table) declares the same columns inline
instead -- see :class:`fsh_lib.async_ops.AsyncOperationRunMixin`.
"""
s3_key: Mapped[str | None] = mapped_column(Text, nullable=True, unique=True)
content_type: Mapped[str | None] = mapped_column(Text, nullable=True)
size_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
[docs]
class FileMixin(StorageColumnsMixin):
"""codegen-database mixin supplying the storage columns of a file record.
Subclass on a codegen-database-mapped model alongside a PK plugin (the
plugin owns ``id``):
.. code-block:: python
from fsh_lib.files import FileMixin
from codegen_database.factory import CodegenDatabaseSimple
from codegen_database.plugins.pk import UUIDV7PKPlugin
class Attachment(Base, FileMixin):
__tablename__ = "attachments"
__table_args__ = {"schema": "public"}
__factory__ = CodegenDatabaseSimple
__plugins__ = [UUIDV7PKPlugin()]
The mixin deliberately doesn't declare ``id`` -- codegen-database's idiom
is that primary keys are plugin-owned, and declaring it on the
mixin would collide with the plugin's column at table-build time.
The ``TYPE_CHECKING`` annotation below keeps ``file.id`` typed
for the action helpers without committing to a column.
A row with ``uploaded_at is None`` represents a file the
server has reserved a key for (and handed the client a presigned
PUT URL) but whose upload hasn't yet been confirmed. Consumers
typically clear or expire these rows on a schedule.
Inherits ``s3_key``, ``content_type``, ``size_bytes`` from
:class:`StorageColumnsMixin`. For direct uploads where the key
is known at row-creation, override ``s3_key`` to be non-nullable
if desired.
"""
if TYPE_CHECKING:
id: Mapped[uuid.UUID]
s3_key: Mapped[str] = mapped_column(Text, unique=True)
original_filename: Mapped[str | None] = mapped_column(
Text,
nullable=True,
)
"""Filename the client supplied; useful for ``Content-Disposition``
on download. Not used for storage -- the canonical name is
:attr:`s3_key`."""
uploaded_at: Mapped[datetime.datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
"""When the upload was confirmed. ``None`` means pending --
metadata exists but the blob may or may not be in S3."""
[docs]
@dataclass
class S3Storage:
"""``boto3``-backed S3 client wrapper.
The constructor takes explicit config so tests can build an
instance pointed at a stub or a localstack endpoint without
setting env vars. :func:`default_storage` is the env-driven
factory for production use.
``client_factory`` is plumbed through so tests can inject a
``MagicMock`` instead of a real ``boto3.client``.
"""
bucket: str
region: str | None = None
endpoint_url: str | None = None
client_factory: Callable[..., Any] = field(default=boto3.client)
[docs]
@cached_property
def client(self) -> Any:
"""Lazily-built ``boto3`` S3 client.
Cached so a single :class:`S3Storage` instance reuses one
connection pool across calls.
"""
kwargs: dict[str, Any] = {"service_name": "s3"}
if self.region is not None:
kwargs["region_name"] = self.region
if self.endpoint_url is not None:
kwargs["endpoint_url"] = self.endpoint_url
return self.client_factory(**kwargs)
[docs]
def presigned_put_url(
self,
key: str,
*,
expires_in: int = DEFAULT_PRESIGN_TTL,
content_type: str | None = None,
) -> str:
"""Mint a presigned PUT URL for *key*.
When *content_type* is supplied, the client must send a
matching ``Content-Type`` header on the PUT or S3 rejects
the request -- this binds the upload to the type the row
was created for.
"""
params: dict[str, Any] = {"Bucket": self.bucket, "Key": key}
if content_type is not None:
params["ContentType"] = content_type
url = self.client.generate_presigned_url(
"put_object",
Params=params,
ExpiresIn=expires_in,
)
return str(url)
[docs]
def presigned_get_url(
self,
key: str,
*,
expires_in: int = DEFAULT_PRESIGN_TTL,
filename: str | None = None,
) -> str:
"""Mint a presigned GET URL for *key*.
When *filename* is supplied, the URL carries a
``ResponseContentDisposition`` so the browser saves the
download under that name -- the key is an opaque uuid, so this
is what restores the original filename on download.
"""
params: dict[str, Any] = {"Bucket": self.bucket, "Key": key}
if filename is not None:
params["ResponseContentDisposition"] = _attachment_disposition(
filename,
)
url = self.client.generate_presigned_url(
"get_object",
Params=params,
ExpiresIn=expires_in,
)
return str(url)
[docs]
def put_bytes(
self,
key: str,
*,
blob: bytes,
content_type: str,
) -> None:
"""Write *blob* to *key* with *content_type* (server-side).
The server-upload counterpart to :meth:`presigned_put_url`:
for bytes the server already holds (e.g. a generated report)
rather than a client-direct upload.
"""
self.client.put_object(
Bucket=self.bucket,
Key=key,
Body=blob,
ContentType=content_type,
)
[docs]
def delete(self, key: str) -> None:
"""Delete the object at *key*.
S3's ``DeleteObject`` is idempotent -- deleting a missing
key returns 204 the same as deleting an existing one -- so
callers don't need to guard against double-delete races.
"""
self.client.delete_object(Bucket=self.bucket, Key=key)
[docs]
@dataclass
class LocalStorage:
"""Local-disk storage backend for development without S3.
Mirrors :class:`S3Storage`'s interface so :func:`default_storage`
can return either transparently: server-held bytes write under
*root* via :meth:`put_bytes`, and both presigned URLs resolve to
a *base_url*-rooted path the app serves -- GET via a static mount,
PUT via a write route (see :func:`mount_local_storage`). So the
client-direct upload flow (request_upload -> PUT -> complete) works
unchanged on local disk; no ``FSH_S3_*`` needed for dev.
"""
root: Path
base_url: str = "/_files"
"""URL prefix download links are rooted at. Relative (``/_files``)
works when the API serves the app same-origin; set an absolute
prefix (e.g. ``http://localhost:8000/_files``) when the frontend
is a separate origin in dev, so links resolve to the API, not the
page. :func:`mount_local_storage` derives the static-mount path
from this."""
[docs]
def presigned_put_url(
self,
key: str,
*,
expires_in: int = DEFAULT_PRESIGN_TTL, # noqa: ARG002 -- no TTL local
content_type: str | None = None, # noqa: ARG002 -- from PUT header
) -> str:
"""Return the upload URL for *key* (no signing or expiry locally)."""
return f"{self.base_url}/{key}"
[docs]
def presigned_get_url(
self,
key: str,
*,
expires_in: int = DEFAULT_PRESIGN_TTL, # noqa: ARG002 -- no TTL local
filename: str | None = None, # noqa: ARG002 -- static mount, no header
) -> str:
"""Return the static-mount URL for *key* (no expiry locally).
*filename* is accepted for interface parity with
:class:`S3Storage` but ignored: the local static mount can't
set ``Content-Disposition``, so dev downloads save under the
uuid key. Acceptable for local-only development.
"""
return f"{self.base_url}/{key}"
[docs]
def put_bytes(
self,
key: str,
*,
blob: bytes,
content_type: str, # noqa: ARG002 -- type is served by extension
) -> None:
"""Write *blob* under *root*, creating parent dirs as needed."""
path = self.root / key
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(blob)
[docs]
def delete(self, key: str) -> None:
"""Delete *key*; idempotent, matching S3's ``DeleteObject``."""
(self.root / key).unlink(missing_ok=True)
[docs]
def default_storage() -> S3Storage | LocalStorage:
"""Build the storage backend from ``FSH_*`` env vars.
Returns an :class:`S3Storage` when ``FSH_S3_BUCKET`` is set (the
production path; infra injects this -- see the ECS task env).
Otherwise returns a :class:`LocalStorage` *only when local-disk is
explicitly opted into* (``FSH_LOCAL_STORAGE_DIR`` /
``FSH_LOCAL_STORAGE_URL``), and raises otherwise -- so a prod task
that's missing its bucket config fails loudly instead of silently
writing to ephemeral container disk. Reports and attachments
share this one backend and selection rule.
S3 (``FSH_S3_BUCKET`` set):
* ``FSH_S3_REGION`` -- AWS region; optional, falls back to the
boto3 default chain.
* ``FSH_S3_ENDPOINT_URL`` -- override for MinIO / localstack /
non-AWS S3-compatible endpoints; optional.
Local (opt-in; ``FSH_S3_BUCKET`` unset):
* ``FSH_LOCAL_STORAGE_DIR`` -- on-disk root; defaults to a
``fsh_storage`` dir under the system temp dir when only the URL
is given.
* ``FSH_LOCAL_STORAGE_URL`` -- URL prefix the app serves it at;
defaults to ``/_files`` (see :func:`mount_local_storage`).
Raises:
RuntimeError: When neither S3 nor an explicit local backend is
configured -- the misconfigured-deployment guard.
"""
storage = _resolve_storage()
if storage is None:
msg = (
"No object storage configured: set FSH_S3_BUCKET "
"(production / MinIO), or opt into local-disk dev storage "
"with FSH_LOCAL_STORAGE_DIR / FSH_LOCAL_STORAGE_URL."
)
raise RuntimeError(msg)
return storage
def _resolve_storage() -> S3Storage | LocalStorage | None:
"""Build the backend from ``FSH_*`` env, or ``None`` if unconfigured."""
bucket = os.environ.get("FSH_S3_BUCKET")
if bucket:
return S3Storage(
bucket=bucket,
region=os.environ.get("FSH_S3_REGION"),
endpoint_url=os.environ.get("FSH_S3_ENDPOINT_URL"),
)
dir_env = os.environ.get("FSH_LOCAL_STORAGE_DIR")
url_env = os.environ.get("FSH_LOCAL_STORAGE_URL")
if not (dir_env or url_env):
return None
root = (
Path(dir_env)
if dir_env
else Path(tempfile.gettempdir()) / "fsh_storage"
)
return LocalStorage(root=root, base_url=url_env or "/_files")
[docs]
def mount_local_storage(app: FastAPI) -> None:
"""Serve local-disk storage over HTTP when S3 isn't configured."""
storage = _resolve_storage()
if not isinstance(storage, LocalStorage):
return
storage.root.mkdir(parents=True, exist_ok=True)
mount_path = urlsplit(storage.base_url).path or "/"
@app.put(f"{mount_path.rstrip('/')}/{{key:path}}", status_code=204)
async def _put_local_file(key: str, request: Request) -> Response:
storage.put_bytes(
key,
blob=await request.body(),
content_type=request.headers.get(
"content-type", "application/octet-stream"
),
)
return Response(status_code=status.HTTP_204_NO_CONTENT)
app.mount(
mount_path,
StaticFiles(directory=str(storage.root)),
name="fsh-local-files",
)
# --- Action request/response schemas --------------------------------------
[docs]
class UploadRequest(BaseModel):
"""Body for the request-upload action.
Carries everything :func:`~fsh_lib.files.request_upload` needs to reserve
a key and bind the presigned PUT URL to the right content type.
"""
filename: str
content_type: str
size_bytes: int
[docs]
class UploadResponse(BaseModel):
"""Response for the request-upload action.
The client PUTs the file bytes to ``upload_url`` (it must send
a matching ``Content-Type`` header), then calls the
complete-upload action with ``id`` to flip the row out of
pending state.
"""
id: uuid.UUID
upload_url: str
[docs]
class DownloadResponse(BaseModel):
"""Response for the download action -- a short-lived GET URL."""
download_url: str
# --- Action functions -----------------------------------------------------
[docs]
async def request_upload(
*,
model_cls: type[FileMixin],
db: AsyncSession,
body: UploadRequest,
) -> UploadResponse:
"""Reserve a key and return a presigned PUT URL.
The row is created with ``uploaded_at=NULL``; the client
confirms the actual byte upload via :func:`complete_upload`.
*model_cls* is supplied by the action handler, which detects
the ``type[FileMixin]`` annotation and passes the resource's
mapped class. No per-resource factory binding needed --
consumers point a resource's ``action`` config at this function
directly.
"""
key = f"{uuid.uuid4().hex}_{_sanitize_filename(body.filename)}"
file_id = (
await db.execute(
insert(model_cls)
.values(
s3_key=key,
content_type=body.content_type,
size_bytes=body.size_bytes,
original_filename=body.filename,
)
.returning(model_cls.id)
)
).scalar_one()
storage = default_storage()
upload_url = storage.presigned_put_url(
key,
content_type=body.content_type,
)
return UploadResponse(
id=file_id,
upload_url=upload_url,
)
[docs]
async def complete_upload(
file: FileMixin,
*,
db: AsyncSession,
) -> None:
"""Mark *file* as uploaded.
Returns ``None`` so the action op emits 204 No Content -- a
completed upload has no useful body to return; the client
already knows the id.
Issues a Core ``UPDATE`` rather than mutating the loaded ORM
instance, so the persistence path is identical regardless of
whether the caller's session has autoflush quirks. Idempotent
-- calling twice just refreshes the timestamp.
"""
cls = type(file)
await db.execute(
update(cls)
.where(cls.id == file.id)
.values(uploaded_at=datetime.datetime.now(tz=datetime.UTC)),
)
[docs]
async def download(
file: FileMixin,
*,
db: AsyncSession, # noqa: ARG001 -- action handler passes this
) -> DownloadResponse:
"""Return a presigned GET URL for *file*.
Refuses with 404 when ``uploaded_at is None`` -- the row exists
but the client never confirmed the PUT, so the object may not
be in S3 and a presigned URL would just 404 noisily.
"""
if file.uploaded_at is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="File upload not complete",
)
storage = default_storage()
return DownloadResponse(
download_url=storage.presigned_get_url(
file.s3_key,
filename=file.original_filename,
),
)
[docs]
async def delete_file(
file: FileMixin,
*,
db: AsyncSession,
) -> None:
"""Cascade-delete *file*: remove the S3 object then the row.
Returns ``None`` so the action op emits 204 No Content -- the
client doesn't need a body to know the row is gone.
S3 first because :meth:`S3Storage.delete` is idempotent -- a
crash between the two steps leaves an orphan row, which the
next delete attempt cleans up. Reversing the order would
instead leak S3 objects, which are harder to find later.
"""
storage = default_storage()
storage.delete(file.s3_key)
cls = type(file)
await db.execute(delete(cls).where(cls.id == file.id))