Getting started

This page takes a project from an empty directory to a script that reads and writes objects: installation, the choice between the sync and the async variant, the connection, typed models, a select(), a direct SQL query, an upsert(), a document with its file, a context switch to another user, and error handling.

Every code example is available in both variants. Pick the tab that matches your project and stay in it, the tabs keep their selection for the rest of the page.

Prerequisites

  • Python >= 3.12

  • uv or pip

  • Host name, port (default 4000) and credentials for the enaio server

  • TCP access from the machine running the code to that server

1. Installation

  • uv

  • pip

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

1.1. A new project from scratch

uv init creates the project skeleton, uv add resolves the dependency and writes it to pyproject.toml and uv.lock:

uv init invoice-import
cd invoice-import
uv add ecmind-blue-client

Directory afterwards:

invoice-import/
├── main.py
├── pyproject.toml
├── README.md
└── uv.lock

The virtual environment is created by the first uv command that needs it, so nothing has to be activated by hand. Run the project with uv run:

uv run main.py
uv add pins the resolved versions in uv.lock. Commit that file so every machine and every build installs exactly the same versions.

2. 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: linear code, no event loop
ecm = ECM(SyncPoolClient(servers="<host>:4000:1", username="<user>", password="<pass>"))

for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.system.id, folder.Title)
# FastAPI endpoint: inside an existing event loop
ecm = ECM(AsyncPoolClient(servers="<host>:4000:1", username="<user>", password="<pass>"))

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

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>, and multiple servers are separated by . The weight controls how connections are spread across them, so "<host1>:4000:2<host2>:4000:1" sends twice as many connections to the first server as to the second.

TLS encryption (use_ssl)

SyncPoolClient and AsyncPoolClient default to use_ssl=True — the enaio® server has required encrypted TCP connections for many versions. Leave this default untouched.

use_ssl=False is reserved for legacy enaio versions without TLS support and is considered deprecated. In any current production environment, use_ssl must always be True.

The client is created once and reused for the entire runtime, never per operation. A short script may leave the pool to garbage collection. A long-running process should shut it down explicitly with client.close() (sync) or await client.aclose() (async), which stops the background workers and closes every idle connection. For the equivalent in a web service see FastAPI integration; to keep idle connections alive through firewalls and load balancers see Keepalive in the pool.

4. Creating the models

A model class describes one object type and provides typed field access, code completion, and static type checking. ecm-generate-models ships with the library and writes those classes straight from the object definition on the server, one file per cabinet:

uv run ecm-generate-models \
    --host enaio.example.com \
    --username admin \
    --password secret \
    --cabinet Invoices \
    --output-dir ./models
Written: models/Invoices.py

The generated file contains a class per object type, an ECMField per index field, and an enum per list catalog field:

class InvoiceFolder(ECMFolderModel):
    _internal_name_ = "InvoiceFolder"

    Title: ECMField[str] = ECMField(str, mandatory=True)
    Year: ECMField[int] = ECMField(int, default=None)
    Status: ECMField[InvoiceFolder_StatusEnum] = ECMField(InvoiceFolder_StatusEnum, default=None)

Regenerate the file whenever the schema on the server changes, and never edit it by hand. Every argument of the command, including generating from an exported asobjdef XML instead of a live server, is described in ecm-generate-models.

Where generation is not an option, a model class can also be written by hand, or built at runtime from the internal name alone with make_folder_model() and its siblings. Both are described in ECM model.

5. Querying objects

ecm.dms.select() returns a query builder: .where() filters, .order_by() sorts, and the query runs on .stream() or .execute().

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.

  • Sync

  • Async

from models.Invoices import InvoiceFolder

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

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

.execute() returns the same hits as a list instead, which is handy when the total count is needed up front: folders = ecm.dms.select(InvoiceFolder).where(InvoiceFolder.Year == 2024).execute(), with await in front of the chain in the async variant.

