Plugin Authoring

Recotem discovers DataSource plugins via Python entry points. A plugin is any installed package that registers in the recotem.datasources group.

The examples/plugins/echo-source/ directory in this repository is a minimal, runnable reference implementation.

Plugin contract

A plugin must provide a class with three class-level attributes and one required method (fetch); __init__ and the optional probe are described below.

from __future__ import annotations

import random
from typing import ClassVar, Literal

import pandas as pd
from pydantic import BaseModel, Field
from recotem.datasource.base import DataSourceError, FetchContext


class EchoSource:
    """Returns a synthetic DataFrame — useful for testing and CI."""

    # 1. type_name: discriminator value matched against the recipe YAML
    #    `source.type` field.  Must be a non-empty string and unique across
    #    all installed plugins.  By convention use a short lower-case slug.
    type_name: ClassVar[str] = "echo"

    # 2. Config: pydantic BaseModel describing the recipe sub-fields for this
    #    source.  All fields appear under `source:` in the YAML alongside the
    #    `type:` discriminator.  Config MUST declare `type` as a Literal field
    #    whose single value equals type_name — recotem builds a pydantic
    #    discriminated union keyed on `type` across every registered plugin,
    #    and the training pipeline reads the field back to resolve the source
    #    class.  Omitting it is not an option: pydantic's default
    #    `extra="ignore"` would silently drop the YAML `type:` key and training
    #    would fail with "Recipe source has no discriminator 'type' field."
    #    `validate_plugin_contract` rejects a Config that omits it, types it as
    #    anything other than `Literal`, or whose Literal disagrees with
    #    type_name.
    class Config(BaseModel):
        type: Literal["echo"] = "echo"
        n_users: int = Field(default=10, ge=1)
        n_items: int = Field(default=20, ge=1)
        n_rows: int = Field(default=100, ge=1)
        seed: int = Field(default=42)

    # 3. extras_required: pip extras to suggest when optional dependencies
    #    are missing.  Leave empty if the plugin has no optional deps.
    extras_required: ClassVar[list[str]] = []

    # 4. no_expand_fields: frozenset of field names inside the source config
    #    whose string values must NEVER receive ${RECOTEM_RECIPE_*} env-var
    #    expansion.  List any fields that carry raw SQL, query parameters, or
    #    other content where ${} should be treated as literals.
    #    Use frozenset() (empty) when no fields need protection beyond the
    #    global baseline (query, query_parameters) that is always guarded.
    #    This attribute is REQUIRED — validate_plugin_contract enforces its
    #    presence and its type (frozenset).  A missing or wrong-type attribute
    #    raises DataSourceError at plugin discovery with a pointer to this doc.
    no_expand_fields: ClassVar[frozenset[str]] = frozenset()

    def __init__(self, config: "EchoSource.Config") -> None:
        self._config = config

    def fetch(self, ctx: FetchContext) -> pd.DataFrame:
        """Return a DataFrame whose columns include those named in
        the recipe `schema` block (user_column, item_column, optional
        time_column).

        Returns a DataFrame with columns: user_id (str), item_id (str),
        timestamp (int epoch seconds).
        """
        cfg = self._config
        max_possible = cfg.n_users * cfg.n_items
        if cfg.n_rows > max_possible:
            raise DataSourceError(
                f"EchoSource: n_rows ({cfg.n_rows}) exceeds n_users * n_items "
                f"({max_possible}).  Reduce n_rows or increase n_users/n_items."
            )
        rng = random.Random(cfg.seed)
        users = [f"user_{i}" for i in range(cfg.n_users)]
        items = [f"item_{j}" for j in range(cfg.n_items)]
        all_pairs = [(u, v) for u in users for v in items]
        sampled = rng.sample(all_pairs, cfg.n_rows)
        base_ts = 1_700_000_000
        rows = [
            {"user_id": u, "item_id": v, "timestamp": base_ts + idx}
            for idx, (u, v) in enumerate(sampled)
        ]
        return pd.DataFrame(rows, columns=["user_id", "item_id", "timestamp"])

    def probe(self) -> None:
        """Optional. Called by recotem validate to test connectivity.

        Should be cheap — never load full data.
        Raise DataSourceError on failure.
        Return value is ignored by recotem (Protocol declares -> None).
        """
        cfg = self._config
        max_possible = cfg.n_users * cfg.n_items
        if cfg.n_rows > max_possible:
            raise DataSourceError(
                f"EchoSource: n_rows ({cfg.n_rows}) exceeds n_users * n_items "
                f"({max_possible})."
            )
        # discarded by recotem validate — kept here for illustration only
        return {"status": "ok", "rows_to_emit": cfg.n_rows, "items": cfg.n_items}  # type: ignore[return-value]

