| name | plugin-implementation |
| description | Write or change the code of a DataIntegration task - reaching a Corporate Memory deployment, logging, the plugin icon, declaring ports, honouring cancellation, reporting progress, and writing custom parameter types with autocompletion. Use whenever a WorkflowPlugin or TransformPlugin body, its @Plugin block, its ports or its parameter types are added or edited. |
Implementing a DataIntegration task
These are the conventions the eccenca plugin fleet converged on. They are not
style preferences - each one exists because the obvious alternative behaves
worse inside a running workflow.
Reaching a Corporate Memory deployment
Use cmem-client, declared as
cmem-client = "^1.0.0". Build it from the context you were handed rather than
from configuration:
from cmem_client.client import Client
def execute(self, inputs: Sequence[Entities], context: ExecutionContext) -> Entities | None:
client = Client.from_context(context=context)
Client.from_context() carries the executing user's identity. In tests, where
there is no execution context, use Client.from_env() instead.
Sub-APIs live under the same package, for example
from cmem_client.repositories.graphs import ImportConflictPolicy.
cmempy is deprecated. Do not add imports from cmem.cmempy.*, and do not
use setup_cmempy_user_access(). Plenty of existing plugins still call it -
that is legacy, not a pattern to copy. cmem-plugin-base continues to depend
on cmem-cmempy transitively, which does not make it available for new code.
Logging
The base class already provides a logger as self.log. Use it:
self.log.info("Fetched %s records", count)
Do not create a module logger with logging.getLogger(__name__). self.log is
a PluginLogger that routes into DataIntegration under
plugins.python.<plugin_id>, so its output is visible where an operator looks
for it; a private logger is not.
The icon
Ship an SVG beside the plugin module and reference it by package:
@Plugin(
label="My task",
icon=Icon(file_name="my_task.svg", package=__package__),
...
)
Always package=__package__ rather than a hard-coded package name - it keeps
working when the module moves or the project is renamed.
What the SVG contains matters as much as where it lives, and neither
task check nor plugin discovery says a word about it - a wrong icon is only
found by somebody looking at the workspace.
Give the mark no colour of its own: put fill="currentColor" (or
stroke="currentColor") on the root svg and name no fill on the shapes, so
it follows the surrounding text colour and stays legible on a light and a dark
workspace alike. Hard-coding a brand colour produces the one icon in the task
list that looks broken when the theme changes. Leave the background
transparent - no full-size <rect> - because every other icon in that list is,
and an opaque tile reads as a coloured block among them.
cmem-plugin-parameters and cmem-plugin-pyshacl are the icons to copy the
shape of.
Declaring ports
Say what the task accepts and produces; do not leave it implicit.
input_ports=FixedNumberOfInputs([FixedSchemaPort(schema=MY_SCHEMA)]),
output_port=FixedSchemaPort(schema=MY_SCHEMA),
Use UnknownSchemaPort when the schema is only known at runtime, and
FlexibleNumberOfInputs when the task genuinely accepts any number of inputs.
A task that consumes nothing declares FixedNumberOfInputs([]).
Honouring cancellation
A long-running task must stop when the user cancels the workflow. Check the
status inside the entity loop, and guard the access:
from contextlib import suppress
for entity in inputs[0].entities:
with suppress(AttributeError):
if context.workflow.status() == "Canceling":
break
...
The suppress(AttributeError) is required, not defensive noise:
context.workflow is absent in some contexts - notably the test contexts - and
an unguarded check raises there while working in production.
Reporting progress
Report through the execution context so the workflow UI can show what is
happening:
context.report.update(
ExecutionReport(
entity_count=processed,
operation="write",
operation_desc="entities written",
)
)
operation is a short label. Use read, write, wait or
done; do not invent new verbs, and do not use past tense.operation_desc describes the counted thing in plural, so it reads correctly
after the number: "entities written", "files uploaded".- Update inside the loop, not only once at the end. A report emitted after
the work is finished shows a user nothing while the task is running, which is
exactly when they are looking.
A constructor with six or more parameters
Ruff's PLR0913 and PLR0917 both complain about a long argument list, and a
task's constructor takes one argument per PluginParameter - so its arity is
decided by the configuration surface the task offers, not by a style choice.
Neither escape ruff suggests is available: keyword-only arguments change the
signature DataIntegration instantiates, and folding parameters into one object
breaks the one-argument-per-parameter mapping the framework depends on.
This is the standing exception the lint rules describe, so it needs no fresh
decision. Suppress both codes on the def, with the reason above it:
def __init__(
self,
...
PLR0917 became a stable rule in ruff 0.16, so a task that passed before may
start failing on it without anything in the task changing. Suppress it on the
constructor only - it stays a real finding on an ordinary function, which is
why it is not in the ignore list in pyproject.toml.
Parameters that carry secrets
A password, token or API key is typed, never a plain string:
from cmem_plugin_base.dataintegration.parameter.password import Password, PasswordParameterType
PluginParameter(
name="api_key",
label="API key",
param_type=PasswordParameterType(),
)
The value arrives as a Password; call .decrypt() only where it is used.
Typing it as str puts the secret in plain text in the task configuration and
in the project export.
Custom parameter types
A PluginParameter without an explicit param_type gets one derived from the
constructor's type annotation, and that derivation cannot handle a union.
An annotation like JinjaCode | str or str | Password raises
TypeError: issubclass() arg 1 must be a class while the module is imported,
which aborts discovery for the whole package - every plugin the package
ships disappears from the workspace, task check stays green, and the
traceback only shows up in PluginDiscoveryResult.errors.
So annotate a single type, or pass param_type explicitly. Writing a union is
almost always the moment you needed param_type anyway: cmem-plugin-ssh
declares private_key: str | Password and works only because it also passes
param_type=PasswordParameterType().
Reach for a shipped type first - ChoiceParameterType, GraphParameterType,
DatasetParameterType, PasswordParameterType, and the code, multiline
and resource types. Write your own only when the value is a thing the user
should pick from a list that only your plugin can produce: a folder on a
remote host, a collection in a store, a model offered by an API.
Subclass StringParameterType, not ParameterType directly - the value is
carried as a string and StringParameterType already handles that:
from typing import Any, ClassVar
from cmem_plugin_base.dataintegration.context import PluginContext
from cmem_plugin_base.dataintegration.types import Autocompletion, StringParameterType
class CollectionParameterType(StringParameterType):
"""Autocomplete the collections available on the configured server."""
allow_only_autocompleted_values: bool = True
def autocomplete(
self,
query_terms: list[str],
depend_on_parameter_values: list[Any],
context: PluginContext,
) -> list[Autocompletion]:
"""Return the collections matching all query terms."""
results = [
Autocompletion(value=name, label=f"{title} ({name})")
for name, title in self._fetch(context)
]
if not query_terms:
return results
return [
r
for r in results
if all(term.lower() in (r.label or r.value).lower() for term in query_terms)
]
- An empty
query_terms must return everything. That is the list the user
sees before typing, and returning nothing looks like a broken parameter.
- Match against every term, not any of them - the UI splits what the user typed
on whitespace.
- Return a stable order. Sort at the end; do not deduplicate with
set()
after sorting, because that throws the ordering away again.
The flags
allow_only_autocompleted_values = True makes the parameter a closed
vocabulary: the UI refuses values that did not come from your list. Set it
when a value the server does not know is always an error.autocomplete_value_with_labels = True tells the UI the labels matter and
must be shown instead of the raw values.- Implement
label() when a stored value is not human-readable on its own. A
saved task shows the raw value until label() resolves it, so a task
configured last week displays an opaque id without it.
Depending on other parameters
An autocompletion that needs a hostname, or credentials, declares the
parameters it reads:
autocompletion_depends_on_parameters: ClassVar[list[str]] = [
"hostname",
"port",
"password",
]
Their values arrive in depend_on_parameter_values positionally, in exactly
this order - depend_on_parameter_values[2] is password only because it is
third in the list. Reordering the list silently repoints every index, so read
them once at the top of autocomplete() and unpack by name:
hostname, port, password = depend_on_parameter_values
A dependency that is itself a secret arrives as a Password, not a string, and
needs password.decrypt() before use.
Until every declared parameter has a value, no autocompletion happens at all -
so keep the list to what you genuinely need. Each extra entry is one more field
the user must fill before the list appears.
DataIntegration also clears a dependent parameter's own value whenever one
of the parameters it declares changes, and that propagates: in a chain of
token -> resource -> sub-resource, changing the token clears the resource,
which clears the sub-resource. Two things follow, and both change how such a
chain is designed. A chained parameter never holds a value left over from an
earlier selection, so there is no stale combination to validate against and no
caveat to write into the task documentation about one. And a chained parameter
can safely be made mandatory, because a user who invalidates it is asked to
choose again rather than being stranded on a value they cannot correct.