Multiple conditions in one .where() call are combined with AND, and & and | combine them explicitly. Both operators bind tighter than == in Python, so every comparison needs its own parentheses, and or / and are no substitute: they would drop a condition and are therefore rejected with a TypeError.

query = ecm.dms.select(InvoiceFolder)

query.where((InvoiceFolder.Year == 2024) | (InvoiceFolder.Year == 2025))   # OR group
query.where(InvoiceFolder.Year == 2024 | InvoiceFolder.Year == 2025)       # TypeError
query.where((InvoiceFolder.Year == 2024) or (InvoiceFolder.Year == 2025))  # TypeError

All builder methods, including paging, field restriction, and full-text search: select().

6. Reading and changing a single object

select() searches, get() fetches one known object by its ID:

  • Sync

  • Async

folder = ecm.dms.get(InvoiceFolder, 4711)
print(folder.Title)
folder = await ecm.dms.get(InvoiceFolder, 4711)
print(folder.Title)

An ID that does not exist raises ECMNotFoundException.

A loaded instance tracks its own changes, so update() sends only the fields that really changed:

  • Sync

  • Async

folder.Title = "Invoice 2024 (corrected)"
print(folder.system.is_modified)       # True
print(folder.system.modified_fields)   # {'Title': 'Invoice 2024 (corrected)'}

ecm.dms.update(folder)
folder.Title = "Invoice 2024 (corrected)"
print(folder.system.is_modified)       # True
print(folder.system.modified_fields)   # {'Title': 'Invoice 2024 (corrected)'}

await ecm.dms.update(folder)

When nothing has changed, update() skips the server call altogether, and force=True sends it anyway. update_and_get() performs the update and returns the refreshed instance in one call. How change tracking works in detail: ECM model.

7. Querying the database directly

ecm.dms covers archive objects and ecm.security covers users and groups. Everything beyond that, above all aggregates, joins, and administrative tables, is reachable with SQL over ado.ExecuteSQL:

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

Values belong in placeholders (%s for strings, %d for integers, %u for identifiers), never in an f-string, because the server substitutes them with quoting and thereby prevents SQL injection. A row is read as a raw string with row["name"] or converted with row.typed("name", int). All placeholders, the column metadata, and the return type: db.select().

8. Creating and updating objects

8.1. Inserting

insert_and_get() creates the object and returns it fully populated, including the assigned ID. Where the new object goes depends on its level:

Level Base class Filed with

Folder

ECMFolderModel

nothing, a folder is the top level

Register

ECMRegisterModel

folder_id, plus register_id inside another register

Document

ECMDocumentModel

folder_id, plus register_id when it is filed in a register

  • Sync

  • Async

from datetime import date

folder = ecm.dms.insert_and_get(InvoiceFolder(Title="Invoice 2024", Year=2024))
document = ecm.dms.insert_and_get(
    InvoiceDocument(Title="Invoice 4711", InvoiceDate=date(2024, 3, 1)),
    folder_id=folder,
)
print(folder.system.id, document.system.id)
from datetime import date

folder = await ecm.dms.insert_and_get(InvoiceFolder(Title="Invoice 2024", Year=2024))
document = await ecm.dms.insert_and_get(
    InvoiceDocument(Title="Invoice 4711", InvoiceDate=date(2024, 3, 1)),
    folder_id=folder,
)
print(folder.system.id, document.system.id)

A document may sit directly in a folder, as here, or in one of its registers via register_id. insert() and insert_and_get() accept either a numeric ID or the model instance itself for both, so a fresh object can be passed straight on as the parent of the next one. upsert() takes the numeric IDs only, hence folder.system.id below.

8.2. Upserting

upsert() leaves the decision between insert and update to the server: .search() defines the conditions for the duplicate check, and the server inserts when nothing matches and updates when exactly one object matches. That is what makes an import script safe to run twice.

.execute() returns the object ID, the object type ID, the number of hits, and the action the server actually performed:

  • Sync

  • Async