Rules

1. type_name is the discriminator value. It appears as source.type: echo in the recipe. The registry validates that it is a non-empty string and unique across all loaded plugins; duplicate type_name values are reported with both conflicting fully-qualified class names. recotem train and recotem validate exit 2, not 3 — plugin discovery runs inside recipe loading, so the registry's DataSourceError is re-raised as a RecipeError. recotem serve does not exit: it logs recipe_load_error_skipped and keeps running with the affected recipes unloaded.

2. Config is a pydantic BaseModel. Fields are validated at recipe load. Use pydantic validators for constraints. Required fields without defaults cause a RecipeError when missing from the recipe.

   Config must declare the discriminator field type: Literal["<type_name>"] = "<type_name>", matching the class's type_name exactly. Recotem assembles every registered Config into a pydantic discriminated union keyed on type (build_source_config_union), and recotem.training.pipeline reads the field back to resolve the source class. validate_plugin_contract raises DataSourceError at plugin-discovery time when the field is missing, is not a typing.Literal, or carries a value that disagrees with type_name. Because discovery happens inside recipe loading, that error is wrapped into a RecipeError and the process exits 2 — the same code as the extra="ignore" mistake below, not the 3 a data-source failure produces.

   Do not rely on pydantic's default extra="ignore" to absorb the YAML type: key instead. That combination loads the recipe successfully but drops the discriminator, and training then fails with Recipe source has no discriminator 'type' field. (exit code 2).

3. extras_required is purely documentation. The registry only validates that it is a list[str]; recotem never auto-installs or auto-checks these extras. Surface a helpful message yourself in __init__ (see Deferred imports) — the value of the attribute is what you cite there.

4. no_expand_fields is required and must be a frozenset[str]. It names every field in the source Config whose string values must never receive ${RECOTEM_RECIPE_*} environment-variable expansion. validate_plugin_contract checks that this attribute is present and is a frozenset; a missing or wrong-type declaration raises DataSourceError at plugin-discovery time with a pointer to this doc.

   • For most plugins, declare no_expand_fields: ClassVar[frozenset[str]] = frozenset() — the global baseline (query, query_parameters) is already guarded unconditionally by the recipe loader.
   • For plugins with SQL or parameterised-query fields, list them explicitly: no_expand_fields: ClassVar[frozenset[str]] = frozenset({"sql", "bind_params"}). This provides defence-in-depth and documents the security intent for future maintainers.

5. fetch(ctx) must return a pandas.DataFrame. When the source backs the recipe's top-level source block, the DataFrame must contain at least the columns referenced in recipe.schema (user_column, item_column, and optionally time_column). The training pipeline accesses those columns by name immediately after fetch — a missing column surfaces as a KeyError and exits the train run.

   The recipe.schema rule applies to the interaction source only. The same registry also serves features.item.source / features.user.source, where the required columns are instead that side's id_column and every declared columns[].name. A plugin needs no special handling for this — FetchContext carries no interaction-specific fields, so any registered source can serve as a feature table — but do not hard-code an assumption that a user_column / item_column will be wanted.

6. fetch() must raise DataSourceError for any external or transient failure (auth errors, network errors, query errors, empty results). DataSourceError is mapped to exit code 3. Any other exception raised from __init__ or fetch() is wrapped by Recotem and also reported as exit code 3 — by train as Data fetch failed: <exc> (construction happens inside the fetch step, so an __init__ failure reports under that wording too) and, for __init__ only, by validate as DataSource probe failed [source]: DataSource construction failed: <exc>. __init__ is the only hook both commands call: validate never calls fetch() and train never calls probe(), so a failure in either of those reaches one command and not the other. In particular a plugin whose probe() raises but whose fetch() works trains to exit 0 and writes a signed artifact, while recotem validate on the identical recipe reports 3 — so a precondition enforced only in probe() does not gate a training run. Wrapping is therefore about the message, not the exit code: an unwrapped exception reaches the operator as the third-party library's own wording, which names neither the extra nor the credential that is missing. Wrap third-party exceptions explicitly:

   def fetch(self, ctx: FetchContext) -> pd.DataFrame:
       try:
           return self._do_fetch()
       except SomeLibraryError as exc:
           raise DataSourceError(str(exc)) from exc

