Schnellstart

1. Sync oder Async?

Die Bibliothek bietet zwei API-Varianten, die sich in der Ausführungsweise unterscheiden:

Sync (SyncPoolClient) Async (AsyncPoolClient)

Blockiert den aufrufenden Thread bis die Antwort vorliegt.

Gibt die Kontrolle an den Event Loop zurück, während auf die Antwort gewartet wird.

Einfacher, linearer Code ohne async/await.

Erfordert async def-Funktionen und await.

Geeignet für Skripte, Batch-Prozesse und Importer.

Geeignet für Web-Anwendungen und andere Event-Loop-basierte Systeme.

Sync eignet sich überall dort, wo der Code sequenziell läuft und keine parallele Verarbeitung nötig ist — zum Beispiel in Importskripten, Migrationen oder Kommandozeilen-Werkzeugen.

Async ist die richtige Wahl, wenn der Client in einen bestehenden Event Loop eingebunden wird, etwa in FastAPI-Endpunkten. In diesem Fall würde ein blockierender Sync-Client den gesamten Event Loop blockieren und damit alle parallelen Anfragen verzögern.

  • Sync

  • Async

# Importskript: SyncPoolClient mit make_folder_model
from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.pool import SyncPoolClient
from ecmind_blue_client.ecm.model import make_folder_model

ecm = ECM(SyncPoolClient(servers="<host>:4000:1", username="<user>", password="<pass>"))
InvoiceFolder = make_folder_model("InvoiceFolder")

for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.system.id, folder["Title"])
# FastAPI-Endpunkt: AsyncPoolClient mit make_folder_model
from fastapi import FastAPI
from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.pool import AsyncPoolClient
from ecmind_blue_client.ecm.model import make_folder_model

app = FastAPI()
ecm = ECM(AsyncPoolClient(servers="<host>:4000:1", username="<user>", password="<pass>"))
InvoiceFolder = make_folder_model("InvoiceFolder")

@app.get("/folders")
async def list_folders():
    return [
        {"id": folder.system.id, "title": folder["Title"]}
        async for folder in ecm.dms.select(InvoiceFolder).stream()
    ]

2. Installation

  • uv

  • pip

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

3. Verbindung herstellen

  • Sync

  • Async

from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.pool import SyncPoolClient

client = SyncPoolClient(
    servers="<host>:4000:1",
    username="<username>",
    password="<password>",
)
ecm = ECM(client)
from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.pool import AsyncPoolClient

client = AsyncPoolClient(
    servers="<host>:4000:1",
    username="<username>",
    password="<password>",
)
ecm = ECM(client)

Das Format für servers ist <host>:<port>:<gewichtung>. Mehrere Server werden durch # getrennt. Die Gewichtung bestimmt die Verteilung der Verbindungen beim Load-Balancing.

  • Sync

  • Async

# Mehrere Server mit Gewichtung
client = SyncPoolClient(
    servers="<host1>:4000:2#<host2>:4000:1",
    username="<username>",
    password="<password>",
)
# Mehrere Server mit Gewichtung
client = AsyncPoolClient(
    servers="<host1>:4000:2#<host2>:4000:1",
    username="<username>",
    password="<password>",
)

4. Objekte abfragen

Die API bietet zwei Methoden zum Abrufen von Ergebnissen:

stream()

Liest die Ergebnisse seitenweise vom Server und liefert jedes Objekt einzeln. Empfohlen für große Ergebnismengen, da nie alle Objekte gleichzeitig im Speicher gehalten werden.

execute()

Ruft alle Ergebnisse vollständig ab und gibt eine Liste zurück. Praktisch bei kleinen Ergebnismengen oder wenn die Gesamtanzahl vorab benötigt wird.

execute() lädt alle Objekte vollständig in den Speicher, bevor die Verarbeitung beginnt. Bei großen Ergebnismengen kann dies zu Speicherproblemen führen.

stream() hingegen arbeitet seitenbasiert — das Paging in enaio ist nicht transaktional. Werden während des Iterierens Objekte verändert, kann es vorkommen, dass Seiten inkonsistente Zustände liefern oder Objekte doppelt erscheinen. In diesem Fall sollte execute() bevorzugt werden.

4.1. Generischer Ansatz

Ohne Modelldefinition können Objekte direkt über den internen Namen des Objekttyps abgefragt werden:

  • Sync

  • Async

from ecmind_blue_client.ecm.model import make_folder_model

InvoiceFolder = make_folder_model("InvoiceFolder")

for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.system.id, folder["Title"])
from ecmind_blue_client.ecm.model import make_folder_model

InvoiceFolder = make_folder_model("InvoiceFolder")

async for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.system.id, folder["Title"])

4.2. Mit typisiertem Model

Für Code-Completion und Typprüfung in der IDE empfiehlt sich die Definition eines Models. Änderungen an der Objektdefinition auf dem Server werden durch den Typchecker direkt sichtbar.

  • Sync

  • Async

from ecmind_blue_client.ecm.model import ECMFolderModel, ECMField

class InvoiceFolder(ECMFolderModel):
    _internal_name_ = "InvoiceFolder"
    Title: ECMField[str]
    Year:  ECMField[int]

for folder in (
    ecm.dms.select(InvoiceFolder)
    .where(InvoiceFolder.Year >= 2024)
    .order_by(InvoiceFolder.Year.DESC)
    .stream()
):
    print(folder.system.id, folder.Title)
from ecmind_blue_client.ecm.model import ECMFolderModel, ECMField

class InvoiceFolder(ECMFolderModel):
    _internal_name_ = "InvoiceFolder"
    Title: ECMField[str]
    Year:  ECMField[int]

async for folder in (
    ecm.dms.select(InvoiceFolder)
    .where(InvoiceFolder.Year >= 2024)
    .order_by(InvoiceFolder.Year.DESC)
    .stream()
):
    print(folder.system.id, folder.Title)

Models können auch automatisch aus einer laufenden ECM-Instanz generiert werden — siehe ecm-generate-models.

5. Objekte anlegen

  • Sync

  • Async

folder = ecm.dms.insert_and_get(InvoiceFolder(Title="Rechnung 2024", Year=2024))
print(folder.system.id)
folder = await ecm.dms.insert_and_get(InvoiceFolder(Title="Rechnung 2024", Year=2024))
print(folder.system.id)

6. SQL-Abfragen

  • Sync

  • Async

result = ecm.db.select(
    "SELECT id, benutzer FROM benutzer WHERE benutzer = %s",
    "admin",
)
for row in result:
    print(row["benutzer"])
result = await ecm.db.select(
    "SELECT id, benutzer FROM benutzer WHERE benutzer = %s",
    "admin",
)
for row in result:
    print(row["benutzer"])

7. Benutzerkontext wechseln

Um Aktionen im Namen eines anderen Benutzers auszuführen, wird impersonate() verwendet. Der ausführende Benutzer benötigt dafür die Systemrolle Kontextwechsel.

  • Sync

  • Async

# Als Context-Manager
with ecm.impersonate("john") as ecm_john:
    folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

# Oder direkt ohne with-Block
ecm_john = ecm.impersonate("john")
folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))
# Als Context-Manager
async with ecm.impersonate("john") as ecm_john:
    folder = await ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

# Oder direkt ohne with-Block
ecm_john = ecm.impersonate("john")
folder = await ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))