FastAPI integration

This guide shows how to embed the ecmind_blue_client into an asynchronous web service built on FastAPI. The service exposes a REST API over an enaio document type:

  • POST /documents creates a document, or updates it if it already exists (upsert).

  • GET /documents lists documents page by page (paging).

  • GET /documents/{object_id} returns a document’s metadata and fields.

  • GET /documents/{object_id}/content downloads the document file.

  • DELETE /documents/{object_id} deletes a document.

In an event-loop context such as FastAPI, always use the AsyncPoolClient. The connection pool is created once at startup and reused for the whole lifetime, not recreated per request.

1. Installation

  • uv

  • pip

uv add ecmind-blue-client fastapi uvicorn
pip install ecmind-blue-client fastapi uvicorn

2. Model class and input schema

The service needs two classes: a typed ECM model class for the document type and a Pydantic schema for the JSON input when creating a document.

The model class can be declared statically (full IDE support) or generated with ecm-generate-models. Here is the static variant:

# models.py
from datetime import date
from pydantic import BaseModel
from ecmind_blue_client.ecm.model import ECMDocumentModel, ECMField


class InvoiceDocument(ECMDocumentModel):
    _internal_name_ = "InvoiceDocument"

    InvoiceNumber: ECMField[str] = ECMField(str, mandatory=True)
    Supplier: ECMField[str] = ECMField(str, default=None)
    InvoiceDate: ECMField[date] = ECMField(date, default=None)
    Amount: ECMField[float] = ECMField(float, default=None)


class InvoiceIn(BaseModel):
    """JSON body for creating or updating an invoice."""

    invoice_number: str
    supplier: str | None = None
    invoice_date: date | None = None
    amount: float | None = None

3. Application and lifecycle (lifespan)

The AsyncPoolClient is created once in the lifespan handler and stored in app.state. A dependency hands the ECM instance to the endpoints.

# main.py
import os
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, HTTPException, Query, Request, UploadFile
from fastapi.responses import StreamingResponse

from ecmind_blue_client.ecm import ECM, ECMAsync, ECMNotFoundException
from ecmind_blue_client.pool import AsyncPoolClient
from ecmind_blue_client.rpc import JobRequestFileFromReader

from models import InvoiceDocument, InvoiceIn

# Storage location for new documents (from configuration; depending on the
# schema a register_id may additionally be required).
INBOX_FOLDER_ID = int(os.environ["ECM_INBOX_FOLDER_ID"])


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: build the connection pool. servers format: "<host>:<port>:<weight>".
    client = AsyncPoolClient(
        servers=os.environ["ECM_SERVERS"],
        username=os.environ["ECM_USERNAME"],
        password=os.environ["ECM_PASSWORD"],
    )
    app.state.ecm = ECM(client)  # -> ECMAsync
    yield
    # Shutdown: the pool releases its connections once the client is no longer
    # referenced. There is no explicit close() method.
    app.state.ecm = None


app = FastAPI(title="enaio Document API", lifespan=lifespan)


def get_ecm(request: Request) -> ECMAsync:
    """Provide the shared ECM instance from the app state."""
    return request.app.state.ecm

4. Create or update a document (upsert)

ecm.dms.upsert(…​) runs a server-side search and inserts the object when it does not exist, or updates it otherwise. The search condition defines the business key, here the invoice number. The uploaded file is attached via .files(…​).

@app.post("/documents", status_code=201)
async def upsert_document(
    payload: InvoiceIn,
    file: UploadFile | None = None,
    ecm: ECMAsync = Depends(get_ecm),
):
    model = InvoiceDocument(
        InvoiceNumber=payload.invoice_number,
        Supplier=payload.supplier,
        InvoiceDate=payload.invoice_date,
        Amount=payload.amount,
    )

    builder = (
        ecm.dms.upsert(model, folder_id=INBOX_FOLDER_ID)
        .search(InvoiceDocument.InvoiceNumber == payload.invoice_number)
    )

    if file is not None:
        # Pass the upload reader through directly instead of first loading the
        # whole file into memory.
        extension = (file.filename or "").rsplit(".", 1)[-1] or "bin"
        builder = builder.files([JobRequestFileFromReader(file.file, file.size, extension)])

    object_id, object_type_id, hits, action = await builder.execute()
    return {"id": object_id, "action": action, "hits": hits}

The action field of the response reports what actually happened: "INSERT" for a newly created object, "UPDATE" for an updated one.

If the document type has a table field and its rows should be replaced rather than appended on update, append .replace_table_fields(True) to the builder. See upsert().

5. Paged listing

The list paginates ecm.dms.select(…​) with offset and limit. The page size is driven by query parameters and bounded server-side.

