ecmind-blue-client 1.0.0a2

A client wrapper for blue

ECMind Blue Client

A Python client library for the Blue server by ECMind. Communication with the server uses a proprietary TCP/RPC binary protocol.

Deprecation warning

• ecmind_blue_client.com_client is no longer available.
• ecmind_blue_client.soap_client is no longer available.
• TcpClient and TcpPoolClient are deprecated and superseded by SyncPoolClient / AsyncPoolClient.

Installation

Using uv:

uv add ecmind_blue_client

Using pip:

pip install ecmind_blue_client

Available extras:

Extra	Description
development	Dev tooling: black, isort, pylint, pytest, pytest-cov, build, setuptools-scm, twine
manage	(removed) Previously pulled in ecmind-blue-client-manage
objdef	(removed) Previously pulled in ecmind-blue-client-objdef
portfolio	(removed) Previously pulled in ecmind-blue-client-portfolio
workflow	(removed) Previously pulled in ecmind-blue-client-workflow
tcp	(deprecated) Only required for the deprecated TcpClient and TcpPoolClient

High-Level ECM API (ecm/)

The recommended way to interact with the ECM. The entire API is available as both synchronous and asynchronous variants. The ECM() factory function returns either ECMSync or ECMAsync depending on the client type passed.

Setup

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

client = SyncPoolClient(
    servers=[ServerConnectionSettings(hostname="<host>", port=4000)],
    username="<username>",
    password="<password>",
    name="MyApp",
)

ecm = ECM(client)

For async code, use AsyncPoolClient instead — ECM() will then return an ECMAsync instance with identical await-based methods.

Object related operations (ecm.dms)

Accessed via ecm.dms, this namespace covers all operations on folders, registers, and documents.

Object types can be defined as typed model classes (recommended) or created generically at runtime using the factory functions make_folder_model, make_register_model, and make_document_model — useful when the object type is only known at runtime.

Typed model classes bring a practical advantage in day-to-day development: because all fields are declared as typed attributes, IDEs such as VS Code offer full code completion for field names and their expected data types. More importantly, when the ECM object definition changes — for example when an internal field name is renamed on the server — regenerating the model class causes all affected references across the codebase to be immediately flagged by the IDE or type checker. This makes it straightforward to locate and update every call site without relying on text search.

Model generator

Typed model classes can be generated automatically from a live server or a local asobjdef XML file using the ecm-generate-models command, which is installed alongside the package:

# Generate from a live server
ecm-generate-models --host <host> --username <username> --password <password> --output-dir ./models

# Generate from a local asobjdef XML file
ecm-generate-models --file asobjdef.xml --output-dir ./models

# Generate only a specific cabinet
ecm-generate-models --host <host> --username <username> --password <password> --cabinet MyCabinet --output-dir ./models

# Print to stdout instead of writing files
ecm-generate-models --host <host> --username <username> --password <password>

Replace <host>, <username> and <password> with the actual hostname and login credentials for the target server. Each cabinet produces one .py file in --output-dir containing ready-to-use model classes. SSL is enabled by default; use --no-ssl to disable it. The default port is 4000.

Typed model class (recommended) — definition:

from ecmind_blue_client.ecm.model import ECMFolderModel, ECMField, ECMTableField, ECMTableRowModel

class InvoiceRow(ECMTableRowModel):
    Amount: ECMField[float]
    Description: ECMField[str]

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

Typed model class — query with where clauses:

results = ecm.dms.select(InvoiceFolder).where(
    InvoiceFolder.Title == "Invoice 2024",
    (InvoiceFolder.Year >= 2020) & (InvoiceFolder.Year <= 2024),
).order_by(InvoiceFolder.Year.DESC).execute()

for folder in results:
    print(folder.system.id, folder.Title, folder.Year)

Generic model — definition:

from ecmind_blue_client.ecm.model import make_folder_model

InvoiceFolder = make_folder_model("InvoiceFolder")

Generic model — query with where clauses:

results = ecm.dms.select(InvoiceFolder).where(
    InvoiceFolder["Title"] == "Invoice 2024",
    InvoiceFolder["Year"] >= 2020,
).execute()

for folder in results:
    print(folder.system.id, folder["Title"], folder["Year"])

Inserting and updating objects:

# Insert and immediately retrieve the created object
folder = ecm.dms.insert_and_get(InvoiceFolder(Title="Invoice 2024", Year=2024))
print(folder.system.id)

# Update an existing object by its system ID
folder.Title = "Updated Title"
ecm.dms.update(folder)

Upsert (insert-or-update):

object_id, type_id, hits, action = (
    ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
    .search(InvoiceFolder.Title == "Invoice 2024")
    .execute()
)

Deleting objects:

ecm.dms.delete(folder)

Streaming large result sets:

for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.Title)

# Async variant
async for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.Title)

Security operations (ecm.security)

User and group management is accessed via ecm.security:

# Roles of the currently logged-in user
roles = ecm.security.roles()

# All users
users = ecm.security.users(extended_info=True)

# Groups for a specific user
groups = ecm.security.groups(user_guid="<guid>")

User impersonation

impersonate executes subsequent requests in the security context of another user. All operations performed on the returned instance are treated by the server as if that user had issued them directly — applied rights, audit trail entries, and access restrictions all reflect the target user rather than the authenticated connection user.

The connecting user must hold the system role SERVER_SWITCH_JOB_CONTEXT (ECMSystemRole.SERVER_SWITCH_JOB_CONTEXT, role ID 72) for the server to accept the context switch. Without this role the server will reject the request with an error.

with ecm.impersonate("john") as ecm_john:
    folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

The instance can also be used without a with block when no automatic cleanup is needed:

ecm_john = ecm.impersonate("john")
folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

Low-Level RPC API (rpc/)

The RPC layer provides direct TCP socket access to the Blue server. It is the foundation the high-level ECM API is built on. Use it directly only when you need access to server jobs not yet covered by the ECM API.

Connection and job execution

from ecmind_blue_client.pool import SyncPoolClient, ServerConnectionSettings
from ecmind_blue_client.rpc import Jobs

client = SyncPoolClient(
    servers=[ServerConnectionSettings(hostname="<host>", port=4000)],
    username="<username>",
    password="<password>",
)

result = client.execute(Jobs.KRN_GETSERVERINFO, Flags=0, Info=6)
print(result.get("Value", str))

JobResult

The execute() call returns a JobResult:

Property	Description
result.get(name, type)	Retrieve a typed output parameter
result.files	List of JobResponseFile output file attachments
result.result_code	Server result code (0 = success)
result.error_messages	Server error string, or None on success

Session lifecycle

The pool clients manage the full session lifecycle automatically:

1. krn.SessionAttach — establish session
2. krn.SessionLogin — authenticate
3. ECM operations
4. krn.SessionLogout — close session

SSL/TLS is enabled by default. Pass use_ssl=False or a custom cadata PEM string to SyncPoolClient / AsyncPoolClient to override.

Load balancing

Both SyncPoolClient and AsyncPoolClient support weighted load balancing across multiple servers:

from ecmind_blue_client.pool import SyncPoolClient, ServerConnectionSettings

client = SyncPoolClient(
    servers=[
        ServerConnectionSettings(hostname="server1", port=4000, weight=2),
        ServerConnectionSettings(hostname="server2", port=4000, weight=1),
    ],
    username="<username>",
    password="<password>",
    pool_size=10,
)