Batch import

This guide builds the counterpart to FastAPI integration: not a service that waits for requests, but a script that works through a data set once and then exits. It imports invoices into enaio:

One folder per invoice year, created on first use.
One register per month inside that folder.
One document per invoice, with the invoice file attached.
Every step keyed on a business key, so a second run updates instead of duplicating.
One bad row is logged and counted, the run continues.

A batch importer is the classic case for the SyncPoolClient. The work is sequential, one invoice after the other, so there is nothing for an event loop to interleave, and async/await would only add noise. The async client earns its keep where many operations wait at the same time, as in a web service.

1. Installation

uv add ecmind-blue-client

pip install ecmind-blue-client

Everything else the importer needs is in the standard library: csv for the input, pathlib for the files, collections.Counter for the statistics, logging for the protocol.

2. Input data

The importer reads a CSV whose paths in the file column are resolved relative to the CSV itself, so data and files can be moved together:

invoice_number,invoice_date,supplier,amount,file
4711,2024-03-01,Example Supplier AG,1234.50,files/invoice-4711.txt
4712,2024-03-14,Example Supplier AG,87.00,files/invoice-4712.txt

Column Meaning

Column	Meaning
`invoice_number`	Business key of the invoice. Identifies the document on the server, so it must be unique.
`invoice_date`	Invoice date in ISO format. Determines the year folder and the month register.
`supplier`	Supplier name, optional.
`amount`	Invoice amount, optional.
`file`	Path to the invoice file, relative to the CSV.

invoice_number

Business key of the invoice. Identifies the document on the server, so it must be unique.

invoice_date

Invoice date in ISO format. Determines the year folder and the month register.

supplier

Supplier name, optional.

amount

Invoice amount, optional.

file

Path to the invoice file, relative to the CSV.

3. Model classes

Three object types, one per level of the hierarchy. In a real project these are generated with ecm-generate-models; written out by hand they look like this:

# models.py
from datetime import date

from ecmind_blue_client.ecm.model import (
    ECMDocumentModel,
    ECMField,
    ECMFolderModel,
    ECMRegisterModel,
)


class InvoiceFolder(ECMFolderModel):
    _internal_name_ = "InvoiceFolder"

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


class InvoiceRegister(ECMRegisterModel):
    _internal_name_ = "InvoiceRegister"

    Name: ECMField[str] = ECMField(str, mandatory=True)


class InvoiceDocument(ECMDocumentModel):
    _internal_name_ = "InvoiceDocument"

    Title: ECMField[str] = ECMField(str, mandatory=True)
    InvoiceNumber: ECMField[str] = ECMField(str, default=None)
    Supplier: ECMField[str] = ECMField(str, default=None)
    InvoiceDate: ECMField[date] = ECMField(date, default=None)
    Amount: ECMField[float] = ECMField(float, default=None)

4. Configuration and start-up

Credentials belong in the environment, not in the script. The pool is created once and lives for the whole run, and the timezone of the installation is set before the first job, because the invoice date is a timestamp the server computes in its own zone:

import logging
import os

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

logging.basicConfig(level=logging.INFO, format="%(levelname)-7s %(name)s: %(message)s")
log = logging.getLogger("batch-import")

set_server_timezone(os.environ.get("ECMIND_BLUE_SERVER_TIMEZONE", "Europe/Zurich"))

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

A sequential importer keeps one connection busy, so the pool default (pool_size=10) is more than enough. For a run that takes hours, keepalive_interval=300 stops idle connections from being dropped by a firewall in between, see Keepalive in the pool.

client.close() belongs in a finally block. It stops the background workers and closes the idle connections, which matters as soon as the importer runs from a scheduler that expects the process to exit cleanly.

5. Folder and register per invoice

Both levels are created with an upsert() keyed on their own business key, so the importer does not have to know whether they already exist. The ID is cached per run, which keeps the second invoice of the same month from asking the server again:

def folder_for_year(ecm, year: int, cache: dict[int, int]) -> int:
    """Return the ID of the folder for one invoice year, creating it if needed."""
    if year in cache:
        return cache[year]

    title = f"Invoices {year}"
    object_id, _, _, action = (
        ecm.dms.upsert(InvoiceFolder(Title=title, Year=year))
        .search(InvoiceFolder.Title == title)
        .execute()
    )
    log.info("folder %r: %s (id %s)", title, action, object_id)
    cache[year] = object_id
    return object_id


def register_for_month(ecm, folder_id: int, month_key: str, cache: dict[str, int]) -> int:
    """Return the ID of the month register inside a year folder, creating it if needed."""
    if month_key in cache:
        return cache[month_key]

    object_id, _, _, action = (
        ecm.dms.upsert(InvoiceRegister(Name=month_key), folder_id=folder_id)
        .search(InvoiceRegister.Name == month_key)
        .execute()
    )
    log.info("register %r: %s (id %s)", month_key, action, object_id)
    cache[month_key] = object_id
    return object_id

