Blueprints

The blueprint system enables dynamic schema extension for domain models.

Overview

Blueprints allow you to define additional fields for models at runtime based on data stored in the graph database. This enables flexible, user-defined schemas without code changes.

How It Works

1. Blueprint definitions are stored in the graph with a JSON Schema payload
2. At runtime, get_model() queries matching Blueprint nodes from the graph
3. get_model() calls pydantic.create_model to return a subclass with the extra fields added
4. The extended model validates data according to the blueprint schema

Basic Usage

from imbi_common import blueprints, models, graph

async def example(db: graph.Graph) -> None:
    # Store a blueprint in the graph
    blueprint = models.Blueprint(
        name="Cloud Provider",
        slug="cloud-provider",
        type="Project",
        description="Cloud provider details",
        json_schema={
            "type": "object",
            "properties": {
                "cloud_provider": {
                    "type": "string",
                    "enum": ["AWS", "GCP", "Azure"]
                },
                "cloud_region": {
                    "type": "string"
                }
            },
            "required": ["cloud_provider"]
        }
    )
    await db.create(blueprint)

    # Get extended model for Project — blueprints whose type == "Project"
    # are fetched and applied
    ProjectModel = await blueprints.get_model(db, models.Project)

    # ProjectModel now has cloud_provider and cloud_region fields
    # and validates according to the blueprint schema

Filtering by Context

Blueprints can be filtered by project_type or environment:

# Only apply blueprints that match the given project type
ProjectModel = await blueprints.get_model(
    db,
    models.Project,
    context={"project_type": "microservice"},
)

API Reference

get_model async

get_model(
    database: Graph,
    model: type[ModelType],
    context: dict[str, str | list[str]] | None = None,
) -> type[ModelType]

Return a model class with matching blueprints applied.

Parameters:

Name	Type	Description	Default
model	type[ModelType]	The base Pydantic model class to extend.	required
context	dict[str, str | list[str]] | None	Optional filter context. When provided, only blueprints whose filter matches this context are applied. Blueprints with no filter are always included. Example:: {'project_type': 'apis'} {'project_type': 'apis', 'environment': 'production'}	None

Source code in src/imbi_common/blueprints.py

async def get_model[ModelType: pydantic.BaseModel](
    database: graph.Graph,
    model: type[ModelType],
    context: dict[str, str | list[str]] | None = None,
) -> type[ModelType]:
    """Return a model class with matching blueprints applied.

    Parameters:
        model: The base Pydantic model class to extend.
        context: Optional filter context. When provided, only
            blueprints whose ``filter`` matches this context
            are applied.  Blueprints with no filter are always
            included.  Example::

                {'project_type': 'apis'}
                {'project_type': 'apis',
                 'environment': 'production'}

    """
    all_blueprints = await database.match(
        models.Blueprint,
        {
            'type': model.__name__,
            'enabled': True,
        },
        order_by='priority',
    )
    matched = [
        bp
        for bp in all_blueprints
        if bp.kind == 'node' and _matches_filter(bp, context)
    ]
    return apply_blueprints(model, matched)