Bedrock

Bedrock Application Development

Bedrock is a modular Python framework for building applications with manifest-driven module loading, lifecycle orchestration, and clean architecture conventions. This skill teaches you how to use Bedrock to build maintainable, extensible applications.

Quick Reference

When to Read References

• Creating or modifying a module → Read references/module-guide.md
• Using dependency injection → Read references/di-guide.md
• Using the hook system → Read references/hooks-guide.md
• Using database features → Read references/database-guide.md
• Using CLI commands → Read references/cli-guide.md
• Using signals / events → Read references/signals-guide.md
• Understanding project structure → Read references/module-hierarchy.md
• Understanding architecture / layer boundaries → Read references/architecture.md

Quick Start: New Project

Create a Bedrock application with this structure:

myproject/
├── __init__.py
├── app.py             
├── settings.py         # Application settings
├── manifest.yaml
├── entities.py
├── service.py
├── exc.py
└── bootstrap.py

import bedrock

bedrock.setup("myproject.modules.users")

Settings (settings.py):

from pydantic_settings import BaseSettings, SettingsConfigDict
from bedrock.conf import SettingsProxy


class AppSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="MYAPP_")
    DEBUG: bool = False
    DATABASE_URL: str = "sqlite:///app.db"


app_settings: AppSettings = SettingsProxy(AppSettings)  # type: ignore[assignment]

First module (my_app/users/manifest.yaml):

title: User Management
version: "0.1.0"
description: User authentication and profiles

Quick Start: New Module

To add a module to an existing Bedrock project:

1. Create manifest.yaml with title and version (required) 2. Create __init__.py package marker (required) 3. Create entities.py for domain data models 4. Create exc.py for module exceptions 5. Create service.py for business logic 6. Create bootstrap.py for lifecycle hooks (optional) 7. Create models.py for database tables (optional) 8. Create installation.py for install hooks (optional)

Core Workflow

1. Bootstrap the Application

Every Bedrock process starts with bedrock.setup():

import bedrock
bedrock.setup()  # Reads BEDROCK_APP env var, or pass app path explicitly
bedrock.setup("myproject.app")  # Explicit app path

Call once at process start. Safe to call multiple times.

2. Module Structure (Standard)

<name>/
├── __init__.py          # Package marker
├── manifest.yaml        # Required: title, version, depends_on
├── entities.py          # Pydantic models (extend BedrockEntity)
├── service.py           # Business logic (sync + async pairs)
├── exc.py               # Module exceptions (extend BedrockExc)
├── bootstrap.py         # Lifecycle hooks: on_load, ready, on_shutdown
├── models.py            # SQLAlchemy models (extend BedrockModel) — optional
├── installation.py      # install(), pre_install(), post_install() hooks — optional
└── commands.py          # Typer CLI app — optional

Not every module needs every file. Minimum viable module: __init__.py + manifest.yaml.

3. Manifest Schema

title: My Module           # Required: display name
version: "0.1.0"           # Required: module version
description: Short desc    # Optional
depends_on:                # Optional: import paths of dependencies
  - myproject.users
commands: commands:app      # Optional: relative Typer app import path

The name at runtime is the Python import path (e.g. myproject.modules.users), NOT a field in the manifest. Dependencies use fully qualified import paths.

4. Entity Pattern

Extend BedrockEntity for all domain data models:

from bedrock.entities import BedrockEntity


class UserCreateRequest(BedrockEntity):
    name: str
    email: str
    password: str


class UserResponse(BedrockEntity):
    id: int
    name: str
    email: str
    is_active: bool = True

BedrockEntity is a Pydantic BaseModel with arbitrary_types_allowed=True. Use Pydantic v2 syntax.

6. Exception Pattern

Create module-level exceptions by subclassing BedrockExc:

from bedrock.exc import BedrockExc


class UserError(BedrockExc):
    """Base exception for user module."""
    detail: str = "User operation failed."

Every exception MUST set a detail class attribute with a default message. Constructor accepts optional msg to override. Create a module-level base exception (e.g. UserError) for catching all module errors.

7. Bootstrap Hooks

Lifecycle hooks execute at specific points during module loading. Bedrock always calls hooks with keyword arguments — use keyword-only signatures:

# bootstrap.py
from bedrock.module import ModuleRegistry, AppConfig


def on_load(*, registry: ModuleRegistry, app: AppConfig) -> None:
    """Called during install(), after module is added."""
    pass


def ready(*, registry: ModuleRegistry, app: AppConfig) -> None:
    """Called after ALL modules are installed."""
    from .service import user_service
    user_service.initialize()


def on_shutdown(*, registry: ModuleRegistry, app: AppConfig) -> None:
    """Called during shutdown, in REVERSE install order."""
    pass

Hooks use keyword-only signatures. The registry inspects each hook's parameters and injects only what it declares: registry (ModuleRegistry), app (AppConfig), container (DI container), hooks (HookRegistry).

8. Database Models

Extend BedrockModel for database tables:

from bedrock.database.base import BedrockModel
from sqlalchemy import String, Boolean
from sqlalchemy.orm import Mapped, mapped_column


class UserModel(BedrockModel):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

BedrockModel provides these methods on all instances:

• .dict() — Serialize to dict
• .update(**kwargs) — Update fields
• .delete() — Delete instance

9. Database Queries

Use search_filter_sort_paginate for filtered, paginated queries:

from bedrock.database import db
from bedrock.database.service import search_filter_sort_paginate

result = search_filter_sort_paginate(
    db_session=db.session,
    model=UserModel,
    filter_specs=[{"field": "name", "op": "==", "value": "Alice"}],
    sort_key="name",
    sort_dir="asc",
    page=1,
    limit=10,
)
# Returns: {"items": [...], "total": N, "page_info": {"page": 1, "limit": 10, "has_more": True}}

Filter operators: ==, !=, >, <, >=, <=, like, ilike, in, not_in, between, has, any, text_search, fuzzy_search

Boolean combinators: or, and

{"field": "name", "op": "==", "value": "Alice"}
{"or": [{"field": "age", "op": ">", "value": 30}, {"field": "name", "op": "==", "value": "Bob"}]}

Nested filters: Use . (join) or : (any/has) notation for related models.

10. Settings

Create module-level settings using BaseSettings. Wrap with SettingsProxy when you need deferred initialization (module-level singletons where env vars may not be ready at import time):

from pydantic_settings import BaseSettings, SettingsConfigDict
from bedrock.conf import SettingsProxy


class MyModuleSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="MYMODULE_")
    API_KEY: str = ""
    DEBUG: bool = False
    MAX_RETRIES: int = 3


my_settings: MyModuleSettings = SettingsProxy(MyModuleSettings)  # type: ignore[assignment]

When environment variables are guaranteed to be ready at construction time (e.g. inside a ready() hook), use BaseSettings directly:

from pydantic_settings import BaseSettings, SettingsConfigDict


class DbSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="DATABASE_")
    URL: str = "sqlite:///app.db"
    ECHO: bool = False

Rules:

• Use SettingsProxy for module-level singletons that may be imported before env vars are ready
• Use BaseSettings directly when env vars are guaranteed ready at construction time
• Always use env_prefix in model_config to namespace environment variables

11. Signal System

Bedrock provides lifecycle signals for cross-module notification (fire-and-forget):

from bedrock.signal import Signal

# Define a custom signal
user_created = Signal("user_created")

# Connect receivers (decorator form)
@user_created.connect
def on_user_created(sender, user):
    print(f"User created: {user.name}")

# Connect receivers (explicit form)
def on_user_created(sender, user):
    print(f"User created: {user.name}")

user_created.connect(on_user_created)

# Send signal (sync) — in pure sync context, adapts async receivers
user_created.send(sender, user=new_user)

# Send signal (async) — canonical for mixed sync/async receivers
await user_created.asend(sender, user=new_user)

Sync/async contract: send() is sync-facing. In a pure sync context it can adapt async receivers. Inside a running event loop, it raises RuntimeError if an async receiver is reached. Use await asend() for async or mixed contexts. See references/signals-guide.md for details.

Built-in lifecycle signals:

from bedrock.module.signals import (
    module_loaded,     # After each module's on_load hook
    module_ready,      # After each module's ready hook
    module_shutdown,   # After each module's on_shutdown hook
    registry_ready,    # After all modules ready
    registry_shutdown, # When registry shuts down
)

Common patterns:

12. Hook System (Call/Response)

For structured, multi-implementation extension points that return values, use the hook system instead of signals:

from bedrock.hooks import HookNamespace

auth = HookNamespace("auth")

@auth.spec(firstresult=True)
def authenticate(request):
    """Hook spec: first non-None result wins."""

@auth.impl(priority=10)
def default_auth(request):
    return verify_token(request.token)

