Building Custom Modules

This page covers extending Kirak by building new modules. If you want to use the built-in modules (Notifications, Payments, Storage, Scheduler, AI), see the Modules section in the navigation.

All of Kirak’s modules – Auth, Notifications, Payments, Storage, Scheduler, AI – are built on BaseModule. You can build your own module with the same hook, logging, and dispatch infrastructure, and register it so it’s accessible as kirak.mymodule just like the built-ins.

`BaseModule` Overview

File: kirak/core/base_module.py

BaseModule provides:

Hook registry (defaultdict of lists, keyed by (model_name, hook_type))
dispatch() pipeline (auth dependency -> before hooks -> operation -> after hooks)
run_hooks() with asyncio timeout and error isolation
Shared rotating log file (kirak.log) with request correlation IDs
execute_query() for raw SQL access
HookBuilder for fluent hook registration (.on().hook())
process_request() for query parameter type coercion

Creating a Custom Module

1. The Module Class

Extend BaseModule and declare your module’s identity in _setup_module. The logger is created after _setup_module returns, so self.slug set there is what the logger and log file will be named.

from kirak.core.base_module import BaseModule
from kirak.core.exceptions import ValidationError
from typing import Any, Dict, Optional


class ReportingModule(BaseModule):

    def __init__(self, kirak=None):
        # Set instance vars BEFORE super().__init__() -- _setup_module runs inside it.
        # Receive the full Kirak facade -- use it for fetch(), create(), execute_query(), etc.
        # Never accept or store a raw db_pool; go through the facade instead.
        self._kirak = kirak
        self._config = None
        super().__init__()  # no db_pool, no slug= -- both handled in _setup_module

    def _setup_module(self):
        # This runs BEFORE the logger is created -- do NOT call self.logger here.
        self.slug = "reporting"           # logger will be getLogger("reporting")
        self.model_name = "reporting"
        self._default_model = "reporting"  # enables shorthand: @kirak.reporting.hook(...)

        from .operations.generate_report import _generate_report
        self.operations = {
            "_generate_report": _generate_report,
        }

    async def generate_report(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        if not params:
            raise ValidationError("params are required")
        return await self.dispatch(self.model_name, "generate_report", params)

    def router(self):
        from .router import get_router
        return get_router(self)

Key rules:

Accept the Kirak facade (kirak=k), not db_pool. The facade exposes fetch(), create(), execute_query(), and everything else the module needs. Raw pool access is an implementation detail that modules should not depend on.
Set self.slug in _setup_module – that is the single source of truth for logger name and log file.
Do not call self.logger inside _setup_module. Use it freely anywhere after super().__init__() returns.
Set lazy-loaded state (self._config = None, self._kirak = kirak) before super().__init__(), because _setup_module runs inside it.

2. Operations

Operations are plain async functions that live outside the class. They receive the module instance as their first argument:

from kirak.core.error_response import create_success_response

async def _generate_report(module, model_name: str, params: dict, current_user: dict):
    report_type = params.get("type", "daily")
    user_id = current_user.get("user_id")

    # All data access goes through the Kirak facade stored on the module.
    # Model-based operations:
    result = await module._kirak.fetch("orders", {
        "user_id": user_id,
        "status": "completed",
    })
    rows = result.get("data", [])

    # Raw SQL is also available when models don't cover the query:
    # rows = await module._kirak.execute_query("SELECT ...", (user_id,), fetch="all")

    return create_success_response(data={
        "report_type": report_type,
        "records": rows,
        "generated_by": user_id,
    })

self.operations = {
    "_generate_report": _generate_report,
}

dispatch() calls self.operations["_" + operation](self, model_name, payload, current_user) and runs before/after hooks automatically.

3. Registering with the Module Registry

Calling Kirak.register_module() from your module’s __init__.py makes your module:

Accessible as kirak.reporting anywhere a Kirak instance is available
Accepted in modules=['reporting'] in create_kirak_app()
Lazy-loaded on first access (zero cost if unused)

from .reporting import ReportingModule
from kirak.kirak_instance import Kirak

__all__ = ["ReportingModule"]


def _factory(k):
    """
    Factory receives the live Kirak instance (k) and returns the module instance.
    Pass k itself -- not k.db or k._crud_engine. The module uses the facade for
    all data access, staying decoupled from internal implementation details.
    """
    return ReportingModule(kirak=k)


Kirak.register_module("reporting", _factory)

Import the module somewhere before create_kirak_app() runs to trigger registration:

import myapp.modules.reporting  # registers ReportingModule

from kirak import create_kirak_app

app = create_kirak_app(
    models_path="./models/",
    modules=["reporting"],      # now valid -- it's in the registry
)

For a distributable package (kirak-reporting on PyPI), the pattern is identical – users import kirak_reporting and the module self-registers.

4. Using the Module

Once registered, access it as a first-class attribute of the Kirak instance:

# In routes
@app.post("/reports")
async def create_report(payload: dict):
    kirak = app.get_kirak()
    return await kirak.reporting.generate_report(payload)

# In hooks
@kirak.reporting.hook("after_generate_report")
async def notify_on_report(result):
    # result is the dict returned by _generate_report
    return result

5. The Router

from fastapi import APIRouter, Body, Request
from kirak.core.context import set_request_context


def get_router(reporting):
    router = APIRouter()  # prefix is set by mount_routers(), not here

    @router.post("/generate")
    async def generate(request: Request, payload: dict = Body(...)):
        set_request_context(request)  # required for request correlation
        return await reporting.generate_report(payload)

    return router

Always call set_request_context(request) at the start of every route – it’s required for the request ID and user ID to appear in logs and hook context.

Hooks in Custom Modules

Setting _default_model enables the shorthand hook syntax on your module:

# Shorthand (requires _default_model set)
@kirak.reporting.hook("after_generate_report")
async def log_report(result):
    return result

# Equivalent verbose form
@kirak.reporting.on("reporting").hook("after_generate_report")
async def log_report(result):
    return result

Hooks fire automatically before and after every dispatch() call. The hook mutation contract:

Return a value -> that value becomes the payload for the next hook.
Return None -> payload passes through unchanged.
Exceptions are caught and logged – they never crash the request.
Hooks time out at 5 s by default (hook_timeout_seconds in kirak.json).

App-Specific Modules (Without Registry)

For a module that belongs only to your application (not reusable across projects), you can skip register_module and wire it manually in on_kirak_ready:

def on_kirak_ready(kirak):
    from myapp.modules.reporting import ReportingModule
    kirak.reporting = ReportingModule(kirak=kirak)

    @kirak.reporting.hook("after_generate_report")
    async def after_report(result):
        return result

    kirak.app.include_router(kirak.reporting.router(), prefix="/reporting")

app = create_kirak_app(
    models_path="./models/",
    on_kirak_ready=on_kirak_ready,
)

This is simpler for one-off modules, but register_module is preferred for anything you plan to share or reuse.

Logging

Every module shares one log file at {log_path}/kirak.log (log_path in kirak.json, default logs). Log rotation triggers at 5 MB, keeping 1 backup. All log lines automatically include request ID and user ID from the request context, plus the emitting module’s name (%(name)s), so you can still tell which module logged a given line even though they all land in one file.

self.logger.info("Generating %s report for user %s", report_type, user_id)
self.logger.warning("Empty result for report %s", report_type)
self.logger.error("Report failed", exc_info=True)

Database Access

All data access goes through the Kirak facade stored on the module (self._kirak).

# Model-based -- preferred; respects access rules and fires hooks
result = await self._kirak.fetch("orders", {"status": "pending"})
rows = result.get("data", [])

await self._kirak.create("report_log", {
    "data": {"user_id": user_id, "type": report_type}
})

# Raw SQL -- use when models don't cover the query
count = await self._kirak.execute_query(
    "SELECT COUNT(*) as total FROM orders WHERE status = %s",
    params=("pending",),
    fetch="one",
)

The Kirak facade manages connection acquisition and release automatically.

Module Checklist

When building a production-ready module: