Python API Reference¶
This page documents the key public APIs for using Honey Badgeria programmatically.
Execution¶
run_flow()¶
Execute a flow synchronously.
from honey_badgeria.back.runtime import run_flow
result = run_flow(
graph, # Graph object
flow_name="create_user", # Optional: specific flow
handlers={ # Handler mapping
"normalize": normalize_fn,
"validate": validate_fn,
"save": save_fn,
},
initial_data={"name": "alice"}, # Starting data
cache_enabled=False, # Optional overrides
parallel_enabled=False,
async_enabled=False,
max_workers=4,
transaction_backend=None,
)
# Returns dict[str, Any] with all vertex outputs
result["save.user_id"] # Access specific output
run_flow_async()¶
Execute a flow asynchronously.
from honey_badgeria.back.runtime import run_flow_async
result = await run_flow_async(
graph,
handlers={...},
initial_data={...},
async_enabled=True,
)
Graph Construction¶
load_graph_from_yaml()¶
Load a graph from a YAML file.
from honey_badgeria.back.dsl import load_graph_from_yaml
graph = load_graph_from_yaml("flows/create_user.yaml")
GraphBuilder.from_dict()¶
Build a graph from a Python dictionary.
from honey_badgeria.back.loader import GraphBuilder
graph = GraphBuilder.from_dict({
"flow": {
"my_flow": {
"vertex_a": {
"handler": "mod.fn_a",
"effect": "pure",
"outputs": {"x": "int"},
"next": ["vertex_b"],
},
"vertex_b": {
"handler": "mod.fn_b",
"inputs": {"x": "vertex_a.x"},
},
}
}
})
Graph Model¶
Vertex¶
from honey_badgeria.back.graph import Vertex
v = Vertex(
name="fetch_user",
handler="vertices.users.fetch",
inputs={"user_id": "int"},
outputs={"user": "dict"},
version="1",
effect="pure",
)
Edge¶
Graph¶
from honey_badgeria.back.graph import Graph
graph = Graph()
graph.add_vertex(v)
graph.add_edge(e)
sources = graph.get_sources()
sinks = graph.get_sinks()
Topology¶
GraphTopology¶
from honey_badgeria.back.topology import GraphTopology
topo = GraphTopology(graph)
sorted_vertices = topo.topological_sort()
stages = topo.execution_stages()
sources = topo.sources()
sinks = topo.sinks()
deps = topo.dependencies("vertex_name")
dependents = topo.dependents("vertex_name")
Atomicity¶
AtomicGroup¶
from honey_badgeria.back.atomicity import AtomicGroup, FailurePolicy
group = AtomicGroup(
name="payment",
vertices=["reserve", "charge", "confirm"],
failure_policy=FailurePolicy.ROLLBACK,
no_cache=True,
no_parallel=True,
)
InMemoryBackend¶
from honey_badgeria.back.atomicity import InMemoryBackend
backend = InMemoryBackend()
# After execution:
backend.was_committed() # bool
backend.was_rolled_back() # bool
backend.get_snapshot() # dict
backend.reset()
SagaCoordinator¶
from honey_badgeria.back.atomicity.saga import SagaCoordinator, SagaStep
saga = SagaCoordinator("name", backend, logger)
saga.add_step(SagaStep(
name="step_name",
vertex_name="vertex",
compensate_handler=lambda inputs, result: undo(result),
))
results = saga.execute(step_executor_fn)
Contracts¶
VertexContract¶
from honey_badgeria.back.contracts import VertexContract
contract = VertexContract(
inputs={"user_id": "int"},
outputs={"user": "dict"},
version="1.0",
effect="pure",
)
contract.validate_input({"user_id": 123})
contract.validate_output({"user": {...}})
Validation & Linting¶
from honey_badgeria.back.dsl.validator import validate_dsl
from honey_badgeria.back.graph.validator import validate_graph
from honey_badgeria.back.lint import AILinter
# DSL validation
errors = validate_dsl(yaml_dict)
# Graph validation
errors = validate_graph(graph)
# Linting
linter = AILinter()
issues = linter.lint(yaml_dict)
Handler Resolution¶
from honey_badgeria.back.registry import HandlerResolver
resolver = HandlerResolver(allowed_prefixes=("myproject.vertices",))
fn = resolver.resolve("myproject.vertices.users.fetch")
Discovery¶
from honey_badgeria.back.discovery import GraphAutoDiscover
discover = GraphAutoDiscover(handler_modules=["vertices"])
graph_dict = discover.run()
Configuration¶
from honey_badgeria.conf import Settings, get_settings, configure
settings = Settings(CACHE_ENABLED=True, PARALLEL_ENABLED=True)
configure(settings)
current = get_settings()
Frontend¶
Loading Frontend Graphs¶
from honey_badgeria.front.dsl import (
load_ui_graph,
load_state_graph,
load_effect_graph,
load_event_graph,
)
from honey_badgeria.front.graph import FrontendDomain
ui = load_ui_graph("graph/dashboard/ui/")
state = load_state_graph("graph/dashboard/state/")
effects = load_effect_graph("graph/dashboard/effects/")
events = load_event_graph("graph/dashboard/events/")
domain = FrontendDomain("dashboard", ui, state, effects, events)
Code Generation¶
from honey_badgeria.front.codegen import FrontendCodegen
codegen = FrontendCodegen(domains=[domain])
codegen.generate(output_dir="hbia-runtime/")
Frontend Linting & Explanation¶
from honey_badgeria.front.lint import FrontendLinter
from honey_badgeria.front.explain import FrontendExplainer
linter = FrontendLinter()
issues = linter.lint(domain)
explainer = FrontendExplainer(domain)
explainer.explain()
Decorators¶
from honey_badgeria import vertex, flow
@vertex
def my_handler(x: int) -> dict:
return {"y": x * 2}
@vertex(name="custom_name")
def another() -> dict:
return {"done": True}
@flow("my_flow")
def define_flow():
pass
Utilities¶
Graph Tools¶
from honey_badgeria.utils.graph_tools import merge_graphs, subgraph
combined = merge_graphs(graph1, graph2)
sub = subgraph(graph, ["vertex_a", "vertex_b"])
Import Tools¶
from honey_badgeria.utils.import_tools import import_string, try_import
cls = import_string("mymodule.MyClass")
yaml = try_import("yaml") # Returns None if not installed