# Dispatch
request = {"token": "..."}
results = auth.call("authenticate", request=request)

Signals vs Hooks: Signals are notification-only (no return value). Hooks are call/response (implementations return values, ordered by priority, with optional firstresult short-circuit).

Anti-Patterns (Avoid These)

1. Don't put HTTP types in service/entity layers — keep Request/Response at adapter boundaries or in separate schemas.py 2. Don't subclass `BedrockExc` without `detail` — error messages depend on it 3. Don't bypass `ModuleRegistry.install()` to load modules manually — dependency resolution and hooks will be skipped 4. Don't call `db.init()` or `apps.populate()` more than once per process

Commands Reference

See references/cli-guide.md for the complete CLI command reference.

Key commands: <module> is app import path (e.g. my_erp.oms)

• bedrock run --app <module> — Run app commands
• bedrock manage install — Install modules with migrations + hooks
• bedrock db revision <module> -m "msg" — Create migration
• bedrock db upgrade <module> — Apply migrations
• bedrock app info <module> — Inspect module
• bedrock app playbook <module> — Inspect module

Architecture Reference

Bedrock's layered architecture, data flow, and singleton lifecycle.

---

Core Principles

1. Framework-agnostic core — No HTTP dependencies in the module system. FastAPI support belongs in an adapter layer. 2. Explicit modularity — Modules declare identity and dependencies in manifest.yaml. No import-time side effects for registration. 3. Clean layer boundaries — Models → Entities → Services → API. Each layer has a single responsibility. 4. Predictable conventions — Strict file structure helps developers and AI assistants produce accurate code.

---

Layer Boundaries

Constraints

• API → Logic: API calls services, never the reverse
• Logic → Data: Services access persistence through models only
• Logic → Validation: Services use Pydantic entities for contracts
• No HTTP in Logic: Service functions never accept Request or Response
• No ORM in API: Controllers never interact with SQLAlchemy sessions directly

---

Data Flow

Inbound → Validation → Dispatch → Business Logic → Persistence → Response

1. Inbound — Request arrives at API layer (HTTP, CLI, background task) 2. Validation — Adapter deserializes payload into Pydantic entity, rejects malformed input 3. Dispatch — API calls service function with validated entities 4. Business Logic — Service orchestrates: queries data layer, applies rules, emits signals 5. Persistence — Data layer executes ORM operations in managed session 6. Response — Service returns entities, API serializes for transport

Same business logic works from FastAPI routes, Celery tasks, or CLI commands.

---

Singleton Lifecycle

Key Singletons

Initialization Sequence

import time → apps exists (empty), db exists (unconfigured), container exists, hooks exists
    ↓
bedrock.setup() → apps.populate(module_list)
    ↓
on_load hooks → modules call db.init(url) if needed
    ↓
apps.mark_ready() → ready hooks fire
    ↓
hooks.validate() → warn about orphaned impls or empty specs
    ↓
registry_ready signal emitted

Shutdown

Reverse install order: each module's on_shutdown hook runs, then registry_shutdown signal fires. Close connections and release resources here.

---

Runtime Primitives

DI Container (bedrock.di)

A lightweight dependency injection container with three lifetimes: SINGLETON, TRANSIENT, and SCOPED. Thread-safe singleton resolution via double-checked locking.

• `container.register(key, factory=..., lifetime=...)` — Register a service factory.
• `container.resolve(key)` — Resolve by type or string key.
• `container.scope(name)` — Context manager for scoped services.
• `container.override(key, instance)` — Context manager for test doubles.
• `@provider` / `@inject(mappings)`** — Decorator shortcuts.

The container is injected into bootstrap hooks that declare a container parameter.

Hook System (bedrock.hooks)

Structured call/response protocol for multi-implementation extension points. Unlike signals (notification-only), hooks return values and support priority ordering and firstresult short-circuit.

• `@hookspec` / `@hookimpl(priority=0)` — Marker decorators for global registration.
• `HookNamespace(name)` — Module-facing scoped API: ns.spec(), ns.impl(), ns.call(), ns.acall().
• `hooks.call(fqn, kwargs)** / **hooks.acall(fqn, kwargs)` — Dispatch to all implementations.

The hook registry is injected into bootstrap hooks that declare a hooks parameter.

---

When to Use Each Layer

Decision Rules

• Touches Request/Response → API layer
• Imports sqlalchemy → Data layer
• Imports pydantic for shapes → Validation layer
• Contains logic/orchestration → Logic layer
• Runs once at startup → Lifecycle hooks (bootstrap.py)

Bedrock CLI Reference

CLI commands for building and managing Bedrock applications.

Invocation

uv run bedrock <command> [options]

---

bedrock run

Run commands declared by your Bedrock modules.

Usage

bedrock run --app <module_import_path> <subcommand> [args...]
bedrock run -a <module_import_path> <subcommand> [args...]

Example

Given a module myapp.users with commands: commands:app in its manifest:

bedrock run --app myapp.users create-user --name Alice --email alice@example.com
bedrock run --app myapp.users list-users

---

bedrock manage install

Install a module: run database migrations, then execute installation hooks.

Usage

bedrock manage install [options]

Options

What Happens

For each module (and its dependencies): 1. Run Alembic migrations (create/upgrade schema) 2. Execute installation.py hooks: pre_install() → install() → post_install()

Examples

bedrock manage install
bedrock manage install -A myapp.users
bedrock manage install -A myapp.users --skip-migrations

---

bedrock app info

Display module metadata.

bedrock app info <import_path>

Example: bedrock app info myapp.users

Shows: title, description, version, dependencies, bootstrap/models status, declared commands.

---

bedrock app inspect

Validate a module's structure.

bedrock app inspect <import_path>

Checks: manifest.yaml validity, bootstrap.py importability, models.py importability, installation hooks.

---

bedrock app playbook

Read documentation from a module's playbook.

bedrock app playbook <module> [path]

Without a path, prints playbook/PLAYBOOK.md. With a path, prints the specified file from the playbook/ directory. Path traversal is blocked for security.

Examples:

bedrock app playbook myapp.auth
bedrock app playbook myapp.auth references/api-reference.md
bedrock app playbook myapp.auth examples/login-flow.py

Use this when your module depends_on an external module and you need to understand its API.

---

bedrock db revision

Create a new database migration.

bedrock db revision <app> -m "description" [--autogenerate | --no-autogenerate]

Examples:

bedrock db revision myapp.users -m "add users table"
bedrock db revision myapp.users -m "seed data" --no-autogenerate

---

bedrock db upgrade

Apply migrations.

bedrock db upgrade <app> [target]

Targets: head (default), +N (upgrade N steps), <revision_id>

Examples:

bedrock db upgrade myapp.users
bedrock db upgrade myapp.users +1

---

bedrock db downgrade

Roll back migrations.

bedrock db downgrade <app> <target>

Targets: -N (downgrade N steps), base (empty), <revision_id>

Examples:

bedrock db downgrade myapp.users -1
bedrock db downgrade myapp.users base

---

bedrock db heads

Show latest revision.

bedrock db heads myapp.users

---

bedrock db current

Show current database revision.

bedrock db current myapp.users

---

bedrock db history

Show migration history.

bedrock db history myapp.users

---

bedrock db uninstall

Downgrade to base and clean up. Use when removing a module.

bedrock db uninstall myapp.users

---

Module CLI Commands

How to create module-level CLI commands that integrate with bedrock run.

Steps

1. Create `commands.py` in your module package with a Typer app:

# myapp/users/commands.py
import typer

app = typer.Typer()


@app.command()
def create_user(name: str, email: str):
    """Create a new user."""
    print(f"Creating user: {name} <{email}>")


@app.command()
def list_users():
    """List all users."""
    print("Listing users...")

2. Register in `manifest.yaml`:

title: users
description: User management
version: 0.1.0
commands: "commands:app"

3. Invoke with `bedrock run`:

bedrock run --app myapp.users create-user --name Alice --email alice@example.com
bedrock run --app myapp.users list-users

Bedrock CLI Reference (For Application Developers)

CLI commands you'll use when building applications with Bedrock.

Invocation

uv run bedrock <command> [options]

---

bedrock run

Run commands declared by your Bedrock modules.

Usage

bedrock run --app <module_import_path> <subcommand> [args...]

Example

Given a module myapp.users with commands: commands:app in its manifest:

bedrock run --app myapp.users create-user --name Alice --email alice@example.com
bedrock run --app myapp.users list-users

---

bedrock manage install

Install a module: run database migrations, then execute installation hooks.

Usage

bedrock manage install [options]

Options

What Happens

For each module (and its dependencies): 1. Run Alembic migrations (create/upgrade schema) 2. Execute installation.py hooks: pre_install() → install() → post_install()