The search key has to be unique on its own

The <Search> section of an upsert is built from the .search() conditions alone. folder_id and register_id place the new object, they do not narrow the search.

A register named March would therefore match the March register of every other year as well, and the default .action_multiple("ERROR") would abort the row rather than update a register in the wrong folder. Hence month_key carries the year: 2024-03.

upsert() takes the numeric IDs for folder_id and register_id. insert() and insert_and_get() also accept a model instance and read the ID from it.

6. The document with its file

The document is filed in the register of its month and brings the invoice file along. .files(…, replace=True) means a second run replaces the file instead of appending a second copy:

from datetime import date
from pathlib import Path

from ecmind_blue_client.rpc import JobRequestFileFromPath


def import_document(ecm, row: dict[str, str], base_dir: Path, folder_id: int, register_id: int) -> str:
    """Upsert one invoice document with its file and return the action the server performed."""
    file_path = (base_dir / row["file"]).resolve()
    if not file_path.is_file():
        raise OSError(f"file not found: {file_path}")

    document = InvoiceDocument(
        Title=f"Invoice {row['invoice_number']}",
        InvoiceNumber=row["invoice_number"],
        Supplier=row["supplier"] or None,
        InvoiceDate=date.fromisoformat(row["invoice_date"]),
        Amount=float(row["amount"]) if row["amount"] else None,
    )

    _, _, _, action = (
        ecm.dms.upsert(document, folder_id=folder_id, register_id=register_id)
        .search(InvoiceDocument.InvoiceNumber == row["invoice_number"])
        .files([JobRequestFileFromPath(file_path)], replace=True)
        .execute()
    )
    return action

JobRequestFileFromPath reads the file when the job is sent, so nothing is buffered in memory beforehand. Content that is already in memory goes through JobRequestFileFromBytes(data, extension).

The returned action is what the server actually did, "INSERT" or "UPDATE", and it is the honest source for the run statistics. Anyone who needs a strict insert instead adds .action1("NONE") and skips rows that already exist, see upsert().

7. One bad row must not stop the run

An importer that dies on record 700 of 1000 is worse than one that reports 999 successes and one failure. Every row therefore runs in its own try block, and only errors that concern a single row are caught:

from ecmind_blue_client.ecm import ECMException

# ECMException covers everything the server refuses, ValueError the client-side checks (mandatory
# and read-only fields) plus a malformed date or amount, KeyError a missing CSV column, and OSError
# an unreadable file.
ROW_ERRORS = (ECMException, ValueError, KeyError, OSError)

counts: Counter[str] = Counter()

for line_no, row in enumerate(csv.DictReader(handle), start=2):
    label = row.get("invoice_number") or f"line {line_no}"
    try:
        invoice_date = date.fromisoformat(row["invoice_date"])
        folder_id = folder_for_year(ecm, invoice_date.year, folders)
        month_key = f"{invoice_date.year}-{invoice_date.month:02d}"
        register_id = register_for_month(ecm, folder_id, month_key, registers)

        action = import_document(ecm, row, base_dir, folder_id, register_id)
        counts[action.lower()] += 1
        log.info("invoice %s: %s", label, action)
    except ROW_ERRORS as error:
        counts["failed"] += 1
        log.warning("invoice %s failed: %s", label, error)

What deliberately stays uncaught is everything that makes the rest of the run pointless: a missing environment variable, an unreachable server, a CSV that is not there. Those fail immediately, with a traceback, which is exactly what a scheduler should see.

The exit code turns the statistics into something a cron job or a pipeline can act on:

log.info(
    "done: %s inserted, %s updated, %s failed",
    counts["insert"],
    counts["update"],
    counts["failed"],
)
return 1 if counts["failed"] else 0

8. Running

export ECM_SERVERS="enaio.example.com:4000:1"
export ECM_USERNAME="<user>"
export ECM_PASSWORD="<password>"
export ECMIND_BLUE_SERVER_TIMEZONE="Europe/Zurich"

uv run batch_import.py

INFO    batch-import: folder 'Invoices 2024': INSERT (id 4711)
INFO    batch-import: register '2024-03': INSERT (id 4712)
INFO    batch-import: invoice 4711: INSERT
INFO    batch-import: invoice 4712: INSERT
INFO    batch-import: done: 2 inserted, 0 updated, 0 failed

Run it a second time and every line reports UPDATE, with nothing created twice. That is the point of keying each level on a business key, and it is what makes the importer safe to retry after a partial failure.