7. Deferred imports. Do not import optional dependencies at module top-level. Defer to __init__ or fetch():

   def __init__(self, config: "MySource.Config") -> None:
       try:
           import my_optional_dep  # noqa: F401
       except ImportError as exc:
           raise DataSourceError(
               "MySource requires 'recotem[myextra]'. "
               "Install with: pip install 'recotem[myextra]'"
           ) from exc
       self.config = config

   This ensures missing extras produce a clear DataSourceError mentioning the required extra by name. An unwrapped ImportError reports the same exit code 3, but reaches the operator as No module named 'my_optional_dep' — which names neither the extra nor the fix.

Package structure

The reference plugin under examples/plugins/echo-source/ uses this layout:

recotem-echo-source/
├── pyproject.toml
└── src/
    └── recotem_echo/
        ├── __init__.py     # re-exports EchoSource so "recotem_echo:EchoSource" resolves
        └── source.py       # EchoSource class definition

A flatter recotem_echo/__init__.py containing the class directly also works — what matters is that the entry-point string <module>:<class> resolves.

pyproject.toml:

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "recotem-echo-source"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["recotem>=2.0,<3", "pandas>=2.2,<4"]

[project.entry-points."recotem.datasources"]
echo = "recotem_echo:EchoSource"

[tool.hatch.build.targets.wheel]
packages = ["src/recotem_echo"]

The entry-point key (echo) is the name reported in registry log/error messages but is not used as the discriminator — Recotem uses the loaded class's type_name attribute. By convention, keep them the same.

Install and use

uv pip install -e examples/plugins/echo-source/

Verify discovery by running recotem validate against a recipe that uses the plugin — the loader resolves source.type through the entry-point registry and will report Unknown DataSource type 'echo' if the plugin is not installed in the same environment as recotem.

recotem schema includes plugin configs

recotem schema builds the JSON Schema at runtime by constructing a discriminated union of every registered DataSource Config class (including plugin-provided ones) and substituting it into the Recipe model. Plugin Config schemas do appear in the output — this is what makes IDE autocompletion work for source.* fields. The union is assembled via build_source_config_union() at invocation time, so the plugin must be installed in the same Python environment as recotem.

Recipe:

name: echo_test

source:
  type: echo
  n_users: 200
  n_items: 100
  n_rows: 6000    # see the note below before shrinking these
  seed: 42        # optional; omit to use the default seed

schema:
  user_column: user_id
  item_column: item_id
  time_column: timestamp   # EchoSource emits integer epoch-second timestamps
  time_unit: s             # required: a numeric time_column has no implied unit

training:
  algorithms: [TopPop]
  metric: ndcg
  cutoff: 10
  n_trials: 1

output:
  path: ./artifacts/echo_test.recotem

Train:

recotem train recipe.yaml

time_unit is required for a numeric time column

time_unit is mandatory whenever the column named by schema.time_column holds numbers rather than strings or datetimes — omitting it exits 4 with code: time_unit_required. recotem validate does not catch this: it checks the recipe schema and probes the source, but the unit is only needed once rows are parsed, so a recipe missing time_unit validates clean and then fails at recotem train. If your source emits epoch integers, say so in the plugin's README so recipe authors set the unit up front.

Why the row counts are this large

EchoSource samples user/item pairs uniformly at random, so the data carries no signal for a recommender to find: the reported best_score is noise around zero and says nothing about model quality. That matters operationally, because training exits 4 with code: zero_score when the best trial scores exactly 0.0. With a small synthetic dataset that is a real possibility — at n_users: 50 / n_rows: 500 roughly one run in thirty landed on 0.0, varying between otherwise identical runs because split.seed does not control the id ordering irspack derives the held-out set from — the size of the held-out set is fixed, but which interactions land in it is not. The counts above put 511 interactions in the held-out set instead of 31, which makes an all-miss evaluation vanishingly unlikely. Shrink them and the walkthrough becomes intermittently red for reasons that have nothing to do with your plugin.

FetchContext

FetchContext carries metadata that fetch() can optionally use:

@dataclass
class FetchContext:
    recipe_name: str                            # the recipe's name field
    run_id: str                                 # unique ID for this training run (UUID)
    extra: dict[str, Any] = field(default_factory=dict)  # reserved for future use

Most plugins ignore ctx. It is useful for logging and for idempotency keys when fetching from write-heavy sources.

Constraints on fetch()

