diff --git a/docs/developers/external-packs.mdx b/docs/developers/external-packs.mdx new file mode 100644 index 000000000..bc5351391 --- /dev/null +++ b/docs/developers/external-packs.mdx @@ -0,0 +1,391 @@ +--- +title: "External packs (plugins)" +description: "Ship enrichers and data types from your own Python package, outside the Flowsint repository. Use the flowsint.enrichers and flowsint.types entry-point groups to keep private, domain-specific, or independently-versioned packs separate from the core tree while still being auto-discovered at runtime." +category: "Developers" +order: 12 +author: "Flowsint Team" +tags: ["tutorial", "developers", "plugins", "external-enrichers", "external-types", "entry-points"] +version: "1.2.8" +last_updated_at: "2026-06-03" +--- + +## Why external packs + +The [Managing enrichers](/docs/developers/managing-enrichers) and [Managing types](/docs/developers/managing-types) guides show how to add enrichers and types *inside* the Flowsint repository. That is the right place for anything you intend to contribute upstream. But sometimes you want to keep your work **outside** the main tree: + +- **Private sources.** Enrichers that hit a paid, regional, or otherwise non-public data source you don't want to publish. +- **Domain or country packs.** A coherent set of types and enrichers for a specific domain — for example Brazilian `CPF`/`CNPJ` types with their lookups — versioned and released on its own cadence. +- **Clean upstream tracking.** Keeping your additions in a separate distribution means your Flowsint checkout stays pristine and upstream updates merge without conflicts. + +Flowsint supports this through **entry-point discovery**. Your package declares itself under one of two entry-point groups, and Flowsint imports it at startup — firing the same `@flowsint_enricher` and `@flowsint_type` decorators you'd use in-tree. No edits to the Flowsint source are required. + +## How discovery works + +Flowsint discovers built-in enrichers and types by walking the `flowsint_enrichers` and `flowsint_types` package directories. On top of that, `load_all_enrichers()` and `load_all_types()` consult two [entry-point](https://packaging.python.org/en/latest/specifications/entry-points/) groups: + +| Group | Discovered by | Registers into | +|---|---|---| +| `flowsint.enrichers` | `load_all_enrichers()` (`flowsint-enrichers/src/flowsint_enrichers/registry.py`) | `ENRICHER_REGISTRY` | +| `flowsint.types` | `load_all_types()` (`flowsint-types/src/flowsint_types/registry.py`) | `TYPE_REGISTRY` | + +For each entry point in a group, Flowsint imports the module it points at. Importing that module runs your decorated classes, which register themselves in the global registry — exactly as if they lived inside the core tree. The API and the Celery worker both call `load_all_*()` at startup, so a discovered pack is available everywhere. + +Discovery is **idempotent** (guarded by the existing `_enrichers_loaded` / `_types_loaded` flags) and **fault-tolerant**: if importing a pack raises, the error is logged to stderr and skipped, so one broken pack can never prevent the rest of Flowsint from loading. + +## Building a pack + +We'll build a small pack called `my_pack` that adds one type and one enricher. The layout: + +``` +my-pack/ +├── pyproject.toml +└── src/ + └── my_pack/ + ├── __init__.py + ├── types/ + │ ├── __init__.py + │ └── my_type.py + └── enrichers/ + ├── __init__.py + └── my_enricher.py +``` + +### Declaring the entry points + +The pack registers itself in `pyproject.toml`. Each entry point names a module that, when imported, pulls in everything you want registered: + +```toml +[project] +name = "my-pack" +version = "0.1.0" +requires-python = ">=3.12,<4.0" +dependencies = [ + "flowsint-core", + "flowsint-types", + # plus whatever your enrichers need, e.g. "httpx>=0.28,<0.29" +] + +[project.entry-points."flowsint.types"] +my_pack = "my_pack.types" + +[project.entry-points."flowsint.enrichers"] +my_pack = "my_pack.enrichers" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/my_pack"] +``` + +The entry-point **name** (`my_pack`) is just a label; the **value** (`my_pack.types`) is the import path Flowsint loads. `flowsint-core` and `flowsint-types` are not published on PyPI, so during local development resolve them from your Flowsint checkout, for example with a sibling path source: + +```toml +[tool.uv.sources] +flowsint-core = { path = "../flowsint/flowsint-core", editable = true } +flowsint-types = { path = "../flowsint/flowsint-types", editable = true } +``` + +### Writing a type + +Types are defined exactly as in [Managing types](/docs/developers/managing-types) — inherit from `FlowsintType`, decorate with `@flowsint_type`. The decorator registers into the same global `TYPE_REGISTRY`, regardless of which package the class lives in. + +```python +# src/my_pack/types/my_type.py +from typing import Self + +from flowsint_types.flowsint_base import FlowsintType +from flowsint_types.registry import flowsint_type +from pydantic import Field, model_validator + + +@flowsint_type +class MyType(FlowsintType): + """An example external type.""" + + value: str = Field( + ..., + description="Primary identifier", + json_schema_extra={"primary": True}, + ) + + @model_validator(mode="after") + def compute_label(self) -> Self: + self.nodeLabel = self.value + return self +``` + +### Writing an enricher + +Same contract as [Managing enrichers](/docs/developers/managing-enrichers): inherit from `Enricher`, decorate with `@flowsint_enricher`, implement `scan()` and `postprocess()`. You can import and pivot between your own types and the built-in ones (`Organization`, `Individual`, ...). + +```python +# src/my_pack/enrichers/my_enricher.py +from typing import Any, Dict, List + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_enrichers.registry import flowsint_enricher + +from my_pack.types.my_type import MyType + + +@flowsint_enricher +class MyEnricher(Enricher): + """[My source] Enrich MyType values.""" + + InputType = MyType + OutputType = Dict[str, Any] + + @classmethod + def name(cls) -> str: + return "mytype_to_infos" + + @classmethod + def category(cls) -> str: + return "My pack" + + @classmethod + def key(cls) -> str: + return "value" + + async def scan(self, data: List["MyType"]) -> List[Dict[str, Any]]: + results: List[Dict[str, Any]] = [] + for item in data: + try: + # ... call your source ... + results.append({"value": item.value, "info": "..."}) + except Exception as e: + results.append({"value": item.value, "error": str(e)}) + Logger.error(self.sketch_id, {"message": f"mytype_to_infos failed: {e}"}) + return results + + def postprocess(self, results, input_data=None): + for result in results: + if "error" in result: + continue + self.create_node(MyType(value=result["value"])) + return results + + +InputType = MyEnricher.InputType +OutputType = MyEnricher.OutputType +``` + +### Wiring the subpackages + +The entry points point at `my_pack.types` and `my_pack.enrichers`, so each subpackage's `__init__.py` must import the modules that define your decorated classes. This is what fires the decorators when Flowsint loads the entry point: + +```python +# src/my_pack/types/__init__.py +from .my_type import MyType # noqa: F401 + +__all__ = ["MyType"] +``` + +```python +# src/my_pack/enrichers/__init__.py +from . import my_enricher # noqa: F401 + +__all__ = ["my_enricher"] +``` + + +Keep the **top-level** `my_pack/__init__.py` free of side effects — do **not** import the `enrichers` subpackage from it. The enricher stack imports `flowsint_core`, so eagerly importing it would force the (heavier) core dependencies whenever someone merely loads your **types** via the `flowsint.types` entry point. Let each subpackage register itself independently. + +```python +# src/my_pack/__init__.py (intentionally side-effect free) +__version__ = "0.1.0" +``` + + +## Installing a pack + +Install the pack into the **same environment** as Flowsint, then restart the API and Celery worker so `load_all_*()` runs again. + +**Local development** (editable install): + +```bash +# from the Flowsint environment +uv pip install -e ../my-pack +``` + +**Docker / production** — add the pack to the API and worker images, for example over SSH for a private repo: + +```dockerfile +RUN --mount=type=ssh uv pip install "git+ssh://git@github.com//my-pack.git" +``` + +## Verifying discovery + +After installing and restarting, your types and enrichers appear through the API just like built-in ones. To confirm registration without the UI: + +```python +from flowsint_enrichers import ENRICHER_REGISTRY, load_all_enrichers +from flowsint_types.registry import TYPE_REGISTRY, load_all_types + +load_all_types() +load_all_enrichers() + +assert TYPE_REGISTRY.get("MyType") is not None +assert ENRICHER_REGISTRY.enricher_exists("mytype_to_infos") +``` + +## Troubleshooting + +If your pack doesn't show up, check that: + +- the distribution is installed in the **same** Python environment as the Flowsint API/worker, and they were **restarted** after installing; +- `pyproject.toml` declares the entry points under `flowsint.types` / `flowsint.enrichers`, and the **value** is the correct import path; +- each subpackage's `__init__.py` actually imports the modules that hold the decorated classes — an empty `__init__.py` registers nothing; +- importing the pack works on its own: run `python -c "import my_pack.enrichers"` in the Flowsint environment and read the traceback. Flowsint logs import failures to stderr (`Failed to load enricher plugin '' ...`) and skips them, so a silent absence usually means an import error you can reproduce directly. + +## Worked example: a Brazilian pack + +To make this concrete, here is the shape of a real pack — `flowsint-enrichers-br` — that adds Brazilian `CPF`/`CNPJ` types and a CNPJ → Organization lookup. It follows the in-tree conventions: a `flowsint_` package using the `src/` layout, `PascalCase` types with a single `primary` field, and `_to_` enrichers organised under an input-type directory. + +``` +flowsint-enrichers-br/ +├── pyproject.toml +└── src/ + └── flowsint_enrichers_br/ + ├── __init__.py # side-effect free + ├── types/ + │ ├── __init__.py # imports cpf, cnpj + │ ├── cpf.py # @flowsint_type class Cpf + │ └── cnpj.py # @flowsint_type class Cnpj + └── enrichers/ + ├── __init__.py # imports cnpj.to_organization + └── cnpj/ + └── to_organization.py # @flowsint_enricher class CnpjToOrganizationEnricher +``` + +The entry points target the two subpackages: + +```toml +[project.entry-points."flowsint.types"] +br = "flowsint_enrichers_br.types" + +[project.entry-points."flowsint.enrichers"] +br = "flowsint_enrichers_br.enrichers" +``` + +The `Cnpj` type stores the document normalised (digits only), validates it, and exposes a formatted graph label: + +```python +# src/flowsint_enrichers_br/types/cnpj.py +from typing import Self + +from flowsint_types.flowsint_base import FlowsintType +from flowsint_types.registry import flowsint_type +from pydantic import Field, field_validator, model_validator + + +@flowsint_type +class Cnpj(FlowsintType): + """Brazilian company taxpayer registry (CNPJ).""" + + cnpj: str = Field( + ..., + description="CNPJ (14 digits, stored without formatting)", + json_schema_extra={"primary": True}, + ) + + @field_validator("cnpj", mode="before") + @classmethod + def normalize(cls, v: str) -> str: + digits = "".join(c for c in str(v) if c.isdigit()) + if len(digits) != 14: # the real pack also validates the check digits + raise ValueError(f"Invalid CNPJ: {v!r}") + return digits + + @model_validator(mode="after") + def compute_label(self) -> Self: + c = self.cnpj + self.nodeLabel = f"{c[0:2]}.{c[2:5]}.{c[5:8]}/{c[8:12]}-{c[12:14]}" + return self +``` + +The enricher pivots a `Cnpj` to a built-in `Organization` (and its partners) using the public BrasilAPI — Receita Federal open data, no key required. Error handling is trimmed here for brevity; see [Managing enrichers](/docs/developers/managing-enrichers) for the full pattern: + +```python +# src/flowsint_enrichers_br/enrichers/cnpj/to_organization.py +from typing import Any, Dict, List + +import httpx +from flowsint_core.core.enricher_base import Enricher +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.organization import Organization + +from flowsint_enrichers_br.types.cnpj import Cnpj + + +@flowsint_enricher +class CnpjToOrganizationEnricher(Enricher): + """[BrasilAPI] Resolve a CNPJ to its registered company and partners.""" + + InputType = Cnpj + OutputType = Dict[str, Any] + + @classmethod + def name(cls) -> str: + return "cnpj_to_organization" + + @classmethod + def category(cls) -> str: + return "Brazil" + + @classmethod + def key(cls) -> str: + return "cnpj" + + async def scan(self, data: List["Cnpj"]) -> List[Dict[str, Any]]: + results: List[Dict[str, Any]] = [] + async with httpx.AsyncClient(timeout=20.0) as client: + for obj in data: + resp = await client.get( + f"https://brasilapi.com.br/api/cnpj/v1/{obj.cnpj}" + ) + resp.raise_for_status() + payload = resp.json() + payload["cnpj"] = obj.cnpj + results.append(payload) + return results + + def postprocess(self, results, input_data=None): + for r in results: + cnpj = Cnpj(cnpj=r["cnpj"]) + org = Organization(name=r.get("razao_social")) + self.create_node(cnpj) + self.create_node(org) + self.create_relationship(cnpj, org, "IS_REGISTERED_AS") + return results + + +InputType = CnpjToOrganizationEnricher.InputType +OutputType = CnpjToOrganizationEnricher.OutputType +``` + +Because CNPJ records are public, this enricher can ship a working data source. A `cpf_to_infos` enricher in the same pack would instead leave its source as a configurable `vaultSecret` parameter — CPF holder data is protected and must only be queried against a provider you are authorized to use. + +Finally, the subpackages wire their modules so the entry points register everything on import: + +```python +# src/flowsint_enrichers_br/types/__init__.py +from .cnpj import Cnpj # noqa: F401 +from .cpf import Cpf # noqa: F401 + +__all__ = ["Cpf", "Cnpj"] +``` + +```python +# src/flowsint_enrichers_br/enrichers/__init__.py +from .cnpj import to_organization # noqa: F401 +``` + +Install it alongside Flowsint and restart the API and worker. `Cpf` and `Cnpj` then appear as node types, and `cnpj_to_organization` shows up in the enricher list under the **Brazil** category. + +## Next steps + +External packs are the recommended way to maintain private or specialized intelligence sources alongside Flowsint without forking it. If a pack turns out to be broadly useful and uses only public data, consider contributing it upstream so the whole community benefits. diff --git a/docs/developers/managing-enrichers.mdx b/docs/developers/managing-enrichers.mdx index dba3dff9c..a087cbcb3 100644 --- a/docs/developers/managing-enrichers.mdx +++ b/docs/developers/managing-enrichers.mdx @@ -589,6 +589,8 @@ class MyEnricher(Enricher): ... ``` +If you'd rather keep an enricher **outside** the Flowsint repository — for a private source, or a domain/country-specific pack you version independently — you can ship it from your own Python package and have it auto-discovered through an entry point. See [External packs (plugins)](/docs/developers/external-packs). + ## Testing your enricher Creating tests helps ensure your enricher works correctly and makes debugging easier. Create a test file in `flowsint-enrichers/tests/`: diff --git a/docs/developers/managing-types.mdx b/docs/developers/managing-types.mdx index 82359590c..6580617a4 100644 --- a/docs/developers/managing-types.mdx +++ b/docs/developers/managing-types.mdx @@ -497,6 +497,8 @@ poetry install This updates the package in your development environment so enrichers and the API can import your new type. +You can also define types **outside** the Flowsint repository and have them auto-discovered through an entry point — useful for private or domain-specific packs versioned on their own. See [External packs (plugins)](/docs/developers/external-packs). + ## Integrating with the API The final step is making your type available through the API so frontends can discover it and create instances. diff --git a/flowsint-enrichers/src/flowsint_enrichers/registry.py b/flowsint-enrichers/src/flowsint_enrichers/registry.py index f21fc95a6..0f6032066 100644 --- a/flowsint-enrichers/src/flowsint_enrichers/registry.py +++ b/flowsint-enrichers/src/flowsint_enrichers/registry.py @@ -227,5 +227,40 @@ def load_all_enrichers() -> None: # Log but don't fail - some modules might have optional dependencies print(f"Warning: Failed to import {module_name}: {e}", file=sys.stderr) + # Discover third-party enricher packs declared via the "flowsint.enrichers" + # entry-point group. Any installed distribution can ship additional enrichers + # in its own top-level package without living inside flowsint_enrichers/. + # The imported module (or package __init__) is expected to import its + # submodules so the @flowsint_enricher decorators fire. + _load_enricher_plugins() + # Mark as loaded _enrichers_loaded = True + + +def _load_enricher_plugins() -> None: + """Import every package registered under the ``flowsint.enrichers`` entry-point group. + + External packages register themselves in their pyproject.toml:: + + [project.entry-points."flowsint.enrichers"] + my_pack = "my_pack.enrichers" + + Importing the referenced module triggers the @flowsint_enricher decorators, + registering the plugin's enrichers in the global ENRICHER_REGISTRY. + """ + try: + from importlib.metadata import entry_points + except ImportError: # pragma: no cover - importlib.metadata is stdlib on 3.12 + return + + for ep in entry_points(group="flowsint.enrichers"): + if ep.value in sys.modules: + continue + try: + importlib.import_module(ep.value) + except Exception as e: + print( + f"Warning: Failed to load enricher plugin '{ep.name}' ({ep.value}): {e}", + file=sys.stderr, + ) diff --git a/flowsint-types/src/flowsint_types/registry.py b/flowsint-types/src/flowsint_types/registry.py index cefe5a9eb..f9dfb71ec 100644 --- a/flowsint-types/src/flowsint_types/registry.py +++ b/flowsint-types/src/flowsint_types/registry.py @@ -197,5 +197,38 @@ def load_all_types() -> None: # Log but don't fail - some modules might have optional dependencies print(f"Warning: Failed to import {modname}: {e}", file=sys.stderr) + # Discover third-party type packs declared via the "flowsint.types" + # entry-point group, so external packages can ship their own FlowsintType + # subclasses without living inside flowsint_types/. + _load_type_plugins() + # Mark as loaded _types_loaded = True + + +def _load_type_plugins() -> None: + """Import every package registered under the ``flowsint.types`` entry-point group. + + External packages register themselves in their pyproject.toml:: + + [project.entry-points."flowsint.types"] + my_pack = "my_pack.types" + + Importing the referenced module triggers the @flowsint_type decorators, + registering the plugin's types in the global TYPE_REGISTRY. + """ + try: + from importlib.metadata import entry_points + except ImportError: # pragma: no cover - importlib.metadata is stdlib on 3.12 + return + + for ep in entry_points(group="flowsint.types"): + if ep.value in sys.modules: + continue + try: + importlib.import_module(ep.value) + except Exception as e: + print( + f"Warning: Failed to load type plugin '{ep.name}' ({ep.value}): {e}", + file=sys.stderr, + )