9. Complete example

The script below is shipped ready to run in the repository as examples/batch_import.py, together with the model classes in examples/models.py and the sample data in examples/invoices.csv:

"""Batch importer: file invoice files into enaio, driven by a CSV."""

import csv
import logging
import os
import sys
from collections import Counter
from datetime import date
from pathlib import Path

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 import InvoiceDocument, InvoiceFolder, InvoiceRegister

log = logging.getLogger("batch-import")

ROW_ERRORS = (ECMException, ValueError, KeyError, OSError)


def folder_for_year(ecm, year: int, cache: dict[int, int]) -> int:
    """Return the ID of the folder for one invoice year, creating it if needed."""
    if year in cache:
        return cache[year]

    title = f"Invoices {year}"
    object_id, _, _, action = (
        ecm.dms.upsert(InvoiceFolder(Title=title, Year=year))
        .search(InvoiceFolder.Title == title)
        .execute()
    )
    log.info("folder %r: %s (id %s)", title, action, object_id)
    cache[year] = object_id
    return object_id


def register_for_month(ecm, folder_id: int, month_key: str, cache: dict[str, int]) -> int:
    """Return the ID of the month register inside a year folder, creating it if needed.

    The search section of an upsert is built from the .search() conditions alone, so month_key
    carries the year as well ("2024-03"). A bare month name would match the register of every
    other year, and the default .action_multiple("ERROR") would then abort the row.
    """
    if month_key in cache:
        return cache[month_key]

    object_id, _, _, action = (
        ecm.dms.upsert(InvoiceRegister(Name=month_key), folder_id=folder_id)
        .search(InvoiceRegister.Name == month_key)
        .execute()
    )
    log.info("register %r: %s (id %s)", month_key, action, object_id)
    cache[month_key] = object_id
    return object_id


def import_document(ecm, row: dict[str, str], base_dir: Path, folder_id: int, register_id: int) -> str:
    """Upsert one invoice document with its file and return the action the server performed."""
    file_path = (base_dir / row["file"]).resolve()
    if not file_path.is_file():
        raise OSError(f"file not found: {file_path}")

    document = InvoiceDocument(
        Title=f"Invoice {row['invoice_number']}",
        InvoiceNumber=row["invoice_number"],
        Supplier=row["supplier"] or None,
        InvoiceDate=date.fromisoformat(row["invoice_date"]),
        Amount=float(row["amount"]) if row["amount"] else None,
    )

    _, _, _, action = (
        ecm.dms.upsert(document, folder_id=folder_id, register_id=register_id)
        .search(InvoiceDocument.InvoiceNumber == row["invoice_number"])
        .files([JobRequestFileFromPath(file_path)], replace=True)
        .execute()
    )
    return action


def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(levelname)-7s %(name)s: %(message)s")

    set_server_timezone(os.environ.get("ECMIND_BLUE_SERVER_TIMEZONE", "Europe/Zurich"))

    csv_path = Path(os.environ.get("ECM_IMPORT_CSV", Path(__file__).with_name("invoices.csv")))
    base_dir = csv_path.parent

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

    counts: Counter[str] = Counter()
    folders: dict[int, int] = {}
    registers: dict[str, int] = {}

    try:
        with csv_path.open(newline="", encoding="utf-8") as handle:
            # start=2 so the number in a log line matches the line in the file, header included
            for line_no, row in enumerate(csv.DictReader(handle), start=2):
                label = row.get("invoice_number") or f"line {line_no}"
                try:
                    invoice_date = date.fromisoformat(row["invoice_date"])
                    folder_id = folder_for_year(ecm, invoice_date.year, folders)
                    month_key = f"{invoice_date.year}-{invoice_date.month:02d}"
                    register_id = register_for_month(ecm, folder_id, month_key, registers)

                    action = import_document(ecm, row, base_dir, folder_id, register_id)
                    counts[action.lower()] += 1
                    log.info("invoice %s: %s", label, action)
                except ROW_ERRORS as error:
                    # One bad row must not end the run: log it, count it, keep going.
                    counts["failed"] += 1
                    log.warning("invoice %s failed: %s", label, error)
    finally:
        client.close()

    log.info(
        "done: %s inserted, %s updated, %s failed",
        counts["insert"],
        counts["update"],
        counts["failed"],
    )
    return 1 if counts["failed"] else 0


if __name__ == "__main__":
    sys.exit(main())

Getting started for the individual steps this importer combines
upsert() for action control, files, and table fields
ecm-generate-models for generating the model classes
Timezone of the installation for dates and timestamps
Keepalive in the pool for long-running imports
Checking server reachability for a check before the first job
FastAPI integration for the asynchronous counterpart