• Synchronous, returning a single pandas.DataFrame. Generators, Iterator[DataFrame], and async def are not supported — the training pipeline calls fetch(ctx) directly and reads .columns immediately.
• Whole-DataFrame in memory. Recotem trains on the full result set (irspack constructs a sparse matrix from it). For larger-than-memory sources, do the chunking and aggregation inside fetch() and return a pre-aggregated DataFrame (e.g. counts of (user, item) pairs).
• Credentials never come via FetchContext.extra (it is reserved). Read them from environment variables (preferred — works with K8s Secrets, systemd EnvironmentFile, Docker --env-file) or from recipe-declared Config fields (but never accept secrets in YAML — reference an env var via ${RECOTEM_RECIPE_*} instead).

Item metadata loading

If your plugin's recipe uses item_metadata, the metadata is loaded by recotem.metadata.loader.load_item_metadata. Failures surface as MetadataError (not DataSourceError) so they are distinguishable from source-fetch failures. The exception carries a .cause attribute indicating the failure origin:

.cause	Meaning
"http_fetch"	HTTP/HTTPS fetch failed (SSRF guard, byte cap, sha256 mismatch). __cause__ is HttpFetchError.
"parse"	File could not be parsed as the declared type (CSV/Parquet).
"field_missing"	A required field is absent and on_field_missing="error".
"io"	Local or object-store read failed.
"unknown"	Catch-all for unexpected failures.

The loader accepts an optional recipe_name= keyword argument. When provided, the recipe name is threaded into HTTP fetcher log context so that redirect and byte-cap log events (e.g. metadata_source_redirect) are correlated with the recipe that triggered the load. This is set automatically by the watcher; you only need it when calling load_item_metadata directly (e.g. in tests).

Compatibility

The plugin contract is part of the recotem 2.x public surface. Pin recotem>=2.0,<3 in your plugin's pyproject.toml — the type_name / Config / fetch(ctx) shape is stable within a major version. The probe() hook may gain optional parameters in a future minor release; use **kwargs: Any if you want to be future-proof.

The entry-point key in [project.entry-points."recotem.datasources"] is informational only (used in error messages); the discriminator is the class's type_name. If two installed plugins both declare type_name = "csv", recotem train and recotem validate exit 2 with both fully-qualified class names, and recotem serve keeps running with the affected recipes skipped — uninstall one or rename its type_name.

Validation in recotem validate

recotem validate recipes/my_recipe.yaml instantiates the source class (which exercises the __init__ deferred-import / extras check) but does not call fetch(). If the source defines an optional probe() method, recotem validate calls it for a lightweight connectivity / auth check:

def probe(self) -> dict:
    """Optional. Called by recotem validate to test connectivity.

    Should be cheap (LIMIT 1, dry-run, fs.exists, ...) — never load full data.
    Raise DataSourceError on failure.  Return a small status dict that
    recotem validate logs (e.g. {"status": "ok", "rows_to_emit": n_rows}).
    """
    ...

When probe() is defined, recotem validate reports DataSource: probe OK (<type_name>); when it is not, it reports DataSource: extras OK (<type_name>, no probe defined). The builtin CSVSource / ParquetSource use fsspec exists(), and BigQuerySource uses a dry-run query job.

Feature sources are probed too. A recipe with a features: block has its features.item.source / features.user.source probed the same way as the top-level source, and each reported line carries a [<where>] label — [features.item.source] / [features.user.source] — so a failure names which source failed. If your plugin can be used as a feature table, keep probe() cheap enough to run several times per recotem validate invocation.

probe_columns() — the schema-column check

recotem train rejects a recipe naming a column the data does not have with a DataSourceError (exit 3). recotem validate asks the same question for the top-level source only, via a second optional hook:

def probe_columns(self, ctx: FetchContext) -> bool:
    """Optional. Called by recotem validate with the recipe's schema columns.

    ctx.extra carries user_column / item_column / time_column, exactly as
    fetch() receives them.  Implement this only when the column list is
    cheap to obtain — a CSV header row, a Parquet footer schema — never by
    running the query or downloading the body.

    Return True when the check ran, False when this configuration cannot
    answer cheaply.  Raise DataSourceError when a required column is absent.
    """
    ...

recotem validate prints one of three distinct lines, so it never claims a check that did not run:

Return	Line
True	Schema columns: OK (<type_name>) [<where>]
False	Schema columns: not checked (<type_name> cannot list columns without a full fetch) [<where>]
hook absent	Schema columns: not checked (<type_name> has no header-only column probe; verified at train time) [<where>]

A raised DataSourceError is reported as Schema column check failed [<where>]: <error> and exits 3, matching train.