Examples

bedrock manage install
bedrock manage install -A myapp.users
bedrock manage install -A myapp.users --skip-migrations

---

bedrock app info

Display module metadata.

bedrock app info <import_path>

Example: bedrock app info myapp.users

Shows: title, description, version, dependencies, bootstrap/models status, declared commands.

---

bedrock app inspect

Validate a module's structure.

bedrock app inspect <import_path>

Checks: manifest.yaml validity, bootstrap.py importability, models.py importability, installation hooks.

---

bedrock app playbook

Read documentation from a module's playbook/ directory.

bedrock app playbook <module> [path]

Without a path, prints playbook/PLAYBOOK.md to stdout. With a path, prints the specified file from the playbook/ directory. Path traversal is blocked for security.

Examples:

bedrock app playbook myapp.users
bedrock app playbook myapp.users references/architecture.md
bedrock app playbook myapp.users references/templates/example.py

Useful for AI agents learning how to use a module.

---

bedrock db revision

Create a new database migration.

bedrock db revision <app> -m "description" [--autogenerate | --no-autogenerate]

Examples:

bedrock db revision myapp.users -m "add users table"
bedrock db revision myapp.users -m "seed data" --no-autogenerate

---

bedrock db upgrade

Apply migrations.

bedrock db upgrade <app> [target]

Targets: head (default), +N (upgrade N steps), <revision_id>

Examples:

bedrock db upgrade myapp.users
bedrock db upgrade myapp.users +1

---

bedrock db downgrade

Roll back migrations.

bedrock db downgrade <app> <target>

Targets: -N (downgrade N steps), base (empty), <revision_id>

Examples:

bedrock db downgrade myapp.users -1
bedrock db downgrade myapp.users base

---

bedrock db heads

Show latest revision.

bedrock db heads myapp.users

---

bedrock db current

Show current database revision.

bedrock db current myapp.users

---

bedrock db history

Show migration history.

bedrock db history myapp.users

---

bedrock db uninstall

Downgrade to base and clean up. Use when removing a module.

bedrock db uninstall myapp.users

Database Operations Guide

Complete guide to database operations in Bedrock applications.

Table of Contents

1. BedrockModel Usage 2. Database Initialization 3. Session Management 4. Migrations 5. Query Patterns 6. Filter Syntax 7. Relationships 8. Common Patterns 9. Configuration 10. Model Observer 11. DatabaseManager API

---

BedrockModel Usage

All database models extend BedrockModel using SQLAlchemy 2.0 Mapped syntax.

from bedrock.database.base import BedrockModel
from sqlalchemy import String, Boolean, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column


class UserModel(BedrockModel):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

CrudMixin Methods

BedrockModel includes CrudMixin, which provides:

---

Database Initialization

Initialize the database once per process before any database access:

from bedrock.database import db

# With explicit URL
db.init(url="postgresql://user:pass@localhost/mydb")

# With settings object
from bedrock.database.config import DbSettings
settings = DbSettings()  # Reads from environment variables
db.init(settings=settings)

Important: Call db.init() exactly once per process. It is not idempotent.

---

Session Management

Access the current context-local session via the db singleton:

from bedrock.database import db

# Get current session (auto-created on first access per context)
session = db.session

# Use in service layer
def create_user(name: str, email: str) -> UserModel:
    user = UserModel(name=name, email=email)
    db.session.add(user)
    db.session.commit()
    return user

Transaction Scopes with session_scope()

The recommended way to manage transactions — automatic commit on success, rollback on exception:

from bedrock.database import db

# Automatic commit on clean exit
with db.session_scope() as session:
    user = UserModel(name="Alice", email="alice@example.com")
    session.add(user)
    # Commits automatically when block exits without exception

# Automatic rollback on exception
try:
    with db.session_scope() as session:
        session.add(UserModel(name="Bob", email="bob@example.com"))
        raise ValueError("Something went wrong")  # Rolls back
except ValueError:
    pass  # "Bob" was never persisted

Key behaviors:

• Binds session to ContextVar so db.session works inside the block
• Commits on clean exit, rolls back on any exception
• Always closes session and resets ContextVar on exit
• Raises RuntimeError if called while a session is already bound

Independent Sessions with independent_session()

For side-effect writes that must not interfere with the current transaction:

from bedrock.database import db

with db.session_scope() as session:
    order = Order(total=100)
    session.add(order)

    # Write audit log in separate transaction
    with db.independent_session() as audit_session:
        audit_log = AuditLog(action="order_created")
        audit_session.add(audit_log)
        # Commits independently — does not affect outer session

    # db.session still returns the outer session
    assert db.session is session

Use cases:

• Audit logging (persists even if main transaction rolls back)
• Event publishing or outbox patterns
• Reads that must not see uncommitted data from the caller
• Background tasks with their own transaction lifecycle

Manual Session Binding

For middleware or request lifecycle management:

# Bind an external session to current context
db.set_session(my_session)

# Clear binding when done (closes session, then clears ContextVar)
db.clear_session()

SessionFactory as Context Manager

SessionFactory can be used as a context manager — closes and unbinds session on exit:

with db.session_factory:
    session = db.session
    session.add(UserModel(name="Charlie", email="charlie@example.com"))
    session.commit()
# Session is closed and unbound here

---

Migrations

Bedrock uses per-app Alembic branches via MigrationsManager. Each app gets its own branch label (equal to its import path), preventing cross-app migration conflicts.

Setup

from bedrock.database import MigrationsManager

mgr = MigrationsManager(
    registry=apps,          # Populated ModuleRegistry
    database_url="postgresql://user:pass@localhost/mydb",
)

First Install — ensure_schema

status = mgr.ensure_schema("myproject.modules.users")
# Returns: "created" | "upgraded" | "up-to-date"

Behavior:

• No revisions applied: Creates tables from SQLAlchemy metadata, stamps at head
• Behind head: Runs upgrade to bring branch current
• At head: No-op

Creating Revisions

mgr.revision(
    app_import_path="myproject.modules.users",
    message="add email column",
    autogenerate=True,  # default
)

Upgrade / Downgrade

mgr.upgrade("myproject.modules.users")             # upgrade to head
mgr.upgrade("myproject.modules.users", target="+1") # one step forward
mgr.downgrade("myproject.modules.users", target="-1")
mgr.downgrade("myproject.modules.users", target="base")

CLI Commands

bedrock db revision <app> -m "description"
bedrock db upgrade <app> [target]
bedrock db downgrade <app> <target>
bedrock db heads <app>
bedrock db current <app>
bedrock db history <app>
bedrock db uninstall <app>

Migration Directory Layout

Each app stores migrations alongside its code:

my_app/users/
├── migrations/
│   ├── abc123_2025-01-15_add_users_table.py
│   └── def456_2025-02-01_add_email_column.py
├── models.py
└── manifest.yaml

---

Query Patterns

search_filter_sort_paginate

The primary query builder for filtered, sorted, paginated results.

from bedrock.database import db
from bedrock.database.service import search_filter_sort_paginate

result = search_filter_sort_paginate(
    db_session=db.session,
    model=UserModel,
    filter_specs=[
        {"field": "is_active", "op": "==", "value": True},
    ],
    sort_key="name",
    sort_dir="asc",
    page=1,
    limit=10,
)

Return value:

{
    "items": [<UserModel>, ...],
    "total": 42,
    "page_info": {
        "total": 42,
        "limit": 10,
        "offset": 0,
        "page": 1,
        "query": "",
        "filters": [...],
        "paginated": True,
        "has_more": True,
    },
}

Full Parameter Reference

---

Filter Syntax

Filters are JSON-serializable dicts passed as filter_specs to search_filter_sort_paginate.

Basic Filter

{"field": "name", "op": "==", "value": "Alice"}

Available Operators

Negation

Prefix the field name with ! to negate any operator:

{"field": "!is_active", "op": "==", "value": True}  # WHERE NOT is_active = True

Boolean Combinators

Combine filters with or / and:

# OR
{"or": [
    {"field": "age", "op": ">", "value": 30},
    {"field": "name", "op": "==", "value": "Bob"},
]}

# AND (explicit)
{"and": [
    {"field": "is_active", "op": "==", "value": True},
    {"field": "role", "op": "in", "value": ["admin", "staff"]},
]}

Nested Filters — Dot Notation (JOIN)

Use . to traverse relationships via SQL JOIN:

# Filter users by their department's name (joins department table)
{"field": "department.name", "op": "==", "value": "Engineering"}

Nested Filters — Colon Notation (any/has)

Use : to traverse relationships via any() or has() (auto-detected from relationship type):

# Filter users who have any order with status "shipped"
{"field": "orders:status", "op": "==", "value": "shipped"}

