Extending kiln¶

kiln is designed to be extended at three levels:

Add an operation to an existing target. The most common extension – contribute a new CRUD-like endpoint to be, a cross-cutting concern (auth, rate limiting, caching), or a completely new file type.
Swap a renderer. Replace or augment how an existing output type is turned into code, without touching the operation that produces it.
Ship a new target. Build a generator for a different framework entirely by using codegen directly – no dependency on any of the existing target packages (be / be_root / fe / fe_root).

This document covers all three, in increasing order of ambition. The worked examples target be since it has the deepest config surface, but the same patterns apply to operations on any target (swap the @operation-decorator scope and the entry-point group for the target you’re extending). For background on the architecture, see Architecture.

Adding an operation ¶

An operation is a class decorated with operation() that produces typed output objects in its build() method. The engine takes care of scheduling: scope walking, dependency ordering, and options parsing.

Step 1 – write the class ¶

from pydantic import BaseModel

from codegen.engine import BuildContext
from codegen.operation import operation
from be.operations.types import RouteHandler, RouteParam


@operation("bulk_create", scope="resource", requires=["create"])
class BulkCreate:
    """POST /bulk -- insert many resources in one request."""

    class Options(BaseModel):
        max_items: int = 100

    def build(
        self,
        ctx: BuildContext,
        options: "Options",
    ) -> list[object]:
        model = ctx.instance.model.rpartition(".")[-1]
        handler = RouteHandler(
            method="post",
            path="/bulk",
            function_name=f"bulk_create_{model.lower()}",
            op_name="bulk_create",
            params=[
                RouteParam(
                    name="payload",
                    annotation=f"list[{model}CreateRequest]",
                ),
            ],
            body_lines=[
                f"if len(payload) > {options.max_items}:",
                "    raise HTTPException(413)",
                f"stmt = insert({model}).values(",
                "    [p.model_dump() for p in payload]",
                ")",
                "await db.execute(stmt)",
                "await db.commit()",
            ],
            status_code=201,
        )
        return [handler]

A few things to notice:

The class name is irrelevant to the engine – the name from the @operation(...) decorator is what matters.
scope="resource" means build() runs once per ResourceConfig.
requires=["create"] ensures the create operation builds first, so its schemas are available in the build store before this runs (useful if you want to inspect or extend them).
Options is optional. When omitted, operations receive EmptyOptions.

Step 2 – register via entry point ¶

Each target declares its own entry-point group for the operations that extend it. be’s group is be.operations; add your operation to your package’s pyproject.toml under that group:

[project.entry-points."be.operations"]
bulk_create = "my_pkg.ops:BulkCreate"

When you run codegen generate --target be, codegen walks the be.operations group and registers every entry alongside be’s own built-ins. Your operation is available as long as your package is pip installed alongside kiln-generator.

The same pattern works for be_root, fe, and fe_root: pick the matching <target>.operations group.

Step 3 – opt resources in ¶

Add the operation name to the resource’s operations list:

{
  resources: [{
    model: "myapp.models.Article",
    operations: [
      "get", "list", "create", "update", "delete",
      { name: "bulk_create", max_items: 50 },
    ],
  }],
}

Options come from the extra keys on the operation entry. Pydantic validates them against the Options class.

Cross-cutting operations with `when()`¶

Some operations should not appear in the user’s operations list at all – they activate themselves based on config and augment other operations’ output. Auth is the canonical example: when config.auth is set, auth silently appends a dependency to every CRUD handler.

Declare a when() method instead of relying on the opt-in list:

from codegen.engine import BuildContext
from codegen.operation import operation
from be.operations.types import RouteHandler


@operation(
    "rate_limit",
    scope="resource",
    requires=["get", "list", "create", "update", "delete"],
)
class RateLimit:
    """Add a rate-limit decorator to every write handler."""

    def when(self, ctx: BuildContext) -> bool:
        return getattr(ctx.config, "rate_limit", None) is not None

    def build(self, ctx, _options):
        limit = ctx.config.rate_limit
        for h in ctx.store.get_by_scope(
            ctx.scope.name, ctx.instance_id,
        ):
            if isinstance(h, RouteHandler) and h.method != "get":
                h.add_decorator(
                    f"@limiter.limit('{limit}')",
                )
                h.extra_imports.append(
                    ("myapp.rate_limit", "limiter"),
                )
        return []

Three important properties:

``when`` bypasses the opt-in list. Cross-cutting operations run whenever their predicate says so, regardless of the user’s operations config. Users don’t have to remember to opt in.
``requires`` orders it after producers. Listing the CRUD operations in requires guarantees their output exists in the build store by the time build runs.
The build method returns ``[]``. Augmenting operations mutate existing objects in place; they produce no new outputs of their own.

See src/be/operations/auth.py for the real-world auth implementation.

Augmenting vs producing ¶

codegen deliberately has one mechanism – operations – for both “produce new output” and “modify earlier output”. This keeps the execution model simple and uniform:

Role	Returns from `build`	Typical `requires`
Producer	New output objects	Nothing (or earlier producers whose output you depend on)
Augmenter	`[]` (nothing new)	All producers whose output you want to mutate

An operation can also do both: produce new objects and tweak earlier ones in the same build call.

Mutating output objects ¶

Every type in codegen.outputs and be.operations.types is a mutable dataclass with helpers for safe modification:

from be.operations.types import RouteHandler, SchemaClass

for handler in ctx.store.get_by_type(RouteHandler):
    handler.add_decorator("@cache(ttl=60)")
    handler.prepend_body("log.info('cached-endpoint-hit')")
    handler.extra_imports.append(("myapp.cache", "cache"))

