Skip to content

stream.contrib

stream.contrib

Entry-point discovery for Stream plugins.

Stream is extended by registering Python callables under these entry-point groups (in a user's pyproject.toml):

  • angelo.stream.feeds -> name = module:factory where factory(FeedConfig) -> FeedAdapter.
  • angelo.stream.routers -> name = module:factory (callable returning a Router, or a Router class).
  • angelo.stream.selectors -> name = module:fn where fn(Document) -> list[str] (schema names) for use as a classifier fallback.
  • angelo.stream.templates -> name = module:obj where obj is a WorkflowTemplate (instance or zero-arg factory).

Discovery is cached for the process lifetime. Entry points only appear after a (re)install writes them into distribution metadata, so during dev you can also register plugins in-process via the register_* helpers in the relevant module.

discover_templates 

discover_templates() -> list[Any]

Return WorkflowTemplate instances from the templates group.

Source code in stream/contrib.py 
82
83
84
85
86
87
88
89
90
def discover_templates() -> list[Any]:
    """Return WorkflowTemplate instances from the templates group."""
    out: list[Any] = []
    for obj in _discover_group(GROUP_TEMPLATES).values():
        try:
            out.append(obj() if callable(obj) and not hasattr(obj, "build") else obj)
        except Exception:  # pragma: no cover - best-effort
            continue
    return out