s6.utils.infra

Infrastructure gateway server for database and file operations.

Provides a Gateway class that encapsulates both the FastAPI application (with DB CRUD, aggregation, and file-proxy endpoints) and a matching Python client for easy use.

class s6.utils.infra.Gateway(config: Any)

Bases: object

property app: FastAPI

Build and return the FastAPI application with DB CRUD and file proxy routes.

property client: GatewayClient

Return a Python client for this gateway using the same routes for DB and files.

static connect(config: Any)

Context manager that launches a local Gateway server on an ephemeral port and yields a GatewayClient configured to localhost on that port.

Usage:

with Gateway.connect(config) as client:

# use client to call the local server

class s6.utils.infra.GatewayClient(gateway: Gateway, base_url: str | None = None)

Bases: object

create(payload: Dict[str, Any], collection: str | None = None) → str

Insert a document into the (default) collection and return its ID.

get(doc_id: str, collection: str | None = None) → Dict[str, Any]

Fetch a single document by ID from the (default) collection.

query(flt: Dict[str, Any], collection: str | None = None) → Any

Query documents in the (default) collection via a JSON filter.

aggregate(pipeline: list, collection: str | None = None)

Generator that streams documents from an aggregation pipeline via WebSocket on the (default) collection.

update(doc_id: str, payload: Dict[str, Any], collection: str | None = None) → int

Update a document by ID in the (default) collection and return count.

delete(doc_id: str, collection: str | None = None) → None

Delete a single document by ID from the (default) collection.

delete_many(flt: Dict[str, Any], collection: str | None = None) → int

Delete multiple documents in the (default) collection via a JSON filter. Returns the number of documents deleted.

download_file(filename: str) → bytes

upload_file(filename: str, fobj: Any, content_type: str) → Any

post_file(filename: str, fobj: Any, content_type: str) → Any

delete_file(filename: str) → None

upload(obj: Dict[str, Any], collection: str | None = None, **kwargs) → str

Serialize and upload a nested structure to the (default) collection. Returns the inserted document ID. Additional top-level kwargs are merged into the stored record.

download(doc_id: str, collection: str | None = None) → Any

Fetch and reconstruct a nested structure previously uploaded.