Getting started

1. Sync or Async?

The library offers two API variants that differ in how they execute requests:

Sync (SyncPoolClient) Async (AsyncPoolClient)

Blocks the calling thread until the response is received.

Yields control back to the event loop while waiting for the response.

Simple, linear code without async/await.

Requires async def functions and await.

Suitable for scripts, batch processes, and importers.

Suitable for web applications and other event-loop-based systems.

Sync is a good fit wherever code runs sequentially and no concurrent processing is needed — for example in import scripts, migrations, or command-line tools.

Async is the right choice when the client is embedded in an existing event loop, such as FastAPI endpoints. In that context a blocking sync client would stall the entire event loop and delay all concurrent requests.

  • Sync

  • Async

# Import script: SyncPoolClient with 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 endpoint: AsyncPoolClient with 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. Connecting

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

The format for servers is <host>:<port>:<weight>. Multiple servers are separated by #. The weight controls how connections are distributed across servers during load balancing.

  • Sync

  • Async

# Multiple servers with weights
client = SyncPoolClient(
    servers="<host1>:4000:2#<host2>:4000:1",
    username="<username>",
    password="<password>",
)
# Multiple servers with weights
client = AsyncPoolClient(
    servers="<host1>:4000:2#<host2>:4000:1",
    username="<username>",
    password="<password>",
)

4. Querying objects

The API offers two methods for retrieving results:

stream()

Reads results page by page from the server, yielding one object at a time. Recommended for large result sets, as it never holds all objects in memory at once.

execute()

Fetches all results upfront and returns a list. Convenient for small result sets or when the total count is needed before processing.

execute() loads all objects into memory before processing begins. With large result sets this can cause memory problems.

stream() works page by page — but paging in enaio is not transactional. If objects are modified while iterating, pages may contain inconsistent state or objects may appear twice. In that case execute() should be preferred.

4.1. Generic approach

Without a model definition, objects can be queried directly using the internal name of the object type:

  • 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. With a typed model

For IDE code completion and static type checking, defining a model class is recommended. Any changes to the object definition on the server become immediately visible via the type checker.

  • 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 can also be generated automatically from a running ECM instance — see ecm-generate-models.

5. Inserting objects

  • Sync

  • Async

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

6. SQL queries

  • 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. Impersonation

To perform actions on behalf of another user, use impersonate(). The executing user requires the Context Switch system role.

  • Sync

  • Async

# As a context manager
with ecm.impersonate("john") as ecm_john:
    folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

# Or directly without a with block
ecm_john = ecm.impersonate("john")
folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))
# As a context manager
async with ecm.impersonate("john") as ecm_john:
    folder = await ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

# Or directly without a with block
ecm_john = ecm.impersonate("john")
folder = await ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))