Handlers¶

Handlers are the Python functions that implement the logic of each vertex. This section covers handler resolution, security, and the handler registry.

Writing Handlers¶

A handler is any Python function that:

Accepts keyword arguments matching the vertex's resolved inputs.
Returns a dictionary with keys matching the vertex's declared outputs.

def normalize_name(name: str) -> dict:
    return {"name": name.strip().title()}

That's it. No base classes, no decorators required, no framework coupling.

Async Handlers¶

For I/O-bound operations, write async handlers:

async def fetch_user(user_id: int) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"/api/users/{user_id}")
    return {"user": response.json()}

Use async_enabled=True and run_flow_async() to execute async handlers.

Handler Resolution¶

HBIA resolves handlers in three ways:

1. Dotted Path Strings (YAML)¶

In YAML, handler: vertices.users.normalize is resolved via importlib:

normalize:
  handler: vertices.users.normalize

At runtime: from vertices.users import normalize.

2. Dictionary Mapping¶

When using the Python API, pass a dict of {vertex_name: callable}:

result = run_flow(graph, handlers={
    "normalize": normalize_fn,
    "validate": validate_fn,
})

3. Callable Resolver¶

A function that takes a handler name and returns a callable:

result = run_flow(graph, handlers=lambda name: registry.get(name))

HandlerResolver¶

The HandlerResolver class handles dotted path resolution with optional security restrictions:

from honey_badgeria.back.registry import HandlerResolver

resolver = HandlerResolver(
    allowed_prefixes=("myproject.vertices", "myproject.handlers"),
)

# Resolves dotted path to callable
fn = resolver.resolve("myproject.vertices.users.fetch")

# Direct callable passed through unchanged
fn = resolver.resolve(my_function)

Caching¶

Import results are cached to avoid repeated module imports. The first resolution of vertices.users.fetch imports the module; subsequent resolutions use the cached reference.

Security: Allowed Prefixes¶

The allowed_prefixes parameter restricts which modules can be imported:

resolver = HandlerResolver(
    allowed_prefixes=("myproject.vertices",)
)

resolver.resolve("myproject.vertices.users.fetch")  # OK
resolver.resolve("os.system")                        # HandlerResolutionError!
resolver.resolve("subprocess.run")                   # HandlerResolutionError!

This prevents malicious or accidental execution of arbitrary Python code. Configure via settings:

from honey_badgeria.conf import Settings, configure

configure(Settings(
    ALLOWED_HANDLER_PREFIXES=("myproject.vertices", "myproject.handlers"),
))

The @vertex Decorator¶

Optionally, you can use the @vertex decorator to register handlers:

from honey_badgeria import vertex

@vertex
def fetch_user(user_id: int) -> dict:
    return {"id": user_id, "name": "Alice"}

@vertex(name="custom_name")
def process():
    return {"done": True}

The decorator:

Sets __hbia_vertex__ = True on the function.
Sets __hbia_vertex_name__ to the function name (or custom name).
Registers the function in VERTEX_REGISTRY.

The @flow Decorator¶

Mark functions as flow definitions:

from honey_badgeria import flow

@flow("my_flow")
def define_my_flow():
    pass

The decorator:

Sets __hbia_flow__ = True on the function.
Sets __hbia_flow_name__ to the provided name.
Registers the function in FLOW_REGISTRY.

Exceptions¶

Exception	When
`HandlerNotFoundError`	No handler found for a vertex name in the provided handlers dict
`HandlerResolutionError`	Dotted path import fails or is blocked by allowed_prefixes