# Multi-level nesting
{"field": "orders:items:product_name", "op": "ilike", "value": "%widget%"}

Filter Examples

# Pagination with multiple filters
result = search_filter_sort_paginate(
    db_session=db.session,
    model=UserModel,
    filter_specs=[
        {"field": "is_active", "op": "==", "value": True},
        {"field": "email", "op": "ilike", "value": "%@company.com"},
        {"or": [
            {"field": "role", "op": "==", "value": "admin"},
            {"field": "age", "op": ">=", "value": 21},
        ]},
    ],
    sort_key="name",
    sort_dir="asc",
    page=1,
    limit=25,
)

# Between filter
{"field": "created_at", "op": "between", "value": ["2025-01-01", "2025-12-31"]}

# Fuzzy search (wraps value with %)
{"field": "name", "op": "fuzzy_search", "value": "alic"}
# Equivalent to: WHERE name ILIKE '%alic%'

---

Relationships

Define relationships using SQLAlchemy 2.0 Mapped syntax:

Foreign Key + Relationship

from bedrock.database.base import BedrockModel
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship


class DepartmentModel(BedrockModel):
    __tablename__ = "departments"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    # One-to-many: department has many users
    users: Mapped[list["UserModel"]] = relationship(back_populates="department")


class UserModel(BedrockModel):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    department_id: Mapped[int] = mapped_column(ForeignKey("departments.id"))

    # Many-to-one: user belongs to a department
    department: Mapped["DepartmentModel"] = relationship(back_populates="users")

Querying Across Relationships

Once relationships are defined, use dot/colon notation in filters:

# JOIN-based: filter users in "Engineering" department
filter_specs=[{"field": "department.name", "op": "==", "value": "Engineering"}]

# any/has-based: departments that have any active user
filter_specs=[{"field": "users:is_active", "op": "==", "value": True}]

---

Common Patterns

CRUD in Service Layer

from bedrock.database import db
from bedrock.database.service import search_filter_sort_paginate


def get_user(user_id: int) -> UserModel | None:
    """Fetch a single user by primary key."""
    return db.session.get(UserModel, user_id)


def list_users(page: int = 1, limit: int = 10, filters: list | None = None) -> dict:
    """List users with filtering and pagination."""
    return search_filter_sort_paginate(
        db_session=db.session,
        model=UserModel,
        filter_specs=filters or [],
        page=page,
        limit=limit,
        sort_key="name",
        sort_dir="asc",
    )


def create_user(name: str, email: str) -> UserModel:
    """Create a new user."""
    user = UserModel(name=name, email=email)
    db.session.add(user)
    db.session.commit()
    db.session.refresh(user)
    return user


def update_user(user_id: int, **kwargs) -> UserModel:
    """Update a user's fields."""
    user = db.session.get(UserModel, user_id)
    user.update(**kwargs)  # CrudMixin method
    db.session.commit()
    return user


def delete_user(user_id: int) -> None:
    """Delete a user."""
    user = db.session.get(UserModel, user_id)
    db.session.delete(user)
    db.session.commit()

Pagination Helper

def paginated_response(result: dict) -> dict:
    """Transform search_filter_sort_paginate output for API response."""
    return {
        "data": [item.dict() for item in result["items"]],
        "pagination": {
            "total": result["page_info"]["total"],
            "page": result["page_info"]["page"],
            "limit": result["page_info"]["limit"],
            "has_more": result["page_info"]["has_more"],
        },
    }

Show All (No Pagination)

result = search_filter_sort_paginate(
    db_session=db.session,
    model=UserModel,
    show_all=True,
)
all_users = result["items"]

---

Configuration

Database settings use environment variables with the DATABASE_ prefix. DbSettings is a Pydantic BaseSettings subclass, so it reads from the environment automatically.

Environment Variables

SQLite

export DATABASE_TYPE=sqlite
export DATABASE_DRIVER=pysqlite
export DATABASE_SCHEMA=bedrock.db

For in-memory databases, set DATABASE_SCHEMA=:memory:.

PostgreSQL

export DATABASE_TYPE=postgresql
export DATABASE_DRIVER=psycopg2
export DATABASE_HOST=localhost
export DATABASE_PORT=5432
export DATABASE_USERNAME=myuser
export DATABASE_PASSWORD=mypassword
export DATABASE_SCHEMA=myapp

MySQL

export DATABASE_TYPE=mysql
export DATABASE_DRIVER=pymysql
export DATABASE_HOST=localhost
export DATABASE_PORT=3306
export DATABASE_USERNAME=root
export DATABASE_PASSWORD=secret
export DATABASE_SCHEMA=myapp

Programmatic Configuration

Pass a DbSettings instance directly to db.init():

from bedrock.database import db, DbSettings

settings = DbSettings(
    TYPE="postgresql",
    DRIVER="psycopg2",
    HOST="localhost",
    PORT=5432,
    USERNAME="admin",
    PASSWORD="secret",
    SCHEMA="myapp",
)

db.init(settings=settings)

You can also override the connection URL entirely:

db.init(url="postgresql+psycopg2://admin:secret@localhost:5432/myapp")

---

Model Observer

The observer system hooks into SQLAlchemy's before_flush event to invoke callbacks when models are inserted, updated, or deleted.

Event Identifiers

Registering Observers

Use the @observes_model decorator:

from bedrock.database import BedrockModel
from bedrock.database.observer import observes_model, observer
from sqlalchemy import Column, Integer, String

class Catalog(BedrockModel):
    __tablename__ = "catalog"
    id = Column(Integer, primary_key=True)
    name = Column(String(100))

@observes_model(Catalog, "on_insert", "on_delete")
def log_catalog_changes(session, target, identifier):
    print(f"Catalog {target.id} was {identifier}")

The callback receives three arguments: the session, the model instance, and the event identifier string.

Callback Signature

callback(session, target, identifier) -> None

• session: The active SQLAlchemy session being flushed.
• target: The model instance that triggered the event.
• identifier: One of "on_insert", "on_update", or "on_delete".

Custom Observer Instances

Pass a custom ModelObserver via the observer keyword:

from bedrock.database.observer import ModelObserver

custom_observer = ModelObserver()

@observes_model(Catalog, "on_insert", observer=custom_observer)
def handle_insert(session, target, identifier):
    # Only fires on the custom observer
    pass

Weak References

Observer callbacks are stored as weak references. Bound methods use a WeakMethod wrapper to prevent memory leaks. If the object owning a bound method is garbage collected, the callback is silently skipped.

---

DatabaseManager API

The db singleton is a DatabaseManager instance. It owns the engine, session factory, and settings. All access goes through this object.

Error Handling

Accessing db.engine, db.session, or db.settings before calling db.init() raises DatabaseNotConfiguredError. Always initialize the database before using any of these properties.

Dependency Injection Guide

Reference for Bedrock's dependency injection container in bedrock.di.

Table of Contents

1. Core API 2. Lifetimes 3. Registering Services 4. Resolving Services 5. Scoped Services 6. Overrides for Tests 7. Decorators 8. Using DI in Modules 9. Testing Utilities 10. Anti-Patterns

---

Core API

Module: bedrock.di

Public exports

from bedrock.di import Container, Lifetime, container, provider, inject

Global singleton

Default vs custom container

• Prefer the default container for normal module/application wiring
• Create Container() manually only when you need an isolated service graph
• Good custom-container scenarios: tests, sandboxes, plugin isolation, short-lived worker flows
• If you do not need isolation, stay on the default container for consistency with Bedrock bootstrap injection

---

Lifetimes

Rules

• Use SINGLETON for long-lived shared services
• Use TRANSIENT when each caller should get a fresh instance
• Use SCOPED only when an explicit scope boundary exists
• Scoped services raise ScopeError if resolved without an active scope

---

Registering Services

container.register(key, *, factory, lifetime=Lifetime.SINGLETON)

Register a factory under a type key or string key.

from bedrock.di import Lifetime, container


class InventoryService:
    def check(self, sku: str) -> int:
        return 10


container.register(
    InventoryService,
    factory=InventoryService,
    lifetime=Lifetime.SINGLETON,
)

container.register_instance(key, instance)

Register a pre-built object as a singleton.

config = {"region": "ap-southeast-1"}
container.register_instance("app.config", config)

Registering into a custom container

from bedrock.di import Container, Lifetime

custom = Container()
custom.register(InventoryService, factory=InventoryService, lifetime=Lifetime.SINGLETON)
custom.register_instance("app.config", {"region": "sandbox"})

Duplicate registration

Both register() and register_instance() raise DuplicateServiceError if the key already exists.

---

Resolving Services

container.resolve(key)

service = container.resolve(InventoryService)
config = container.resolve("app.config")

container.is_registered(key)