@app.get("/documents")
async def list_documents(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    ecm: ECMAsync = Depends(get_ecm),
):
    offset = (page - 1) * page_size
    results = await (
        ecm.dms.select(InvoiceDocument)
        .order_by(InvoiceDocument.InvoiceDate.DESC)
        .offset(offset)
        .limit(page_size)
        .execute()
    )
    return {
        "page": page,
        "page_size": page_size,
        "count": len(results),
        "items": [
            {
                "id": doc.system.id,
                "invoice_number": doc.InvoiceNumber,
                "supplier": doc.Supplier,
                "invoice_date": doc.InvoiceDate,
                "amount": doc.Amount,
            }
            for doc in results
        ],
    }

.execute() loads the requested page fully into memory, which is fine for a bounded page size. For very large, unpaged scans prefer .stream() (an async generator) instead. Paging through .stream() is not transactional.

6. Detail lookup

ecm.dms.get(…​) loads a single object by its ID. With base_params=True and file_properties=True it also retrieves audit metadata and file properties.

@app.get("/documents/{object_id}")
async def get_document(object_id: int, ecm: ECMAsync = Depends(get_ecm)):
    try:
        doc = await ecm.dms.get(
            InvoiceDocument,
            object_id,
            base_params=True,
            file_properties=True,
        )
    except ECMNotFoundException:
        raise HTTPException(status_code=404, detail="Document not found")

    base = doc.system.base_params
    file_props = doc.system.file_properties
    return {
        "id": doc.system.id,
        "fields": {
            "invoice_number": doc.InvoiceNumber,
            "supplier": doc.Supplier,
            "invoice_date": doc.InvoiceDate,
            "amount": doc.Amount,
        },
        "metadata": {
            "creator": base.creator,
            "creation_date": base.creation_date,
            "modifier": base.modifier,
            "modified_date": base.modified_date,
            "version": base.version,
        },
        "file": {
            "extension": file_props.extension,
            "size": file_props.size,
            "mimetype": file_props.mimetype,
            "page_count": file_props.documentpagecount,
        },
    }

7. Download

To avoid loading the whole file into memory, it is read in blocks with ecm.dms.document_stream(…​) and passed straight to the HTTP client through a StreamingResponse. The total size from the file properties bounds the reading cleanly at the end of the file.

@app.get("/documents/{object_id}/content")
async def download_document(object_id: int, ecm: ECMAsync = Depends(get_ecm)):
    doc = await ecm.dms.get(InvoiceDocument, object_id, file_properties=True)
    props = doc.system.file_properties
    if props is None or props.count == 0:
        raise HTTPException(status_code=404, detail="No file available")

    async def stream_file():
        offset = 0
        chunk_size = 256 * 1024
        while offset < props.size:
            chunk = await ecm.dms.document_stream(
                object_id, offset=offset, length=min(chunk_size, props.size - offset)
            )
            if not chunk:
                break
            yield chunk
            offset += len(chunk)

    filename = f"{doc.InvoiceNumber}.{props.extension}"
    return StreamingResponse(
        stream_file(),
        media_type=props.mimetype or "application/octet-stream",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )

If a full retrieval is needed instead (multiple files, server-side conversion to PDF or TIFF, burnt-in annotations), ecm.dms.files(object_id, convert=StoreInCacheByIdConversion.PDF) returns a list[JobResponseFile]. Their .store(path) method writes straight to disk without going through bytes().

8. Delete

ecm.dms.delete(…​) moves the object to the recycle bin by default (soft delete). When passing the ID instead of a model instance, give the object type as the second argument.

@app.delete("/documents/{object_id}", status_code=204)
async def delete_document(object_id: int, ecm: ECMAsync = Depends(get_ecm)):
    await ecm.dms.delete(object_id, "InvoiceDocument")

For permanent deletion set hard_delete=True, and to remove a sub-structure as well set delete_cascading=True.

9. Running

The configuration can be provided through environment variables or a .env file.

  • Environment variables

  • .env file

export ECM_SERVERS="enaio.example.com:4000:1"
export ECM_USERNAME="<user>"
export ECM_PASSWORD="<password>"
export ECM_INBOX_FOLDER_ID="<folder-id>"

uvicorn main:app --reload

Write the values into a .env file and pass it to uvicorn at startup:

# .env
ECM_SERVERS=enaio.example.com:4000:1
ECM_USERNAME=<user>
ECM_PASSWORD=<password>
ECM_INBOX_FOLDER_ID=<folder-id>
uvicorn main:app --reload --env-file .env

The interactive API documentation is then available at http://127.0.0.1:8000/docs.