object_id, type_id, hits, action = (
    ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
    .search(InvoiceFolder.Title == "Invoice 2024")
    .execute()
)
print(object_id, action)  # e.g. 4711 INSERT
object_id, type_id, hits, action = await (
    ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
    .search(InvoiceFolder.Title == "Invoice 2024")
    .execute()
)
print(object_id, action)  # e.g. 4711 INSERT

.execute_and_get() fetches the resulting object back from the server and returns the populated model instead of the tuple:

  • Sync

  • Async

folder = (
    ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
    .search(InvoiceFolder.Title == "Invoice 2024")
    .execute_and_get()
)
print(folder.system.id, folder.Title)
folder = await (
    ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
    .search(InvoiceFolder.Title == "Invoice 2024")
    .execute_and_get()
)
print(folder.system.id, folder.Title)
.search() uses only the field name and the value of each condition. The server always matches for equality, so operators other than == have no effect there.

The action per hit count is configurable, for example insert-only with .action1("NONE"), and documents can carry their file along via .files(). All options: upsert().

Two client-side rules bite on the first write against a real object type:

Mandatory fields

A field declared mandatory=True must carry a value, otherwise insert() and .execute() raise ValueError before anything reaches the server. check_mandatory=False switches the check off.

Read-only fields

A field the object definition marks read-only cannot be written, and attempting it raises ValueError as well. Both rules are described in ECM model.

Date and datetime fields need one decision up front. enaio stores every point in time as an epoch number that it computes in the timezone of the server, and the protocol does not carry that timezone, so the client has to be told once at start-up. Without it, timestamps shift silently whenever the server and the host process sit in different zones:

from ecmind_blue_client import set_server_timezone

set_server_timezone("Europe/Zurich")  # or set ECMIND_BLUE_SERVER_TIMEZONE in the environment

Details and the conversion helpers: Timezone of the enaio installation.

9. Filing documents with their file

Documents are the only objects that carry files. A file is handed over as a JobRequestFile, in one of three flavours:

Class Use for

JobRequestFileFromPath(path)

a file on disk

JobRequestFileFromBytes(data, extension)

content that is already in memory

JobRequestFileFromReader(reader, size, extension)

a stream, for example an upload, without buffering it completely

insert_and_get() takes the files as its second argument, upsert() through .files(), which also decides whether the uploaded files replace the existing ones or are appended:

  • Sync

  • Async

from ecmind_blue_client.rpc import JobRequestFileFromPath

object_id, type_id, hits, action = (
    ecm.dms.upsert(
        InvoiceDocument(Title="Invoice 4711", InvoiceDate=date(2024, 3, 1)),
        folder_id=folder.system.id,
    )
    .search(InvoiceDocument.Title == "Invoice 4711")
    .files([JobRequestFileFromPath("invoice.pdf")], replace=True)
    .execute()
)
from ecmind_blue_client.rpc import JobRequestFileFromPath

object_id, type_id, hits, action = await (
    ecm.dms.upsert(
        InvoiceDocument(Title="Invoice 4711", InvoiceDate=date(2024, 3, 1)),
        folder_id=folder.system.id,
    )
    .search(InvoiceDocument.Title == "Invoice 4711")
    .files([JobRequestFileFromPath("invoice.pdf")], replace=True)
    .execute()
)

Reading them back, files() returns one JobResponseFile per file. Small files stay in memory, larger ones land in a temporary file, and both are read the same way:

  • Sync

  • Async

for response_file in ecm.dms.files(document):
    print(response_file.name, response_file.size())
    response_file.store("/tmp/" + response_file.name)  # write it to disk
    content = response_file.bytes()                    # or take the bytes
for response_file in await ecm.dms.files(document):
    print(response_file.name, response_file.size())
    response_file.store("/tmp/" + response_file.name)  # write it to disk
    content = response_file.bytes()                    # or take the bytes

For files too large to hold in memory, read them in chunks with document_stream() or pass them straight through with files_streaming().

10. Acting on behalf of another user