if container.is_registered(InventoryService):
    service = container.resolve(InventoryService)

Failure mode

Resolving an unknown key raises ServiceNotFoundError.

---

Scoped Services

container.scope(name="default")

Use a context manager to create a scope boundary.

from bedrock.di import Lifetime, container


class Session:
    def close(self) -> None:
        print("closed")


container.register(Session, factory=Session, lifetime=Lifetime.SCOPED)

with container.scope("request"):
    first = container.resolve(Session)
    second = container.resolve(Session)
    assert first is second

Cleanup

When the scope exits, Bedrock calls .close() on scoped objects that provide it.

---

Overrides for Tests

container.override(key, instance)

Temporarily replace a registration.

from bedrock.di import container


class Mailer:
    def send(self, to: str, subject: str, body: str) -> None:
        raise NotImplementedError


class FakeMailer(Mailer):
    def send(self, to: str, subject: str, body: str) -> None:
        print(to)


with container.override(Mailer, FakeMailer()):
    mailer = container.resolve(Mailer)

Notes

• Overrides restore the previous registration on exit
• Overrides also work for keys that were not previously registered
• Nested overrides are supported

---

Decorators

@provider

Register a class in the default global container.

from bedrock.di import Lifetime, provider


@provider(lifetime=Lifetime.SINGLETON)
class AuditService:
    def write(self, message: str) -> None:
        print(message)

custom.provider

Bind decorator-based registration to a specific custom container.

from bedrock.di import Container, Lifetime

custom = Container()


class AuditService:
    def write(self, message: str) -> None:
        print(message)


@custom.provider(lifetime=Lifetime.SINGLETON)
class CustomAuditService(AuditService):
    pass


@custom.provider("audit.name")
class AuditName:
    def __str__(self) -> str:
        return "sandbox"

@inject(**mappings)

Resolve named keyword-only dependencies from the default global container.

from bedrock.di import inject


class Mailer:
    def send(self, to: str, subject: str, body: str) -> None:
        print(to)


@inject(mailer=Mailer)
def send_welcome(user_email: str, *, mailer: Mailer) -> None:
    mailer.send(user_email, "Welcome", "Hello")

Explicit kwargs override injected values.

custom.inject(**mappings)

Resolve dependencies from a specific custom container only.

from bedrock.di import Container

custom = Container()
custom.register_instance(Mailer, Mailer())


@custom.inject(mailer=Mailer)
def send_preview(user_email: str, *, mailer: Mailer) -> None:
    mailer.send(user_email, "Preview", "Hello")

The top-level provider and inject exports remain aliases for the default global container.

---

Using DI in Modules

The most common registration point is bootstrap.py.

from bedrock.di import Lifetime


class InventoryService:
    def check(self, sku: str) -> int:
        return 10


def on_load(*, container) -> None:
    container.register(
        InventoryService,
        factory=InventoryService,
        lifetime=Lifetime.SINGLETON,
    )

Bootstrap injection

Bootstrap hooks may request container by name:

• registry
• app
• container
• hooks

Declare only the parameters you need — the registry injects only what it finds in the signature.

---

Testing Utilities

Package: bedrock.testing

Available DI fixtures:

• clean_container
• di_container
• override_service

---

Anti-Patterns

• Do not turn the container into a hidden service locator for everything
• Do not register unrelated services eagerly at import time
• Do not use DI where direct local construction is clearer
• Do not use SCOPED without a real scope boundary
• Do not introduce a custom Container() unless isolation is part of the requirement

---

See Also

• references/module-guide.md
• references/signals-guide.md
• references/architecture.md

Hook System Guide

Reference for Bedrock's structured hook system in bedrock.hooks.

Table of Contents

1. Core API 2. Key Concepts 3. Declaring Hook Specs 4. Registering Implementations 5. Calling Hooks 6. Scanning Classes and Modules 7. Using Hooks in Modules 8. Validation and Introspection 9. Signals vs Hooks 10. Testing Utilities 11. Anti-Patterns

---

Core API

Module: bedrock.hooks

Public exports

from bedrock.hooks import HookRegistry, HookNamespace, hooks, hookspec, hookimpl

Global singleton

---

Key Concepts

Hook registry

HookRegistry stores hook specs and implementations by fully-qualified name.

Hook namespace

HookNamespace("auth") scopes hook names under a module-owned namespace:

• auth.authenticate
• auth.get_permissions

Hookspec vs hookimpl

Priority and firstresult

• Lower priority values run first
• firstresult=True stops dispatch after the first non-None result

---

Declaring Hook Specs

from bedrock.hooks import HookNamespace

auth_hooks = HookNamespace("auth")


@auth_hooks.spec(firstresult=True)
def authenticate(token: str) -> str | None:
    ...


@auth_hooks.spec
def get_permissions(user_id: str) -> list[str]:
    ...

Rules

• The spec owner defines the namespace
• Specs document the extension contract
• Use firstresult=True only when the caller wants exactly one winning implementation

---

Registering Implementations

Direct registration with @namespace.impl

from bedrock.hooks import HookNamespace

payment_hooks = HookNamespace("payment")


@payment_hooks.impl(priority=0)
def process(method: str, amount: int) -> str | None:
    if method == "card":
        return "charged"
    return None

Marker-based registration with @hookimpl

from bedrock.hooks import hookimpl


class JwtHooks:
    @hookimpl(priority=0)
    def authenticate(self, token: str) -> str | None:
        if token == "jwt-token":
            return "user-1"
        return None

---

Calling Hooks

namespace.call(name, **kwargs)

results = auth_hooks.call("get_permissions", user_id="user-1")

await namespace.acall(name, **kwargs)

results = await auth_hooks.acall("get_permissions", user_id="user-1")

namespace.call_robust(name, **kwargs)

results = auth_hooks.call_robust("get_permissions", user_id="user-1")

Return behavior

• normal hook: returns a list of all results in priority order
• firstresult=True: returns a list containing the first non-None result only
• robust calls: return (callable, result_or_exception) tuples

---

Scanning Classes and Modules

add_specs_from(obj)

from bedrock.hooks import HookNamespace, hookspec

search_hooks = HookNamespace("search")


class SearchSpecs:
    @hookspec
    def providers(self) -> list[str]:
        ...


search_hooks.add_specs_from(SearchSpecs)

add_impls_from(obj, module=...)

from bedrock.hooks import hookimpl


class ElasticHooks:
    @hookimpl(priority=0)
    def providers(self) -> list[str]:
        return ["elastic"]


search_hooks.add_impls_from(ElasticHooks(), module="elastic")

Use instances for implementation scanning so methods are bound correctly.

---

Using Hooks in Modules

Hooks are usually wired in bootstrap.py.

from modules.auth.hookspecs import auth_hooks


class ApiKeyHooks:
    def authenticate(self, token: str) -> str | None:
        if token == "api-key":
            return "user-2"
        return None


def on_load(*, app) -> None:
    auth_hooks.add_impls_from(ApiKeyHooks(), module=app.name)

Bootstrap injection

Bootstrap hooks may request:

• registry
• app
• container
• hooks

Declare only the parameters you need — the registry injects only what it finds in the signature.

---

Validation and Introspection

Validation

warnings = hooks.validate()

Warnings are emitted for:

• implementations without a matching spec
• specs with zero implementations

Introspection

auth_hooks.has_spec("authenticate")
auth_hooks.get_impls("authenticate")
auth_hooks.specs()
hooks.namespaces()

Reset

auth_hooks.reset()   # one namespace
hooks.reset()        # all namespaces

---

Signals vs Hooks

Rule of thumb

• Use signals when the sender should not care about return values
• Use hooks when the caller needs results, ordering, or first-match behavior

---

Testing Utilities

Package: bedrock.testing

Available hook fixtures:

• clean_hooks
• hook_registry
• hook_namespace

---

Anti-Patterns

• Do not use hooks when a normal function call is simpler
• Do not use signals when the caller needs ordered return values
• Do not let unrelated modules own another module's namespace
• Do not register implementations into a namespace that has no meaningful owning spec module

---

See Also

• references/module-guide.md
• references/signals-guide.md
• references/architecture.md

Module Creation & Lifecycle Guide

Complete guide to creating and structuring Bedrock modules for application development.

Table of Contents

1. Module Directory Structure 2. Manifest Schema 3. Entity Pattern 4. Exception Pattern 5. Bootstrap Hooks 6. Signal System 7. Settings Pattern 8. Installation Hooks 9. CLI Commands

---

Module Directory Structure