10. Complete example

The following main.py brings all endpoints together (the model classes from models.py above are imported):

# main.py
import os
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, HTTPException, Query, Request, UploadFile
from fastapi.responses import StreamingResponse

from ecmind_blue_client.ecm import ECM, ECMAsync, ECMNotFoundException
from ecmind_blue_client.pool import AsyncPoolClient
from ecmind_blue_client.rpc import JobRequestFileFromReader

from models import InvoiceDocument, InvoiceIn

INBOX_FOLDER_ID = int(os.environ["ECM_INBOX_FOLDER_ID"])


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncPoolClient(
        servers=os.environ["ECM_SERVERS"],
        username=os.environ["ECM_USERNAME"],
        password=os.environ["ECM_PASSWORD"],
    )
    app.state.ecm = ECM(client)
    yield
    app.state.ecm = None


app = FastAPI(title="enaio Document API", lifespan=lifespan)


def get_ecm(request: Request) -> ECMAsync:
    return request.app.state.ecm


@app.post("/documents", status_code=201)
async def upsert_document(
    payload: InvoiceIn,
    file: UploadFile | None = None,
    ecm: ECMAsync = Depends(get_ecm),
):
    model = InvoiceDocument(
        InvoiceNumber=payload.invoice_number,
        Supplier=payload.supplier,
        InvoiceDate=payload.invoice_date,
        Amount=payload.amount,
    )
    builder = (
        ecm.dms.upsert(model, folder_id=INBOX_FOLDER_ID)
        .search(InvoiceDocument.InvoiceNumber == payload.invoice_number)
    )
    if file is not None:
        extension = (file.filename or "").rsplit(".", 1)[-1] or "bin"
        builder = builder.files([JobRequestFileFromReader(file.file, file.size, extension)])
    object_id, object_type_id, hits, action = await builder.execute()
    return {"id": object_id, "action": action, "hits": hits}


@app.get("/documents")
async def list_documents(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    ecm: ECMAsync = Depends(get_ecm),
):
    offset = (page - 1) * page_size
    results = await (
        ecm.dms.select(InvoiceDocument)
        .order_by(InvoiceDocument.InvoiceDate.DESC)
        .offset(offset)
        .limit(page_size)
        .execute()
    )
    return {
        "page": page,
        "page_size": page_size,
        "count": len(results),
        "items": [
            {
                "id": doc.system.id,
                "invoice_number": doc.InvoiceNumber,
                "supplier": doc.Supplier,
                "invoice_date": doc.InvoiceDate,
                "amount": doc.Amount,
            }
            for doc in results
        ],
    }


@app.get("/documents/{object_id}")
async def get_document(object_id: int, ecm: ECMAsync = Depends(get_ecm)):
    try:
        doc = await ecm.dms.get(
            InvoiceDocument, object_id, base_params=True, file_properties=True
        )
    except ECMNotFoundException:
        raise HTTPException(status_code=404, detail="Document not found")
    base = doc.system.base_params
    file_props = doc.system.file_properties
    return {
        "id": doc.system.id,
        "fields": {
            "invoice_number": doc.InvoiceNumber,
            "supplier": doc.Supplier,
            "invoice_date": doc.InvoiceDate,
            "amount": doc.Amount,
        },
        "metadata": {
            "creator": base.creator,
            "creation_date": base.creation_date,
            "modifier": base.modifier,
            "modified_date": base.modified_date,
            "version": base.version,
        },
        "file": {
            "extension": file_props.extension,
            "size": file_props.size,
            "mimetype": file_props.mimetype,
            "page_count": file_props.documentpagecount,
        },
    }


@app.get("/documents/{object_id}/content")
async def download_document(object_id: int, ecm: ECMAsync = Depends(get_ecm)):
    doc = await ecm.dms.get(InvoiceDocument, object_id, file_properties=True)
    props = doc.system.file_properties
    if props is None or props.count == 0:
        raise HTTPException(status_code=404, detail="No file available")

    async def stream_file():
        offset = 0
        chunk_size = 256 * 1024
        while offset < props.size:
            chunk = await ecm.dms.document_stream(
                object_id, offset=offset, length=min(chunk_size, props.size - offset)
            )
            if not chunk:
                break
            yield chunk
            offset += len(chunk)

    filename = f"{doc.InvoiceNumber}.{props.extension}"
    return StreamingResponse(
        stream_file(),
        media_type=props.mimetype or "application/octet-stream",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


@app.delete("/documents/{object_id}", status_code=204)
async def delete_document(object_id: int, ecm: ECMAsync = Depends(get_ecm)):
    await ecm.dms.delete(object_id, "InvoiceDocument")