for schema in ctx.store.get_by_type(SchemaClass):
    if schema.name.endswith("Resource"):
        schema.add_field("cached_at", "datetime", optional=True)

extra_imports is the recommended way to add imports. The assembler merges every handler’s extra_imports into the route file’s top-of-file import block automatically.

Swapping a renderer ¶

Every output type has a default renderer. Op-specific RouteHandler subclasses register their renderers at the bottom of each be.operations.<op> module; the shared cross-cutting renderers (SchemaClass, EnumClass, SerializerFn, StaticFile, TestCase) live in be.operations.renderers. You can override or supplement any of these without changing the operation that produces the output.

Register an additional renderer with a when predicate against the module-level codegen.render.registry. The registry tries registrations in order and uses the first whose predicate matches:

from be.operations.types import RouteHandler
from codegen.render import registry

@registry.renders(
    RouteHandler,
    when=lambda cfg: getattr(cfg, "use_async_retry", False),
)
def render_retry_handler(handler, ctx):
    # custom Jinja template that wraps the body in a retry loop
    return ctx.env.get_template(
        "my_pkg/retry_handler.j2",
    ).render(h=handler)

Import the module where this decorator runs from one of your operations (or from the package’s __init__) so it registers when codegen discovers operations via the target’s <target>.operations entry-point group. Register the predicate-guarded renderer before the default unguarded one if you want it to win when the flag is on.

Adding a new output type ¶

You are not limited to the built-in output types. A plugin can define its own dataclass, register a renderer for it, and have operations emit instances of it.

from dataclasses import dataclass, field
from codegen.operation import operation
from codegen.render import RenderRegistry


@dataclass
class GraphQLField:
    name: str
    gql_type: str
    resolver: str | None = None


@operation("graphql_fields", scope="resource")
class GraphQLFields:
    def build(self, ctx, _options):
        return [
            GraphQLField(name=f.name, gql_type=f.type.upper())
            for f in ctx.instance.fields
        ]


def register(registry: RenderRegistry) -> None:

    @registry.renders(GraphQLField)
    def render_gql(field, _ctx):
        return f"{field.name}: {field.gql_type}"

The assembler only knows how to group the built-in types. A plugin that introduces a new type is also responsible for extending the assembler (or shipping its own) so the renderer output ends up in the right file.

Shipping a new target ¶

If you are generating code for something that has nothing to do with FastAPI (a Go CLI, a Terraform module, a gRPC service), you can ship your own plugin that registers as a codegen target. Install it alongside codegen and the codegen CLI will discover and dispatch to it the same way it does for be.

Step 1 – write the generator ¶

from codegen.engine import Engine
from codegen.env import create_jinja_env
from codegen.render import RenderCtx, RenderRegistry

def generate_my_thing(config):
    engine = Engine(operations=[...])
    store = engine.build(config)

    registry = RenderRegistry()
    register_my_renderers(registry)

    env = create_jinja_env("my_pkg", "templates")
    ctx = RenderCtx(env=env, config=config)

    files = []
    for obj in store.all_items():
        content = registry.render(obj, ctx)
        files.append(GeneratedFile(path=..., content=content))
    return files

Step 2 – register a `Target`¶

# my_pkg/target.py
from pathlib import Path

from codegen.target import Target
from my_pkg.config import load_my_config
from my_pkg.generate import generate_my_thing

target = Target(
    name="mytarget",
    load_config=load_my_config,
    generate=generate_my_thing,
    default_out=lambda cfg: Path(cfg.out_dir or "."),
)

Step 3 – expose it via the entry-point group ¶

[project.entry-points."codegen.targets"]
mytarget = "my_pkg.target:target"

Install your package and codegen generate --target mytarget --config ... works. Raise subclasses of codegen.errors.CLIError for user-facing mistakes; anything else will propagate with a traceback.

You still need, as with the existing targets:

A Pydantic config schema for your target.
Your own operations under <your-target>.operations (the group named in your Target’s operations_entry_point).
Your own renderers, registered against codegen.render.registry.
A Jinja2 template directory (or a different renderer backend).

Everything in codegen is target-agnostic and reusable.

Operation validation ¶

Per-operation options are validated by Pydantic at config-load time because they are parsed into the operation’s Options model. Any cross-field validation belongs in a Pydantic @model_validator(mode="after") on that model:

from pydantic import BaseModel, model_validator


class BulkCreateOptions(BaseModel):
    max_items: int = 100
    batch_size: int = 10

    @model_validator(mode="after")
    def _check(self):
        if self.batch_size > self.max_items:
            raise ValueError(
                "batch_size must be <= max_items",
            )
        return self


@operation("bulk_create", scope="resource")
class BulkCreate:
    Options = BulkCreateOptions
    ...

Errors raised during config loading are reported to the user before generation begins.

Testing your extension ¶

Run your operations through the engine directly – no CLI needed:

from be.config.schema import ProjectConfig
from codegen.engine import Engine
from be.operations.types import RouteHandler

from my_pkg.ops import BulkCreate


def test_bulk_create_produces_handler():
    cfg = ProjectConfig.model_validate({
        "resources": [
            {"model": "myapp.Article", "operations": ["bulk_create"]},
        ],
    })
    engine = Engine(operations=[BulkCreate])
    store = engine.build(cfg)

    handlers = store.get_by_type(RouteHandler)
    assert any(h.path == "/bulk" for h in handlers)

For full end-to-end coverage, load a fixture config via codegen.config.load_config() and pass it through codegen.pipeline.generate() to get back the full list of GeneratedFile objects.

kiln

Navigation

Related Topics