<name>/
├── __init__.py          # Package marker (can re-export public API)
├── manifest.yaml        # Required: module metadata
├── entities.py          # Pydantic models (extend BedrockEntity)
├── service.py           # Business logic (sync + async pairs)
├── exc.py               # Module exceptions (extend BedrockExc)
├── bootstrap.py         # Lifecycle hooks: on_load, ready, on_shutdown
├── models.py            # SQLAlchemy models (extend BedrockModel) — optional
├── installation.py      # install(), pre_install(), post_install() hooks — optional
└── commands.py          # Typer CLI app — optional

Not every module needs every file. Minimum viable module: __init__.py + manifest.yaml.

---

Manifest Schema

File: manifest.yaml at module package root

Fields

Example

title: User Management
version: "0.1.0"
description: User authentication and profile management
depends_on:
  - myproject.core
commands: commands:app

Key Notes

• The module name at runtime is the Python import path (e.g. myproject.modules.users), NOT a field in the manifest
• depends_on values are fully qualified import paths (e.g. myproject.core, not core)
• commands uses colon syntax: module_part:attribute (e.g. commands:app)

---

Entity Pattern

File: entities.py Base class: BedrockEntity from bedrock.entities

BedrockEntity is a Pydantic BaseModel with arbitrary_types_allowed=True. All domain entities extend it.

Example

from bedrock.entities import BedrockEntity


class UserBase(BedrockEntity):
    name: str
    email: str


class UserEntity(BedrockEntity):
    id: int
    is_active: bool = True

Rules

• All domain entities MUST extend BedrockEntity
• Use Pydantic v2 syntax (model_config = ConfigDict(...), not class Config)
• Entities are pure data models — no business logic, no HTTP types
• Use | None for optional fields with default None
• Write Model that is mainly for validation and data transfer to scehmas.py.

Schemas Pattern

File: schemas.py Base class: BedrockEntity from bedrock.entities

BedrockEntity is a Pydantic BaseModel with arbitrary_types_allowed=True. All domain entities extend it.

Example

from bedrock.entities import BedrockEntity

class UserCreate(BedrockEntity):
    name: str
    email: str
    password: str
    
class UserUpdate(BedrockEntity):
    name: str | None = None
    email: str | None = None
    password: str | None = None

---

Exception Pattern

File: exc.py Base class: BedrockExc from bedrock.exc

BedrockExc is an Exception subclass with a detail class attribute and an __init__ that accepts an optional msg to override it.

Example

from bedrock.exc import BedrockExc


class UserError(BedrockExc):
    """Base exception for user module."""
    detail: str = "User operation failed."


class UserNotFoundError(UserError):
    """Raised when user is not found."""
    detail: str = "User not found."


class DuplicateEmailError(UserError):
    """Raised when email already exists."""
    detail: str = "Email already registered."


class InvalidPasswordError(UserError):
    """Raised when password validation fails."""
    detail: str = "Invalid password."

Usage

# Raise with default detail
raise UserNotFoundError()  # → "User not found."

# Raise with custom message
raise UserNotFoundError("User with ID 42 not found")

# Catch all module errors
try:
    user_service.get_user(42)
except UserError as e:
    print(e.detail)

Rules

• Every exception MUST set a detail class attribute with a default message
• Constructor accepts optional msg to override detail
• Create a module-level base exception (e.g. UserError) for catching all module errors
• Use specific subclasses for distinct error conditions

---

Bootstrap Hooks

File: bootstrap.py Contract: Bedrock always calls hooks with keyword arguments. Use keyword-only signatures.

Available Hooks

Available Named Parameters

The registry inspects each hook's signature and injects only the parameters it declares:

Example

def on_load(*, registry, app) -> None:
    """Called during module installation."""
    if not registry.is_installed("myproject.core"):
        raise RuntimeError("myproject.core must be installed first")


def ready(*, registry, app, container, hooks) -> None:
    """Called after all modules are installed."""
    from .service import user_service
    user_service.initialize()
    # Register services in the DI container
    container.register(IUserRepo, factory=UserRepo, lifetime=Lifetime.SINGLETON)


def on_shutdown(*, registry, app) -> None:
    """Called during shutdown."""
    from .service import user_service
    user_service.cleanup()

Rules

• Use keyword-only signatures (def on_load(*, registry, app)) — Bedrock always passes keyword arguments
• Declare only the parameters you need; the registry injects only what it finds in the signature
• Hooks are called by the registry — do NOT import or call them manually
• ready is the most common hook for service initialization
• on_shutdown runs in REVERSE install order (last installed = first shutdown)

---

Signal System

Bedrock provides lifecycle signals for cross-module communication. Connect handlers to react when modules load, become ready, or shut down.

Available Signals

from bedrock.module.signals import (
    module_loaded,     # After each module's on_load hook
    module_ready,      # After each module's ready hook
    module_shutdown,   # After each module's on_shutdown hook
    registry_ready,    # After all modules ready
    registry_shutdown, # When registry shuts down
)

Connecting to Signals

from bedrock.module.signals import registry_ready, module_loaded


# Decorator form
@registry_ready.connect
def on_ready(sender, **kwargs):
    print("All modules ready!")


# Explicit connect
def on_module_loaded(sender, **kwargs):
    print(f"Module loaded: {sender}")

module_loaded.connect(on_module_loaded)

Sending Signals

Most signals are sent automatically by the registry. If you create custom signals in your module:

from bedrock.signal import Signal

# Define a custom signal
user_created = Signal("user_created")

# Send (sync) — in pure sync context, adapts async receivers
user_created.send(sender, user=new_user)

# Send (async) — canonical for mixed sync/async receivers
await user_created.asend(sender, user=new_user)

Sync/async contract: send() is sync-facing. In a pure sync context it can adapt async receivers. Inside a running event loop, it raises RuntimeError if an async receiver is reached. Use await asend() for async or mixed contexts. See references/signals-guide.md for details.

---

Signals vs Hooks

Bedrock has two cross-module communication mechanisms. Use the right one for the job:

Rule of thumb: If you need a return value to drive logic, use hooks. If you just want to notify listeners, use signals.

---

Dependency Injection

The DI container (bedrock.di.container) provides service registration and resolution with three lifetimes.

Registering Services

from bedrock.di import container, Lifetime

# Register a factory (singleton by default)
container.register(IUserRepo, factory=UserRepo, lifetime=Lifetime.SINGLETON)

# Register a pre-built instance
container.register_instance("cache", my_cache_service)

# Or use the @provider decorator (auto-registers in global container)
from bedrock.di import provider

@provider
class UserService:
    ...

@provider(IUserRepo, lifetime=Lifetime.TRANSIENT)
class PostgresUserRepo:
    ...

For normal Bedrock modules, prefer the default global container shown above. Reach for a custom Container() only when you need an isolated registration graph, such as tests or sandboxed plugin execution.

from bedrock.di import Container, Lifetime

custom = Container()


@custom.provider(IUserRepo, lifetime=Lifetime.SINGLETON)
class SandboxUserRepo:
    ...


@custom.inject(repo=IUserRepo)
def run_preview(*, repo: IUserRepo) -> None:
    ...

Resolving Services

from bedrock.di import container

# By type
repo = container.resolve(IUserRepo)

# By string key
cache = container.resolve("cache")

# Check registration
if container.is_registered(IUserRepo):
    repo = container.resolve(IUserRepo)

Scoped Services

from bedrock.di import container, Lifetime

container.register(IDbSession, factory=create_session, lifetime=Lifetime.SCOPED)

# Scoped services share one instance within a scope
with container.scope("request"):
    s1 = container.resolve(IDbSession)
    s2 = container.resolve(IDbSession)
    assert s1 is s2  # Same instance

Testing with Overrides

from bedrock.di import container

# Temporarily replace a service
with container.override(IUserRepo, FakeUserRepo()):
    # Code here sees the fake
    repo = container.resolve(IUserRepo)
    assert isinstance(repo, FakeUserRepo)
# Original restored automatically

Using @inject

from bedrock.di import inject

@inject(repo=IUserRepo, cache="cache")
def get_user_profile(user_id: int, *, repo, cache):
    cached = cache.get(f"user:{user_id}")
    if cached:
        return cached
    return repo.get_by_id(user_id)

---

Settings Pattern

Create module-level settings using BaseSettings. Wrap module-level singletons with SettingsProxy to defer environment variable reading until first attribute access, preventing import-time side effects.

Creating Module Settings

from pydantic_settings import BaseSettings, SettingsConfigDict
from bedrock.conf import SettingsProxy


class MyModuleSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="MYMODULE_")
    API_KEY: str = ""
    DEBUG: bool = False
    MAX_RETRIES: int = 3


my_settings: MyModuleSettings = SettingsProxy(MyModuleSettings)  # type: ignore[assignment]

Usage

# First access triggers actual env var reading
if my_settings.DEBUG:
    print(f"API key: {my_settings.API_KEY}")

Rules