An importer usually logs in with a technical account, while the objects it files should name the responsible user as their creator, in the object history as well. impersonate() returns a second ECM instance bound to the same pool that adds the context switch to every request, so no second login and no second connection is needed. The executing user requires the Context Switch system role.

The target user can be given as username, user_guid or user_id — exactly one of them. A str is always a user name, never a GUID.

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

# Or by the user's GUID or numeric ID
ecm_john = ecm.impersonate(user_guid="8A1D1F2E4C7B4A9E8F0D3C5B7A9E1D2F")
ecm_john = ecm.impersonate(user_id=42)
# 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"))

As a context manager the switch stays limited to the block, which keeps it visible in the code which operations run under which user. Full description: impersonate() in the API reference.

11. When something goes wrong

Every error the server reports arrives as an exception from one small hierarchy. ECMException is the base class, so catching it catches everything the ECM API raises:

Exception Raised when

ECMNotFoundException

the object, object type, cabinet, register or file does not exist, for example get() with an unknown ID

ECMAccessDeniedException

the logged-in user lacks the rights for the operation

ECMWrongStateException

the operation does not fit the state of the object, for example asking for the active variant of a document that has none

ECMMissingArgumentException

a required identifier is missing, for example a document without a folder or register

ECMException

everything else the server reports, among it unfilled mandatory fields and values a field does not allow

Two errors are raised client-side, before anything is sent: ValueError for a missing mandatory field, a written read-only field or an undecided table-field clear, and TypeError for files on a model that is not a document.

from ecmind_blue_client.ecm import ECMException, ECMNotFoundException

try:
    folder = ecm.dms.get(InvoiceFolder, 4711)
except ECMNotFoundException:
    print("no folder with ID 4711")
except ECMException as error:
    print("the server refused the request:", error)

In the async variant the same block awaits ecm.dms.get(…​), nothing else changes.

The message carries the server’s own text plus a description of the result code behind it, so log it as it is instead of replacing it with a message of your own.

11.1. Seeing what the client does

The library logs through the standard logging module, one logger per module below ecmind_blue_client, and stays silent until the application configures logging. Raising the level shows connection and pool activity, which is usually enough to tell a hanging server from a hanging script:

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("ecmind_blue_client").setLevel(logging.DEBUG)

If nothing works at all, check the servers before the code: Checking server reachability probes every configured server for reachability, login and version.

12. The complete script

The steps above combined into a runnable main.py. The credentials come from the environment, the server timezone is set before the first job, and every server error is caught in one place:

  • Sync

  • Async

import logging
import os
from datetime import date

from ecmind_blue_client import set_server_timezone
from ecmind_blue_client.ecm import ECM, ECMException
from ecmind_blue_client.pool import SyncPoolClient
from ecmind_blue_client.rpc import JobRequestFileFromPath

from models.Invoices import InvoiceDocument, InvoiceFolder, InvoiceRegister

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("invoice-import")


def main() -> None:
    set_server_timezone("Europe/Zurich")

    client = SyncPoolClient(
        servers=os.environ["ECM_SERVERS"],
        username=os.environ["ECM_USERNAME"],
        password=os.environ["ECM_PASSWORD"],
    )
    ecm = ECM(client)

    try:
        # 1. SQL: a report figure that the object API cannot express
        result = ecm.db.select("SELECT COUNT(*) AS anzahl FROM benutzer WHERE aktiv = %d", 1)
        log.info("active users: %s", result.rows[0].typed("anzahl", int))

        # 2. Folder: insert it, or update it if it already exists
        folder = (
            ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
            .search(InvoiceFolder.Title == "Invoice 2024")
            .execute_and_get()
        )
        log.info("folder %s", folder.system.id)

        # 3. Register inside that folder, document with its file inside that register
        register = ecm.dms.insert_and_get(InvoiceRegister(Name="March"), folder_id=folder)
        document = ecm.dms.insert_and_get(
            InvoiceDocument(Title="Invoice 4711", InvoiceDate=date(2024, 3, 1)),
            [JobRequestFileFromPath("invoice.pdf")],
            folder_id=folder,
            register_id=register,
        )
        log.info("document %s with %s file(s)", document.system.id, len(ecm.dms.files(document)))

        # 4. Read: all folders from 2024 onwards, newest first
        for hit in (
            ecm.dms.select(InvoiceFolder)
            .where(InvoiceFolder.Year >= 2024)
            .order_by(InvoiceFolder.Year.DESC)
            .stream()
        ):
            log.info("%s %s %s", hit.system.id, hit.Title, hit.Year)

        # 5. Change one field, only that field goes to the server
        document.Title = "Invoice 4711 (checked)"
        ecm.dms.update(document)

        # 6. File on behalf of john, so the object names him as its creator
        #    (requires the Context Switch system role)
        with ecm.impersonate("john") as ecm_john:
            filed = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Invoice 2024 (john)", Year=2024))
        log.info("filed as john: %s", filed.system.id)

    except ECMException as error:
        log.error("the server refused the request: %s", error)
    finally:
        client.close()


