FastAPI-Integration

Dieser Leitfaden zeigt, wie der ecmind_blue_client in einen asynchronen Web-Service auf Basis von FastAPI eingebunden wird. Der Service stellt eine REST-API über einem enaio-Dokumenttyp bereit:

  • POST /documents legt ein Dokument an oder aktualisiert es, falls es bereits existiert (Upsert).

  • GET /documents listet Dokumente seitenweise auf (Paging).

  • GET /documents/{object_id} liefert Metadaten und Felder eines Dokuments.

  • GET /documents/{object_id}/content lädt die Dokumentdatei herunter.

  • DELETE /documents/{object_id} löscht ein Dokument.

In einem Event-Loop-Kontext wie FastAPI wird stets der AsyncPoolClient verwendet. Der Verbindungspool wird einmal beim Start aufgebaut und über die gesamte Laufzeit wiederverwendet, nicht pro Anfrage neu erstellt.

1. Installation

  • uv

  • pip

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

2. Modellklasse und Eingabe-Schema

Für den Service werden zwei Klassen benötigt: eine typisierte ECM-Modellklasse für den Dokumenttyp und ein Pydantic-Schema für die JSON-Eingabe beim Anlegen.

Die Modellklasse kann statisch deklariert (volle IDE-Unterstützung) oder mit ecm-generate-models generiert werden. Hier die statische Variante:

# 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-Körper für das Anlegen oder Aktualisieren einer Rechnung."""

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

3. Anwendung und Lebenszyklus (Lifespan)

Der AsyncPoolClient wird im Lifespan-Handler einmal erstellt und in app.state abgelegt. Eine Dependency reicht die ECM-Instanz an die Endpunkte weiter.

# 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

# Ablageort für neue Dokumente (aus der Konfiguration; je nach Schema
# kann zusätzlich ein register_id erforderlich sein).
INBOX_FOLDER_ID = int(os.environ["ECM_INBOX_FOLDER_ID"])


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Verbindungspool aufbauen. servers-Format: "<host>:<port>:<gewicht>".
    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: Der Pool gibt seine Verbindungen frei, sobald der Client
    # nicht mehr referenziert wird. Eine explizite close()-Methode gibt es nicht.
    app.state.ecm = None


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


def get_ecm(request: Request) -> ECMAsync:
    """Stellt die gemeinsame ECM-Instanz aus dem App-Status bereit."""
    return request.app.state.ecm

4. Dokument anlegen oder aktualisieren (Upsert)

ecm.dms.upsert(…​) führt serverseitig eine Suche aus und legt das Objekt an, wenn es nicht existiert, oder aktualisiert es andernfalls. Die Suchbedingung legt den fachlichen Schlüssel fest, hier die Rechnungsnummer. Die hochgeladene Datei wird über .files(…​) angehängt.

@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:
        # Den Upload-Reader direkt durchreichen, statt die Datei zuerst
        # vollständig in den Arbeitsspeicher zu laden.
        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}

Das Feld action der Antwort gibt zurück, was tatsächlich passiert ist: "INSERT" für ein neu angelegtes Objekt, "UPDATE" für ein aktualisiertes.

Enthält der Dokumenttyp ein Tabellenfeld und sollen dessen Zeilen beim Aktualisieren ersetzt statt angehängt werden, kann .replace_table_fields(True) an den Builder angehängt werden. Siehe upsert().

5. Paginiertes Listing

Für die Liste wird ecm.dms.select(…​) mit offset und limit paginiert. Die Seitengröße wird über Query-Parameter gesteuert und serverseitig begrenzt.

@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() lädt die angeforderte Seite vollständig in den Speicher, was bei einer begrenzten Seitengröße unproblematisch ist. Für sehr große, ungepagte Durchläufe ist stattdessen .stream() (asynchroner Generator) vorzuziehen. Das Paging über .stream() ist nicht transaktional.

6. Detailabfrage

ecm.dms.get(…​) lädt ein einzelnes Objekt anhand seiner ID. Mit base_params=True und file_properties=True werden zusätzlich Audit-Metadaten und Datei-Eigenschaften abgerufen.

@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="Dokument nicht gefunden")

    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

Damit die Datei nicht vollständig in den Arbeitsspeicher geladen wird, wird sie mit ecm.dms.document_stream(…​) in Blöcken gelesen und über eine StreamingResponse direkt an den HTTP-Client durchgereicht. Die Gesamtgröße aus den Datei-Eigenschaften begrenzt das Lesen sauber am Dateiende.

@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="Keine Datei vorhanden")

    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}"'},
    )

Wird stattdessen der Vollabruf benötigt (mehrere Dateien, serverseitige Konvertierung nach PDF oder TIFF, eingebrannte Annotationen), liefert ecm.dms.files(object_id, convert=StoreInCacheByIdConversion.PDF) eine list[JobResponseFile]. Deren .store(pfad)-Methode schreibt direkt auf die Platte, ohne den Umweg über bytes().

8. Löschen

ecm.dms.delete(…​) verschiebt das Objekt standardmäßig in den Papierkorb (Soft-Delete). Wird die ID anstelle einer Modell-Instanz übergeben, ist der Objekttyp als zweites Argument anzugeben.

@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")

Für eine dauerhafte Löschung wird hard_delete=True gesetzt, für das Mitlöschen einer Unterstruktur delete_cascading=True.

9. Ausführen

Die Konfiguration kann über Umgebungsvariablen oder eine .env-Datei erfolgen.

  • Umgebungsvariablen

  • .env-Datei

export ECM_SERVERS="enaio.example.com:4000:1"
export ECM_USERNAME="<benutzer>"
export ECM_PASSWORD="<passwort>"
export ECM_INBOX_FOLDER_ID="<ordner-id>"

uvicorn main:app --reload

Werte in eine .env-Datei schreiben und beim Start an uvicorn übergeben:

# .env
ECM_SERVERS=enaio.example.com:4000:1
ECM_USERNAME=<benutzer>
ECM_PASSWORD=<passwort>
ECM_INBOX_FOLDER_ID=<ordner-id>
uvicorn main:app --reload --env-file .env

Die interaktive API-Dokumentation steht anschließend unter http://127.0.0.1:8000/docs bereit.

10. Vollständiges Beispiel

Die folgende main.py fasst alle Endpunkte zusammen (die Modellklassen aus models.py oben werden importiert):

# 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="Dokument nicht gefunden")
    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="Keine Datei vorhanden")

    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")

11. Verwandte Themen