• Use SettingsProxy for module-level singletons that may be imported before env vars are ready
• Use BaseSettings directly when env vars are guaranteed ready at construction time
• Use env_prefix in model_config to namespace env vars (e.g. MYMODULE_API_KEY)
• Never force early initialization by accessing settings at import time

Built-in Settings References

---

Installation Hooks

File: installation.py (optional) Used by: bedrock manage install

Available Hooks

Example

from bedrock.database import db


def pre_install() -> None:
    """Run before database migrations."""
    print("Preparing installation...")


def install() -> None:
    """Run after migrations — seed data, etc."""
    from .models import UserModel
    with db.session as session:
        if not session.query(UserModel).first():
            session.add(UserModel(name="admin", email="admin@example.com", hashed_password="..."))
            session.commit()


def post_install() -> None:
    """Run after install — final checks."""
    print("Installation complete!")

Invocation

bedrock manage install -A myproject.modules.users

---

CLI Commands

File: commands.py (optional) Framework: Typer

Creating Commands

import typer
import bedrock

app = typer.Typer(name="users", help="User management commands")


@app.command()
def create_user(name: str, email: str) -> None:
    """Create a new user."""
    bedrock.setup()
    from .service import user_service
    from .entities import UserCreateRequest
    user = user_service.create_user(UserCreateRequest(name=name, email=email, password="temp"))
    typer.echo(f"Created user: {user.name} ({user.email})")


@app.command()
def list_users() -> None:
    """List all users."""
    bedrock.setup()
    from .service import user_service
    users = user_service.list_users()
    for user in users:
        typer.echo(f"  {user.name} <{user.email}>")

Registration

In manifest.yaml:

commands: commands:app

The commands field is a relative import path (module:attribute). The registry resolves it to the full path (e.g. myproject.modules.users.commands:app) and mounts it.

Invocation

bedrock run --app myproject.modules.users create-user --name Alice --email alice@example.com
bedrock run --app myproject.modules.users list-users

---

Playbooks

What is a Playbook

A playbook is AI-readable documentation bundled inside a module package. It teaches agents how to use the module without scanning source code. When an agent encounters a module it doesn't recognize, it reads the playbook to learn the API, patterns, and gotchas.

Directory Structure

my_module/
├── __init__.py
├── manifest.yaml
├── service.py
└── playbook/
    ├── PLAYBOOK.md
    └── references/
        ├── api-reference.md
        └── examples/
            └── usage.py

PLAYBOOK.md is required. The references/ directory holds supplementary docs and code examples.

When to Write One

Rule of thumb: If another project will import this module but won't have its source code in the repo, write a playbook.

This applies to modules published as separate packages (PyPI, private repos) or shared across multiple applications. If the module lives in the same repo and agents can scan it directly, a playbook is optional.

What to Put in PLAYBOOK.md

The main playbook should answer five questions:

1. What does this module do? One paragraph summary. 2. How to import and initialize. Exact import paths and setup steps. 3. Core API. The main classes and functions developers will call. 4. Common patterns. Copy-pasteable code examples for typical workflows. 5. Anti-patterns. What NOT to do, and why.

Keep it concise. Agents scan quickly; they don't need prose. Show code over explanations.

CLI Access

# Read the main playbook
bedrock app playbook <module_import_path>

# Read a reference file
bedrock app playbook <module_import_path> <filename>

# Examples
bedrock app playbook myapp.auth
bedrock app playbook myapp.auth references/api-reference.md
bedrock app playbook myapp.auth examples/usage.py

Playbook vs Agent Skills

Agent skills load at conversation start and stay in context the entire time. If a project uses 10 modules with 10 skills, that's 10 sets of instructions competing for context window space, most irrelevant to the current task.

Playbooks solve this. They live inside the module package and are only read when the agent explicitly calls bedrock app playbook. No context pollution, no wasted tokens.

Module Hierarchy Reference

Bedrock organizes code into three levels: Project, Submodule, Domain Module. This is a development convention, not a runtime enforcement mechanism.

Three Module Types

Project

Top-level directory for your application. Serves as an organizational container.

A project can be a plain Python package (no manifest.yaml) or a Bedrock module (with manifest.yaml). It contains submodules as subdirectories.

# my_app/__init__.py — just a package marker, no lifecycle hooks

Submodule

Business-level module that participates in Bedrock's module lifecycle.

• MUST have manifest.yaml at its root
• Loaded by ModuleRegistry.install() or ModuleRegistry.populate()
• Can declare dependencies on other submodules via depends_on

# oms/manifest.yaml
title: Order Management System
description: Handles orders, fulfillment, and returns
version: 0.1.0
depends_on:
  - my_app.auth
commands: "commands:app"

Domain Module

Code organization convention WITHIN a submodule. No manifest.yaml. Invisible to Bedrock's module system.

Recommended files (all optional):

Domain modules are just Python packages. Import them normally:

from my_app.oms.sales_order.service import create_order

Directory Structure

my_app/                          # Project
├── __init__.py
├── settings.py
├── oms/                         # Submodule
│   ├── __init__.py
│   ├── manifest.yaml
│   ├── bootstrap.py
│   ├── sales_order/             # Domain Module
│   │   ├── __init__.py
│   │   ├── entities.py
│   │   └── service.py
│   └── purchase_order/          # Domain Module
│       ├── __init__.py
│       ├── entities.py
│       └── service.py
└── auth/                        # Submodule
    ├── __init__.py
    ├── manifest.yaml
    └── ...

Cross-Project Dependencies

A separate runtime (e.g., a worker) can depend on submodules from another project via depends_on. When both packages are on the Python path, depends_on resolves import paths normally.

my_app_worker/                   # Another project (e.g., a worker runtime)
├── __init__.py
├── manifest.yaml               # depends_on: ["my_app.oms", "my_app.auth"]
└── ...

# my_app_worker/manifest.yaml
title: Background Worker
version: 0.1.0
depends_on:
  - my_app.oms
  - my_app.auth

Convention, Not Enforcement

• ModuleRegistry does not distinguish between Project, Submodule, or Domain Module.
• All modules with a manifest.yaml are treated identically at runtime.
• The hierarchy helps teams organize code predictably. It does not affect loading, dependency resolution, or lifecycle behavior.

Signal System Reference

Blinker-derived event system with sync/async support and lifecycle signals. Implementation: bedrock.signal.

Need call/response with return values? Use the hook system (bedrock.hooks) instead. Signals are notification-only.

Table of Contents

1. Creating Signals 2. Connecting Receivers 3. Sending Signals 4. Sender Filtering 5. Context Managers 6. Lifecycle Signals 7. Weak References 8. Sync/Async Interop 9. Signals vs Hooks 10. Anti-Patterns

---

Creating Signals

Signal

Base class. Optional docstring for documentation:

from bedrock.signal import Signal

user_created = Signal("Emitted after a new user is persisted")
payment_failed = Signal()

NamedSignal

Subclass with a name attribute for identification and debugging:

from bedrock.signal import NamedSignal

audit_event = NamedSignal("app.audit", doc="Emitted for auditable actions")
print(audit_event.name)  # "app.audit"

Namespace

Dictionary-like container mapping names to NamedSignal instances. Signals are created on first access:

from bedrock.signal import Namespace

my_signals = Namespace()

# Creates NamedSignal on first call
login_signal = my_signals.signal("user.login", doc="User logged in")

# Second call returns the same instance
assert my_signals.signal("user.login") is login_signal

Default Namespace

Module-level convenience for quick signal creation:

from bedrock.signal import signal

# Creates or retrieves a NamedSignal in the default namespace
cache_miss = signal("cache.miss", doc="Cache key not found")

---

Connecting Receivers

A receiver is any callable accepting sender as its first positional argument plus optional keyword arguments.

signal.connect(receiver, sender, weak)

from bedrock.signal import Signal

user_created = Signal()

def log_user(sender, **kwargs):
    print(f"User created by {sender}: {kwargs.get('username')}")

# Connect to all senders (sender defaults to Signal.ANY)
user_created.connect(log_user)

# Connect to a specific sender only
user_created.connect(log_user, sender=admin_module)

# Disable weak referencing (required for closures/lambdas)
user_created.connect(log_user, weak=False)

Parameters:

• receiver: Callable to invoke on signal send.
• sender: Signal.ANY (default) or a specific object. Receiver only fires for matching sender.
• weak: True (default). Uses weakref to track receiver. Set to False for closures, lambdas, or when you need a strong reference.

Returns the receiver (allows decorator-style usage).

@signal.connect_via(sender) Decorator

Connects a receiver for a specific sender. Defaults to weak=False:

from bedrock.signal import Signal

invoice_created = Signal()