if __name__ == "__main__":
    main()
import asyncio
import logging
import os
from datetime import date

from ecmind_blue_client import set_server_timezone
from ecmind_blue_client.ecm import ECM, ECMException
from ecmind_blue_client.pool import AsyncPoolClient
from ecmind_blue_client.rpc import JobRequestFileFromPath

from models.Invoices import InvoiceDocument, InvoiceFolder, InvoiceRegister

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("invoice-import")


async def main() -> None:
    set_server_timezone("Europe/Zurich")

    client = AsyncPoolClient(
        servers=os.environ["ECM_SERVERS"],
        username=os.environ["ECM_USERNAME"],
        password=os.environ["ECM_PASSWORD"],
    )
    ecm = ECM(client)

    try:
        # 1. SQL: a report figure that the object API cannot express
        result = await ecm.db.select("SELECT COUNT(*) AS anzahl FROM benutzer WHERE aktiv = %d", 1)
        log.info("active users: %s", result.rows[0].typed("anzahl", int))

        # 2. Folder: insert it, or update it if it already exists
        folder = await (
            ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
            .search(InvoiceFolder.Title == "Invoice 2024")
            .execute_and_get()
        )
        log.info("folder %s", folder.system.id)

        # 3. Register inside that folder, document with its file inside that register
        register = await ecm.dms.insert_and_get(InvoiceRegister(Name="March"), folder_id=folder)
        document = await ecm.dms.insert_and_get(
            InvoiceDocument(Title="Invoice 4711", InvoiceDate=date(2024, 3, 1)),
            [JobRequestFileFromPath("invoice.pdf")],
            folder_id=folder,
            register_id=register,
        )
        log.info("document %s with %s file(s)", document.system.id, len(await ecm.dms.files(document)))

        # 4. Read: all folders from 2024 onwards, newest first
        async for hit in (
            ecm.dms.select(InvoiceFolder)
            .where(InvoiceFolder.Year >= 2024)
            .order_by(InvoiceFolder.Year.DESC)
            .stream()
        ):
            log.info("%s %s %s", hit.system.id, hit.Title, hit.Year)

        # 5. Change one field, only that field goes to the server
        document.Title = "Invoice 4711 (checked)"
        await ecm.dms.update(document)

        # 6. File on behalf of john, so the object names him as its creator
        #    (requires the Context Switch system role)
        async with ecm.impersonate("john") as ecm_john:
            filed = await ecm_john.dms.insert_and_get(
                InvoiceFolder(Title="Invoice 2024 (john)", Year=2024)
            )
        log.info("filed as john: %s", filed.system.id)

    except ECMException as error:
        log.error("the server refused the request: %s", error)
    finally:
        await client.aclose()


if __name__ == "__main__":
    asyncio.run(main())
export ECM_SERVERS=enaio.example.com:4000:1 ECM_USERNAME=admin ECM_PASSWORD=secret
uv run main.py

The repository ships both variants ready to run, as examples/quickstart_sync.py and examples/quickstart_async.py.

13. Next steps