Feature sources (features.item.source / features.user.source) are not column-checked — a feature table legitimately does not carry the interaction columns. BigQuerySource and SQLSource do not implement the hook either, because their column set is only known once the query runs.

Exit codes a plugin can actually produce

Measured by installing deliberately broken plugins — one violation per package, each with its own recotem.datasources entry point — into a bare pip install recotem environment and running the real CLI.

What your plugin does	train	validate
works	0	0
Config omits the type discriminator	2	2
type is str rather than Literal	2	2
type Literal value disagrees with type_name	2	2
no_expand_fields missing	2	2
no_expand_fields is a set, not a frozenset	2	2
type_name collides with another installed plugin	2	2
__init__ raises DataSourceError	3	3
__init__ raises any other exception	3	3
probe() raises any exception	0 ‡	3
probe() raises HttpFetchError (or wraps one)	0 ‡	7
fetch() raises DataSourceError	3	0 †
fetch() raises any other exception	3	0 †
fetch() returns something that is not a DataFrame	3	0 †
fetch() omits a column named in schema:	3	0 †
fetch() raises HttpFetchError (or wraps one)	7	0 †

† validate never calls fetch(). ‡ train never calls probe().

Four things follow that are easy to get wrong:

• Every contract violation is exit 2, not 3. Plugin discovery runs inside recipe loading, so the registry's DataSourceError is re-raised as a RecipeError. Exit 3 is for a source that loaded and then failed to produce data.
• Nothing your plugin does produces exit 1. An unwrapped exception is wrapped by Recotem on whichever command reaches it and reports 3. The one code that escapes the 2/3 split upward is 7, which a HttpFetchError keeps through the __cause__ chain — so an SSRF-guard refusal inside a plugin still reports 7 rather than being flattened. Note that the structured code field on that failure is still datasource_error: the 7 comes from the __cause__ chain, so an operator grepping code will not find a separate value for it.
• train and validate agree on every failure they can both reach. The rows they disagree on are exactly the rows one of them never executes.
• probe() is not a gate on train. Putting a check only in probe() buys a validate failure and nothing else — a plugin whose probe() raises but whose fetch() works trains successfully and writes a signed artifact. A precondition that must stop a training run has to be enforced in __init__ or in fetch().

One broken plugin breaks every recipe on the host

Discovery is eager across the whole recotem.datasources entry-point group, not lazy per source.type. A recipe that names a completely different, valid source fails too:

Recipe '.../ok.yaml' source: plugin source discovery failed for type 'ok':
DataSource plugin 'MismatchSource' ... declares Config.type as Literal['something_else'],
which does not match its type_name 'mismatch'.

Nothing in that recipe refers to MismatchSource. A contract violation in any installed plugin is therefore a host-wide outage for train and validate, and the error names the offending plugin class rather than the recipe you ran — read the class name in the message, not the file path.

Under serve the same fault is invisible to a naive probe

serve is deliberately lenient: one malformed recipe is not allowed to take down a server hosting others. Measured with a colliding type_name and a single recipe directory:

$ curl -s http://127.0.0.1:8080/v1/health
{"status":"ok","total":0,"loaded":0,"skipped":1}

HTTP 200, "status": "ok", and zero recipes loaded. The process is alive, answering, and serving nothing; the log carries recipe_load_error_skipped and recipes_directory_loaded_lenient. A liveness or readiness check that reads only the status code — or only the status field — reports healthy. Alert on loaded and skipped, not on the process being up.

Testing

Test fetch() directly without the CLI:

from recotem_echo import EchoSource
from recotem.datasource.base import FetchContext

source = EchoSource(EchoSource.Config(n_users=20, n_items=50, n_rows=200))
ctx = FetchContext(recipe_name="test", run_id="abc")
df = source.fetch(ctx)
assert {"user_id", "item_id", "timestamp"}.issubset(df.columns)
assert len(df) == 200

Use recotem.recipe.load_recipe in integration tests to confirm the full YAML → Recipe → DataSource path. recipe.source is an instance of the plugin's Config model:

from recotem.recipe import load_recipe
from recotem_echo import EchoSource

recipe = load_recipe("tests/fixtures/echo_recipe.yaml")
assert isinstance(recipe.source, EchoSource.Config)

Plugin trust

DANGER

Third-party DataSource plugins run with full process privileges. A malicious plugin can read env vars including RECOTEM_SIGNING_KEYS and RECOTEM_API_KEYS. Pin plugin versions, hash-pin via your lock file, and review source code before deploying. See Security — Plugin trust.