@invoice_created.connect_via(billing_module)
def handle_invoice(sender, **kwargs):
    """Only called when billing_module sends invoice_created."""
    process_invoice(kwargs["invoice"])

Disconnecting

user_created.disconnect(log_user)

# Disconnect from a specific sender only
user_created.disconnect(log_user, sender=admin_module)

---

Sending Signals

signal.send(sender, **kwargs)

Sync-facing dispatch. Behavior depends on whether an event loop is running:

• Pure sync context (no running event loop): Adapts async receivers automatically (using the default asgiref bridge).
• Inside a running event loop: Raises RuntimeError if an async receiver is reached. Use await signal.asend(...) or provide _async_wrapper.

Returns list[tuple[receiver, return_value]]:

results = order_placed.send(current_app, order_id="ORD-042")

for receiver, result in results:
    print(f"{receiver.__name__} returned {result}")

signal.send_robust(sender, **kwargs)

Like send(), but catches Exception subclasses from receivers. Returns exceptions in the result list instead of propagating:

results = order_placed.send_robust(current_app, order_id="ORD-042")

for receiver, result in results:
    if isinstance(result, Exception):
        log.error(f"{receiver.__name__} failed: {result}")
    else:
        log.info(f"{receiver.__name__} returned {result}")

Note: BaseException subclasses (e.g. KeyboardInterrupt) still propagate.

await signal.asend(sender, **kwargs)

Async dispatch. Canonical API for async and mixed-context code. Awaits async receivers natively. Sync receivers are wrapped via asyncio.to_thread by default:

results = await order_placed.asend(current_app, order_id="ORD-042")

await signal.asend_robust(sender, **kwargs)

Async equivalent of send_robust(). Captures exceptions from receivers:

results = await order_placed.asend_robust(current_app, order_id="ORD-042")

Custom Wrappers

Override how send() adapts async receivers by passing _async_wrapper:

• send() / send_robust(): _async_wrapper parameter to customize async-to-sync adaptation.
• asend() / asend_robust(): _sync_wrapper parameter to customize sync-to-async adaptation (defaults to asyncio.to_thread).

---

Sender Filtering

Signal.ANY

Sentinel meaning "any sender". This is the default for connect():

from bedrock.signal import Signal

data_changed = Signal()

# These are equivalent:
data_changed.connect(handler)
data_changed.connect(handler, sender=Signal.ANY)

Specific Sender

Pass a concrete sender to restrict which sender triggers a receiver:

from bedrock.signal import Signal

data_changed = Signal()

@data_changed.connect_via(inventory_module)
def on_inventory_change(sender, **kwargs):
    """Only fires when inventory_module sends data_changed."""
    refresh_stock(kwargs["item_id"])

# This triggers the receiver:
data_changed.send(inventory_module, item_id="SKU-99")

# This does NOT:
data_changed.send(billing_module, item_id="SKU-99")

Checking for Receivers

Guard expensive operations with has_receivers_for():

if data_changed.has_receivers_for(inventory_module):
    payload = build_expensive_payload()
    data_changed.send(inventory_module, payload=payload)

---

Context Managers

signal.connected_to(receiver, sender)

Temporarily connects a receiver for the duration of a with block. Disconnects on exit. Useful for testing:

from bedrock.signal import Signal

user_created = Signal()
received = []

def capture(sender, **kwargs):
    received.append(kwargs)

with user_created.connected_to(capture):
    user_created.send(app, username="alice")

assert len(received) == 1

# Receiver is disconnected after the block
user_created.send(app, username="bob")
assert len(received) == 1  # Still 1

signal.muted()

Temporarily suppresses all signal dispatch. No receivers are called while muted:

from bedrock.signal import Signal

audit = Signal()

@audit.connect_via(Signal.ANY, weak=False)
def log_audit(sender, **kwargs):
    write_audit_log(kwargs)

# Suppress during bulk import
with audit.muted():
    for record in bulk_data:
        import_record(record)
        audit.send(importer, action="import")  # Silently ignored

# Normal dispatch resumes here
audit.send(importer, action="complete")  # This fires

---

Lifecycle Signals

Defined in bedrock.module.signals. Emitted by the ModuleRegistry during startup and shutdown.

Available Signals

Import

from bedrock.module.signals import (
    module_loaded,
    module_ready,
    module_shutdown,
    registry_ready,
    registry_shutdown,
)

Usage

from bedrock.signal import Signal
from bedrock.module.signals import module_ready, registry_ready

@module_ready.connect_via(Signal.ANY, weak=False)
def on_module_ready(sender, **kwargs):
    module = kwargs["module"]
    print(f"Module ready: {module.name}")

@registry_ready.connect_via(Signal.ANY, weak=False)
def on_all_ready(sender, **kwargs):
    print("All modules initialized, safe to accept traffic")

Lifecycle Order

During registry.populate(modules):

1. Each module installed in dependency order, module_loaded per module 2. Each module's ready hook called, module_ready per module 3. Registry marked ready, registry_ready once

During registry.shutdown():

4. Each module's on_shutdown hook called in reverse order, module_shutdown per module 5. Registry signals completion, registry_shutdown once

---

Weak References

By default, receivers are stored as weak references (weak=True).

Behavior by Receiver Type

When to Use weak=False

# WRONG: closure is immediately garbage collected
def setup():
    def handler(sender, **kwargs):
        print("received")
    signal.connect(handler)  # handler has no strong reference!

# CORRECT: disable weak refs for closures
def setup():
    def handler(sender, **kwargs):
        print("received")
    signal.connect(handler, weak=False)

The connect_via decorator defaults to weak=False, making it safe for typical decorator usage.

Checking Connected State

Use the receivers attribute for a quick boolean check:

if order_placed.receivers:
    print("At least one receiver is connected")

---

Sync/Async Interop

send() is a sync-facing API. Its behavior depends on whether an event loop is running:

• Pure sync context (no running event loop): send() can adapt async receivers automatically (using the default asgiref bridge).
• Inside a running event loop: If send() encounters an async receiver, it raises RuntimeError instructing you to use await signal.asend(...) or provide _async_wrapper.

asend() is the canonical API for async and mixed-context code. It natively awaits async receivers and wraps sync receivers via asyncio.to_thread.

from bedrock.signal import Signal

order_placed = Signal()

async def async_handler(sender, **kwargs):
    await notify_external_service(kwargs["order_id"])

order_placed.connect(async_handler, weak=False)

# In a running event loop — THIS RAISES RuntimeError!
order_placed.send(app, order_id="ORD-001")
# RuntimeError: Cannot send to an async receiver with send().
# Use await signal.asend(...) or provide _async_wrapper.

# CORRECT: use asend() in async context
await order_placed.asend(app, order_id="ORD-001")

Rule: In mixed sync/async codebases, prefer asend() as the default dispatch method. It handles both sync and async receivers seamlessly. Sync receivers are automatically wrapped via asyncio.to_thread.

---

Signals vs Hooks

Bedrock offers two distinct mechanisms for cross-module communication:

Use signals when you want to notify listeners about something that happened, and you don't care about return values. Example: "a user was created, update your cache."

Use hooks when you want to define an extension point where implementations contribute behavior or return values. Example: "authenticate this request, first valid result wins."

# SIGNAL: notification only
from bedrock.signal import Signal
user_created = Signal("user_created")

@user_created.connect
def on_user_created(sender, **kwargs):
    send_welcome_email(kwargs["user"])  # Fire-and-forget

user_created.send(sender, user=new_user)


# HOOK: call/response with return values
from bedrock.hooks import HookNamespace
auth = HookNamespace("auth")

@auth.spec(firstresult=True)
def authenticate(request):
    """Return user if authenticated, None otherwise."""

@auth.impl(priority=10)
def check_token(request):
    if valid_token(request.token):
        return get_user_from_token(request.token)
    return None  # Let next impl try

request = Request(token="...")
results = auth.call("authenticate", request=request)
user = results[0] if results else None

---

Anti-Patterns

Don't call `send()` from async code when async receivers are connected. Raises RuntimeError. Use await signal.asend() instead, or pass _async_wrapper to provide custom adaptation logic.

Don't rely on receiver execution order. The default set_class is Python's unordered set. If ordered dispatch is needed, provide an ordered set implementation via Signal.set_class.

Don't connect receivers inside hot loops. Each connect() call updates internal bookkeeping dicts. Connect once at module load time, not per-request.

Don't forget `weak=False` for closures. A closure or lambda connected without weak=False gets garbage collected before any signal is sent.

Don't use signals for synchronous control flow. Signals are notification (fire-and-forget). If you need a return value to drive logic, call the function directly.

Don't send lifecycle signals manually. The ModuleRegistry owns lifecycle signal emission. Sending module_ready or registry_ready yourself breaks invariants.