DEV Community: David Landup

Guide to Database Transactions in Python with Mitsuki

David Landup — Sun, 04 Oct 2026 18:09:27 +0000

What a database transaction is, why a service method that writes twice needs one, and how @Transactional gives you one in Mitsuki.

A database transaction groups several statements into one unit of work that either happens completely or not at all.

It's the typical failsafe that prevents a bank transfer from debiting one account without crediting the other, an order from being saved without its line items, or a sign-up from creating a user without their settings.

Mitsuki has @Transactional just like Spring, Django has transaction.atomic(), and with plain SQLAlchemy (and therefore FastAPI) you open one yourself with session.begin().

This guide explains what a transaction does, and then uses Mitsuki's @Transactional to make a small ledger API safe, from a single transfer up to batches that skip the transfers that fail.

Everything here is the transactions example from the Mitsuki repository.

Mitsuki

Mitsuki is an batteries-included, structure-first web development framework for Python. It's opinionated, lightweight and performant.

What are Transactions?

In SQL, a transaction is everything between BEGIN and COMMIT:

BEGIN;
UPDATE accounts SET balance = balance - 50 WHERE id = 1;
UPDATE accounts SET balance = balance + 50 WHERE id = 2;
COMMIT;    -- both changes become permanent together
-- or
ROLLBACK;  -- neither change is applied

If anything fails between BEGIN and COMMIT (an error, divine intervention, a crash, a dropped connection), the database discards every change made so far.

The guarantees are usually summed up as ACID:

Property	Meaning
Atomicity	All statements take effect, or none do
Consistency	Constraints hold before and after the transaction
Isolation	Other connections don't see your half-finished work
Durability	Once committed, the changes survive a crash

For this, a transaction lives on a single database connection (at least for non-distributed databases).

A BEGIN on one connection has no effect on another, so for several statements to be in one transaction, they all have to run on the same connection.

When are Transactions Useful?

By default, a Mitsuki repository call opens a connection, runs its statement, commits and returns the connection to the pool. That's the right default for a single write, but not for two writes that belong together.

The example's transfer withdraws from one account and deposits into the other:

@Service()
class TransferService:
    def __init__(self, accounts: AccountService, audit: AuditService):
        self.accounts = accounts
        self.audit = audit

    async def transfer(self, transfer: Transfer) -> None:
        ...
        await self.accounts.withdraw(transfer.source_id, transfer.amount)
        await self.accounts.deposit(transfer.target_id, transfer.amount)

Without a transaction, a transfer to an account that doesn't exist withdraws the money, fails on the deposit, and returns an error.

The withdrawal has already been committed:

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":70},{"id":2,"owner":"bob","balance":80}]

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 99, "amount": 50}'
{"error":"Account 99 does not exist"}

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":20},{"id":2,"owner":"bob","balance":80}]

Alice is 50 down but nobody received it. Of course, you can catch the exception, and return the money to Alice, but that's another operation that needs to be persisted.

Instead of persisting one operation, then persisting a revert operation, you can simply... rollback the original operation before it's even applied. This is where transactions come into play!

Example Application

The transactions example runs against PostgreSQL.

Pull the repository, then let's run the app:

$ cd examples/transactions
$ docker compose up -d
$ pip install "mitsuki[postgresql]"
$ python app.py

The app's structure:

transactions/
├── app.py
├── application.yml
├── docker-compose.yml
└── src/
    ├── domain/          # Account, AuditEntry, AccountNotFound, InsufficientFunds
    ├── repositories/    # AccountRepository, AuditRepository
    ├── services/        # AccountService, TransferService, AuditService
    └── controllers/     # /api/accounts, /api/transfers, /api/audit

Let's open two accounts in the database to share balance between:

$ curl -X POST http://localhost:8000/api/accounts \
    -H "Content-Type: application/json" \
    -d '{"owner": "alice", "balance": 100}'
{"id":1,"owner":"alice","balance":100}

$ curl -X POST http://localhost:8000/api/accounts \
    -H "Content-Type: application/json" \
    -d '{"owner": "bob", "balance": 50}'
{"id":2,"owner":"bob","balance":50}

Making a Transfer Atomic with `@Transactional`

Add @Transactional() to the method:

from mitsuki import Service, Transactional

@Service()
class TransferService:
    ...

    @Transactional()
    async def transfer(self, transfer: Transfer) -> None:
        ...
        await self.accounts.withdraw(transfer.source_id, transfer.amount)
        await self.accounts.deposit(transfer.target_id, transfer.amount)

The whole method now runs in one transaction. It commits when the method returns, and rolls back when it raises. The exception is then re-raised, so the controller still turns it into an error response:

@PostMapping("/transfers")
@Consumes(Transfer)
async def transfer(self, body: Transfer = RequestBody()) -> ResponseEntity:
    try:
        await self.transfers.transfer(body)
    except AccountNotFound as error:
        return ResponseEntity.not_found({"error": str(error)})
    except InsufficientFunds as error:
        return ResponseEntity.conflict({"error": str(error)})
    return ResponseEntity.ok({"status": "completed"})

A transfer that goes through, then the same failing transfer as before:

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 2, "amount": 30}'
{"status":"completed"}

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 99, "amount": 50}'
{"error":"Account 99 does not exist"}

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":70},{"id":2,"owner":"bob","balance":80}]

The withdrawal from Alice was rolled back with the failed deposit. Neither the repositories nor AccountService changed: built-in repository methods, Query DSL methods, @Query methods and get_connection() all join the transaction in progress.

How Transactions Work in Mitsuki

When transfer starts, Mitsuki opens a connection, runs BEGIN, and binds the connection to the current request through a ContextVar.

Every repository call awaited inside the method, finds that connection and runs on it instead of opening its own.

When the method finishes, Mitsuki commits or rolls back and returns the connection to the pool.

Mitsuki only decides where transactions begin and end. Atomicity, locking and isolation are the database's job, reached through SQLAlchemy.

Propagation: When Transactional Methods Call Each Other

AccountService.withdraw and deposit are transactional as well, because they're also safe to call on their own:

@Service()
class AccountService:
    def __init__(self, accounts: AccountRepository):
        self.accounts = accounts

    @Transactional()
    async def withdraw(self, account_id: int, amount: int) -> Account:
        account = await self.get(account_id)
        if account.balance < amount:
            raise InsufficientFunds(
                f"Account {account_id} holds {account.balance}, not {amount}"
            )
        account.balance -= amount
        return await self.accounts.save(account)

    @Transactional()
    async def deposit(self, account_id: int, amount: int) -> Account:
        account = await self.get(account_id)
        account.balance += amount
        return await self.accounts.save(account)

So what happens when transfer, already in a transaction, calls them?

That's decided by propagation. Mitsuki supports three modes:

Propagation	Behaviour
`REQUIRED` (default)	Join the transaction in progress, or begin one if there is none
`REQUIRES_NEW`	Suspend the transaction in progress and run in a new, independent one
`NESTED`	Run inside the transaction in progress behind a savepoint, so a failure undoes only this block

`REQUIRED`: Join the Caller

withdraw() and deposit() use the default, so inside transfer() they join its transaction rather than committing on their own. That's why the failed deposit above could undo the withdrawal.

`REQUIRES_NEW`: An Audit Log That Survives Rollbacks

In the example app, every transfer request is written to an audit log, including the ones that fail. Though, if the audit entry joined the transfer's transaction, a failed transfer would roll its own audit entry back.

REQUIRES_NEW writes it in a separate transaction on a separate connection, which commits on its own:

@Service()
class AuditService:
    def __init__(self, entries: AuditRepository):
        self.entries = entries

    @Transactional(propagation=Propagation.REQUIRES_NEW)
    async def record(self, message: str) -> None:
        await self.entries.save(AuditEntry(message=message))

transfer() records the request before moving any money:

@Transactional()
async def transfer(self, transfer: Transfer) -> None:
    await self.audit.record(
        f"Transfer of {transfer.amount} from {transfer.source_id} "
        f"to {transfer.target_id} requested"
    )
    await self.accounts.withdraw(transfer.source_id, transfer.amount)
    await self.accounts.deposit(transfer.target_id, transfer.amount)

After the transfers above, let's add one Alice can't cover:

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 2, "amount": 500}'
{"error":"Account 1 holds 70, not 500"}

$ curl http://localhost:8000/api/audit
[{"id":1,"message":"Transfer of 30 from 1 to 2 requested"},
 {"id":2,"message":"Transfer of 50 from 1 to 99 requested"},
 {"id":3,"message":"Transfer of 500 from 1 to 2 requested"}]

Two of the three transfers rolled back, and all three audit entries are kept.

`NESTED`: A Batch That Skips Failures

In the example app, a batch endpoint runs several transfers at once. If any transfer fails, we want to skip it and still keep the rest.

With REQUIRED, a failed transfer would doom the whole batch. With REQUIRES_NEW, each transfer would commit on its own, so a crash halfway through would leave half a batch committed.

NESTED runs each transfer behind a savepoint, which is a marker inside the transaction that the database can roll back to without abandoning the transaction itself:

@Transactional()
async def transfer_all(self, transfers: List[Transfer]) -> List[bool]:
    """Run every transfer that can be covered and skip the rest."""
    completed = []
    for transfer in transfers:
        try:
            async with transaction(propagation=Propagation.NESTED):
                await self.transfer(transfer)
            completed.append(True)
        except (AccountNotFound, InsufficientFunds):
            completed.append(False)
    return completed

transaction() is the context-manager form of @Transactional, with the same options. Here it wraps a single loop iteration.

A failed transfer rolls back to its savepoint, and the batch carries on:

$ curl -X POST http://localhost:8000/api/transfers/batch \
    -H "Content-Type: application/json" \
    -d '{"transfers": [
          {"source_id": 2, "target_id": 1, "amount": 20},
          {"source_id": 1, "target_id": 2, "amount": 1000},
          {"source_id": 1, "target_id": 2, "amount": 10}
        ]}'
{"completed":[true,false,true]}

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":80},{"id":2,"owner":"bob","balance":70}]

The two transfers that could be covered moved the money, and the batch committed them together at the end.

Rollback Rules

Any exception rolls back, including asyncio.CancelledError, so a request cancelled mid-transfer never commits half of it.

To commit despite a particular exception, list it in no_rollback_for:

@Transactional(no_rollback_for=(PaymentDeclined,))
async def checkout(self, cart: Cart):
    ...

The exception is still re-raised so you can log it, handle it depending on your business logic, etc.

Isolation

Atomicity covers failures within one transaction. Isolation covers transactions running at the same time.

Two withdrawals of 80 from a balance of 100, running concurrently, can both read 100 before either writes, both pass the balance check, and both commit.

Whether that happens depends on the isolation level, which you set per method. For example:

@Transactional(isolation=Isolation.SERIALIZABLE)
async def withdraw(self, account_id: int, amount: int) -> Account:
    ...

At SERIALIZABLE, the database aborts one of the two withdrawals, and a retry sees the new balance.

The isolation section of the documentation walks through what each level allows, and how PostgreSQL, MySQL and SQLite differ.

Limitations

Async methods only. @Transactional on a sync method raises TypeError.
A transaction holds a connection until it ends. Keep slow work, such as HTTP calls to other services, out of transactional methods, or enough concurrent requests will exhaust the pool.
No asyncio.gather inside a transaction. A connection runs one statement at a time, so repository calls from concurrent tasks inside a transaction raise TransactionException. Await them one after another.
SQLite allows one writer at a time. Propagation modes like REQUIRES_NEW can't write while its caller holds the write lock. Note that underlying databases/engines have different behavior.

Next Steps

Run the transactions example and try the batch endpoint with your own transfers.
For every option, including class-level @Transactional, get_connection() inside a transaction comparison, read the Transactions documentation.
Learn how the repositories used here work in Guide to Mitsuki Repositories: From Zero to Full CRUD.

Mitsuki API Observability with Grafana and Prometheus

David Landup — Sat, 03 Oct 2026 17:20:36 +0000

A copy-paste-friendly tutorial for running a Mitsuki application with Prometheus and Grafana, and reading its request, component, scheduler and system metrics on ready-made dashboards.

Introducing Automatic Instrumentation covered what Mitsuki records and the two endpoints it exposes.

This post runs the full stack: the application, Prometheus scraping it every five seconds, and Grafana with two dashboards provisioned on startup.

It's the usual Prometheus-and-Grafana setup, the same one you would build for a Spring Boot or FastAPI service.

The difference is that the dashboard comes with the framework, and you get this and much more without writing a single query:

Everything here is the instrumentation_demo example from the Mitsuki repository: a users-and-orders API with a scheduled task.

Mitsuki

Mitsuki is a web development framework. It's opinionated, lightweight and performant.

Running the Stack

From the example directory:

cd examples/instrumentation_demo
docker compose up -d --build

Once it is up:

Service	Address
Mitsuki application	`http://localhost:8000`
Prometheus	`http://localhost:9090`
Grafana	`http://localhost:3000` (no login)

The Compose File

# docker-compose.yml
services:
  mitsuki:
    build:
      context: ../..
      dockerfile: examples/instrumentation_demo/Dockerfile
    container_name: instrumentation-demo-app
    ports:
      - "8000:8000"
    networks:
      - monitoring

  # Emits the Grafana dashboard bundled with Mitsuki into a shared volume,
  # which Grafana provisions from. Runs once and exits.
  dashboard-init:
    build:
      context: ../..
      dockerfile: examples/instrumentation_demo/Dockerfile
    container_name: instrumentation-demo-dashboard-init
    command: ["mitsuki", "grafana-dashboard", "-o", "/dashboards"]
    volumes:
      - grafana-dashboards:/dashboards
    networks:
      - monitoring

  prometheus:
    image: prom/prometheus:latest
    container_name: instrumentation-demo-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    networks:
      - monitoring

  grafana:
    image: grafana/grafana:latest
    container_name: instrumentation-demo-grafana
    ports:
      - "3000:3000"
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
    volumes:
      - ./grafana/provisioning:/etc/grafana/provisioning
      - grafana-dashboards:/var/lib/grafana/dashboards
      - ./grafana/dashboards:/var/lib/grafana/demo-dashboards:ro
      - grafana-storage:/var/lib/grafana
    networks:
      - monitoring
    depends_on:
      prometheus:
        condition: service_started
      dashboard-init:
        condition: service_completed_successfully

networks:
  monitoring:
    driver: bridge

volumes:
  grafana-storage:
  grafana-dashboards:
  prometheus-data:

The parts that matter:

dashboard-init runs mitsuki grafana-dashboard -o /dashboards once and exits. The framework dashboard ships inside the mitsuki package, and this command writes it into the volume Grafana reads.
condition: service_completed_successfully makes Grafana wait until the dashboard file exists. service_started would race it.
Two dashboard sources. grafana-dashboards holds the framework dashboard; ./grafana/dashboards holds a second dashboard for this demo's custom metrics.
GF_AUTH_ANONYMOUS_* lets you open Grafana without creating an account. It gives every visitor admin rights, so keep it to local use.

Configuring Prometheus

# prometheus.yml
global:
  scrape_interval: 5s

scrape_configs:
  - job_name: 'mitsuki'
    metrics_path: '/metrics/prometheus'
    static_configs:
      - targets: ['mitsuki:8000']

metrics_path points at /metrics/prometheus, the text-format endpoint. /metrics is the JSON summary, which Prometheus can't parse.

The target uses the Compose service name, mitsuki, which resolves on the monitoring network.

Prometheus scrapes from inside that network, so the application's metrics.allowed_ips has to include Docker's address range, 172.16.0.0/12. The demo's application.yml already does - see Securing the Endpoints for the full list.

To check that Prometheus is scraping, open Status → Target health at http://localhost:9090: the mitsuki target should be UP. If it's down, the address is usually refused by allowed_ips or the path is wrong.

Generating Traffic

Create some data first:

curl -X POST http://localhost:8000/api/users \
  -H "Content-Type: application/json" \
  -d '{"username": "alice", "email": "alice@example.com"}'

curl -X POST http://localhost:8000/api/orders \
  -H "Content-Type: application/json" \
  -d '{"user_id": 1, "product_type": "digital", "amount": 99.99, "region": "us-east"}'

Then keep traffic flowing for ten minutes: two list endpoints, and a lookup of a user that doesn't exist, which adds a 404 series to Status Code Distribution:

for i in $(seq 1 600); do
  curl -s -o /dev/null http://localhost:8000/api/users
  curl -s -o /dev/null http://localhost:8000/api/orders
  curl -s -o /dev/null http://localhost:8000/api/users/99
  sleep 1
done

The Framework Dashboard

Open http://localhost:3000 → Dashboards → Mitsuki Application Metrics. This is the dashboard mitsuki grafana-dashboard writes.

It only queries metrics Mitsuki itself emits, so it works unchanged for any Mitsuki application.

Row	Panels
Overview Statistics	Total Requests, Requests/Second, Avg Response Time, Error Rate, P95 Latency, P99 Latency
HTTP Performance - Per Route Breakdown	Response Time by Route (a table with average, p95, p99, rate and total per route)
HTTP Metrics Over Time	Request Rate by Route, Response Time Percentiles, Status Code Distribution
Component Performance	Component Metrics (a table per component)
Component Metrics Over Time	Component Call Rate, Component Duration (P95), Component Success vs Failure
Component Performance - Per Method	Call Rate by Method, P95 Duration by Method, Failure Rate by Method
Scheduler	Task Execution Rate, Task Duration (P95), Task Failure Rate, Running Tasks
System Resources	Memory Usage, CPU Usage

The HTTP rows break everything down by route template. Requests that match no route are grouped under <unmatched>:

" width="799" height="386">

The component rows go one level deeper than most HTTP instrumentation - into your own controllers, services and repositories:

And then per method, so a slow repository query shows up by name:

How the Panels Are Built

Three of the queries show how the rest are built.

Request Rate by Route is the per-second rate of the request counter over the last minute, summed per route template:

sum(rate(http_requests_total{path!~"/metrics.*"}[1m])) by (path)

P95 Latency estimates the 95th percentile from the histogram buckets. The buckets have to be summed by le, the bucket boundary label, before histogram_quantile can use them:

histogram_quantile(0.95, sum(increase(http_request_duration_seconds_bucket{path!~"/metrics.*"}[5m])) by (le)) * 1000

Why percentiles rather than the average? A mean hides the one request in a hundred that takes ten times as long - which is usually the one a user notices.

Error Rate is the share of requests that returned a 5xx:

((sum(http_requests_total{status=~"5.*",path!~"/metrics.*"}) or vector(0)) / sum(http_requests_total{path!~"/metrics.*"})) * 100

The or vector(0) keeps the panel at 0 instead of empty while no 5xx has happened yet. The 404s from the traffic loop don't count - they show up in Status Code Distribution instead.

Scheduled Tasks

Background jobs fail quietly: nobody gets an error page when a nightly job dies. The demo runs one, OrderReconciliationService.reconcile_orders, every ten seconds:

# src/services/order_reconciliation_service.py
from src.repositories.order_repository import OrderRepository

from mitsuki import Scheduled, Service


@Service()
class OrderReconciliationService:
    """
    Periodically checks stored orders for invalid amounts.

    As a @Scheduled task, every run is recorded in the scheduler metrics:
    executions, failures and duration. Creating an order with a non-positive
    amount makes every following run fail, which shows up on the dashboard's
    task failure panel.
    """

    def __init__(self, order_repo: OrderRepository):
        self.order_repo = order_repo

    @Scheduled(fixed_rate=10000)
    async def reconcile_orders(self):
        orders = await self.order_repo.find_all()
        invalid = [order.id for order in orders if order.amount <= 0]
        if invalid:
            raise ValueError(f"Orders with non-positive amounts: {invalid}")

The Scheduler row of the framework dashboard shows its execution rate, p95 duration, failure rate and whether a run is in progress:

The execution rate holds at 0.1 per second, one run every ten seconds. Grafana's automatic axis stretches the small jitter around it into a zigzag.

Task Failure Rate shows "No data" because no run has failed yet. To see a failure, create an invalid order:

curl -X POST http://localhost:8000/api/orders \
  -H "Content-Type: application/json" \
  -d '{"user_id": 1, "product_type": "digital", "amount": -1}'

The next run fails and logs the error with its traceback:

2026-09-29 07:32:45,991 - mitsuki - ERROR    - Scheduled task OrderReconciliationService.reconcile_orders failed with error: Orders with non-positive amounts: [4]
Traceback (most recent call last):
  File "/usr/src/app/mitsuki/core/scheduler.py", line 218, in task_loop

The failure is counted in both the scheduler metrics and the component metrics, since the service is instrumented:

component_calls_total{component="OrderReconciliationService",method="reconcile_orders",status="success"} 2.0
component_calls_total{component="OrderReconciliationService",method="reconcile_orders",status="failure"} 1.0
scheduler_task_executions_total{status="success",task="OrderReconciliationService.reconcile_orders"} 2.0
scheduler_task_executions_total{status="failure",task="OrderReconciliationService.reconcile_orders"} 1.0

The Task Failure Rate panel rises and stays up: every run fails until the invalid order is gone.

That makes it the panel to alert on. Its query is the rate of failed executions per task:

sum(rate(scheduler_task_executions_total{status="failure"}[5m])) by (task)

A Grafana alert rule on that query with the condition "is above 0" notifies you when any scheduled task starts failing.

The Demo Dashboard

The demo's OrderService also records custom metrics through InstrumentationProvider. Those only exist in this application, so they get their own dashboard: Dashboards → Demo → Demo Custom Metrics, provisioned from grafana/dashboards/demo-custom-metrics.json.

Panel	Query
Orders Created per Minute	`sum(rate(orders_created_total[5m])) by (product_type, region) * 60`
Database Write Operations	`rate(database_writes_total[5m])`
Rows Returned per Second	`sum(rate(database_rows_returned_total[5m])) by (table, query)`
Full Table Scans	`rate(database_full_scan_total[5m])`
Expensive Aggregations	`rate(expensive_aggregation_total[5m])`

How those metrics are recorded is the subject of Creating Custom Business Metrics in Mitsuki.

Using the Dashboard in Your Own Application

The framework dashboard is not tied to the demo. Write it into your own Grafana provisioning directory:

mitsuki grafana-dashboard -o ./grafana/dashboards

Wrote Grafana dashboard to grafana/dashboards/dashboard.json

Then point a Grafana dashboard provider at that directory, or import the file through the Grafana UI. The demo's provider for the framework dashboard:

# grafana/provisioning/dashboards/mitsuki.yml
apiVersion: 1

providers:
  # Framework dashboard, generated by `mitsuki grafana-dashboard` (dashboard-init).
  - name: Mitsuki
    orgId: 1
    folder: ''
    type: file
    disableDeletion: false
    updateIntervalSeconds: 10
    allowUiUpdates: true
    options:
      path: /var/lib/grafana/dashboards
      foldersFromFilesStructure: false
  ...

Note: The dashboard's panels don't name a datasource - they query whichever Prometheus datasource is Grafana's default. If your Grafana has several, make the one scraping your application the default, or change the panels' datasource after importing.

Cleaning Up

docker compose down      # stop the stack, keep Prometheus and Grafana data
docker compose down -v   # stop the stack and delete its volumes

Next Steps

Record business metrics of your own and add them to a dashboard: Creating Custom Business Metrics in Mitsuki - demo on GitHub.
Before running this beyond your machine, read the per-process and reverse-proxy caveats in Introducing Automatic Instrumentation.
For every configuration option, see the Instrumentation & Metrics documentation.

Introducing Automatic Python Web Application Instrumentation in Mitsuki

David Landup — Tue, 29 Sep 2026 14:14:17 +0000

Mitsuki's built-in instrumentation records what your application is doing (requests per route, latency, calls and failures per service method, scheduled task runs, process CPU and memory) and exposes it on two endpoints: a JSON summary for people and a Prometheus text endpoint for monitoring systems.

It plays the role Spring Boot Actuator and Micrometer play in a Spring application, or prometheus-fastapi-instrumentator in a FastAPI one.

The difference is that it also measures your own components, each invoked method's failure rate, call time, etc., not only the HTTP layer, and you enable it with a single line.

Everything below comes from the instrumentation_demo example in the Mitsuki repository, a small users-and-orders API.

Mitsuki

Mitsuki, is a web development framework, focused on bringing enterprise strength without the enterprise pain to Python.

It's opinionated, lightweight and performant.

Installing

Instrumentation samples process CPU and memory through psutil, which ships as an optional extra:

pip install "mitsuki[metrics]"

Enabling It

Put @Instrumented() on the application class:

# app.py
from mitsuki import Application, Instrumented, Value

@Instrumented()
@Application
class App:
    """
    Application entry point.

    @Instrumented() on the application instruments every 
@RestController, @Service, @Repository and @CrudRepository. It takes effect when instrumentation.enabled and metrics.enabled are set in application.yml.
    """

    port: int = Value("${server.port:8000}")


if __name__ == "__main__":
    App.run(host="0.0.0.0")

Then switch it on in configuration:

# application.yml
instrumentation:
  enabled: true
  track_memory: false

metrics:
  enabled: true
  path: /metrics

Both keys are needed. metrics.enabled turns on the metrics registry and its endpoints; instrumentation.enabled turns on the recording of requests and component calls.

track_memory adds Python's traced memory (via tracemalloc) on top of the process CPU and memory that are always sampled. Note that tracemalloc slows down every allocation. It's meant to be used for debugging rather than monitoring.

The decorator on the application instruments every @Controller, @Service and @Repository:

But also, the methods within them are tracked as well, so each method gets a failure rate, response time, etc:

To scope it to a smaller subset of components instead, put @Instrumented() on individual components and leave it off the application.

To exclude one component from application-wide instrumentation, use @Instrumented(enabled=False) on it. The Instrumentation & Metrics documentation covers both in greater detail.

What Gets Recorded

Metric	Type	Labels	Records
`http_requests_total`	counter	`method`, `path`, `status`	Every HTTP request
`http_request_duration_seconds`	histogram	`method`, `path`	Request latency
`component_calls_total`	counter	`component`, `method`, `status`	Every call to a public method of an instrumented component, with `status` of `success` or `failure`
`component_duration_seconds`	histogram	`component`, `method`	Method duration
`scheduler_task_executions_total`	counter	`task`, `status`	Every `@Scheduled` run
`scheduler_task_duration_seconds`	histogram	`task`	Task duration
`scheduler_tasks_running`	gauge	`task`	Runs in progress
`system_memory_bytes`	gauge	`type` (`rss`, `vms`)	Process memory, sampled every 5 seconds
`system_cpu_percent`	gauge		Process CPU, sampled every 5 seconds
`system_traced_memory_bytes`	gauge	`type` (`current`, `peak`)	Python traced memory, sampled every 5 seconds; only with `track_memory: true`

A few notes:

Methods are wrapped once, at startup. Every public method of an instrumented component is wrapped when the application starts. Methods starting with _ are not instrumented, and so are static methods, class methods and properties.
Repositories are included. For a @CrudRepository, the generated save and find_all, methods implemented from their names like find_by_user_id, @Query methods and methods you write yourself are all recorded under the repository's name.

This is the demo's component table on the Grafana dashboard set up in the next post, with both repositories listed next to the services and controllers:

Human Readable and Prometheus Format

/metrics returns a JSON summary with averages already computed. The demo README's example requests, one more order, and two user lookups, one of them for a user that doesn't exist:

curl http://localhost:8000/metrics

{
  "enabled": true,
  "timestamp": "2026-09-29T07:32:26.689899+00:00",
  "scheduler": {
    "tasks": [
      {
        "name": "OrderReconciliationService.reconcile_orders",
        "executions": 1,
        "failures": 0,
        "average_duration_ms": 9.51,
        "status": "idle"
      }
    ],
    "total_tasks": 1,
    "running_tasks": 0
  },
  "instrumentation": {
    "system": {
      "memory": {
        "rss_bytes": 55386112,
        "rss_mb": 52.82,
        "vms_bytes": 181608448,
        "vms_mb": 173.2
      },
      "cpu": {
        "percent": 0.0
      }
    },
    "http": {
      "total_requests": 11,
      "requests_by_method": {
        "POST": 5,
        "GET": 6
      },
      "responses_by_status": {
        "201": 5,
        "200": 5,
        "404": 1
      },
      "latency": {
        "avg_ms": 3.42,
        "total_seconds": 0.04
      }
    },
    "components": {
      "OrderRepository": {
        "calls": 7,
        "avg_duration_ms": 2.92,
        "methods": {
          "find_all": {
            "calls": 3,
            "avg_duration_ms": 3.92
          },
          "find_by_user_id": {
            "calls": 1,
            "avg_duration_ms": 4.03
          },
          "save": {
            "calls": 3,
            "avg_duration_ms": 1.56
          }
        }
      },
      ...
    }
  }
}

/metrics/prometheus returns every metric as labelled series in the Prometheus text format: the raw counters, gauges and histogram buckets behind that summary, plus any custom metrics your code records, which the JSON summary leaves out:

curl http://localhost:8000/metrics/prometheus

# HELP http_requests_total Total HTTP requests
# TYPE http_requests_total counter
http_requests_total{method="POST",path="/api/users",status="201"} 2.0
http_requests_total{method="GET",path="/api/users",status="200"} 1.0
http_requests_total{method="POST",path="/api/orders",status="201"} 3.0
...

	`/metrics`	`/metrics/prometheus`
Format	Nested JSON	Prometheus text format
Contents	Totals and averages since startup	Raw counters, gauges and histogram buckets
Per-route breakdown	No	Yes, via the `path` label
Percentiles	No	Yes, computed by Prometheus from the histograms
Custom metrics	No	Yes
Use it for	A quick look, health checks, scripts	Dashboards, alerting, history

Securing the Endpoints

Both endpoints expose your route table, component names and traffic volume, so in production they should obviously not be public.

metrics.allowed_ips restricts them to listed addresses and CIDR ranges. The demo allows your machine and Docker networks:

metrics:
  enabled: true
  path: /metrics
  # Addresses allowed to read the metrics endpoints. An empty list allows all.
  # - 127.0.0.1: requests from your machine when running app.py directly
  # - 172.16.0.0/12: Docker networks, which is where Prometheus scrapes from
  # - 192.168.0.0/16: requests from your machine to the container on Docker
  #   Desktop, which forwards them from 192.168.65.1
  allowed_ips: ["127.0.0.1", "172.16.0.0/12", "192.168.0.0/16"]

An empty list allows everyone. A refused request gets a 404 with the body Not Found for every HTTP method, so a client outside the list can't easily probe whether the endpoints are there. The refusal is logged as a warning on the server side. With allowed_ips: ["10.8.0.0/16"], a request from the same machine logs:

2026-09-29 16:31:09,251 - mitsuki - WARNING  - Metrics access denied for IP: 127.0.0.1

Note: The check uses the client address the server reports. Behind a reverse proxy or load balancer, Granian, the default server, reports the proxy as the client, so external requests look internal: allowlisting the proxy's address (for example 127.0.0.1, which the demo allows) opens the endpoints to anyone the proxy serves. In other words, if you use a load balancer or nginx or similar proxies, the external requests are translated into internal requests, and in 0.2.0 - the Granian engine won't be able to tell them apart. This is planned to be patched in future versions alongside broader IP allowlisting support.

Limitations

Metrics are per process. Each worker process keeps its own metrics. With server.workers greater than 1, each scrape reaches one worker, and Prometheus sees counters jump between workers' values.
Metrics are in memory. They reset when the application restarts. Prometheus handles counter resets, but the JSON endpoint only covers the time since startup.

Next Steps

Set up Prometheus and Grafana on top of these endpoints with the demo's Docker Compose file: Mitsuki API Observability with Grafana and Prometheus - demo on GitHub.
Record your own business metrics next to the automatic ones: Creating Custom Business Metrics in Mitsuki - demo on GitHub.
For every configuration option, see the Instrumentation & Metrics documentation.

How to Access the Request Object in Mitsuki Controllers

David Landup — Sun, 14 Dec 2025 14:26:36 +0000

In Mitsuki, you'd usually work abstractions like @QueryParam and RequestBody that handle the details of HTTP requests for you. However, sometimes you need to access the raw request object to read headers, inspect client details, or manage cookies.

This guide shows you how to directly access the underlying starlette.requests.Request object in your controller methods.

The `Request` Type Hint

The easiest way to get the request object is to add a parameter to your controller method and type-hint it as Request. Mitsuki's dependency injection system will automatically provide the object for you.

Here’s how you can inject the request to read its properties:

from mitsuki import RestController, GetMapping
from starlette.requests import Request

@RestController("/api")
class HomeController:
    @GetMapping("/info")
    # The : Request type-hinted object is resolved via Mitsuki's DI
    async def get_request_info(self, request: Request) -> dict:
        return {
            "user_agent": request.headers.get("User-Agent", "Unknown"),
            "client_host": request.client.host,
            "client_port": request.client.port,
            "cookies": request.cookies,
            "path": request.url.path,
        }

Because Mitsuki is built on Starlette, the injected object is a standard starlette.requests.Request instance, giving you access to its features.

Combining with Other Parameters

Injecting the request object doesn't interfere with Mitsuki's other features. You can combine it with path variables, query parameters, and request bodies:

from mitsuki import RestController, GetMapping, QueryParam
from starlette.requests import Request

@RestController("/api/users")
class UserController:
    @GetMapping("/{user_id}")
    async def get_user_and_log_ip(
        self,
        request: Request,
        user_id: int,
        verbose: bool = QueryParam(default=False)
    ) -> dict:
        # Log the client's IP address for auditing purposes... or whatever.
        logger.info(f"Request for user {user_id} received from IP: {request.client.host}")

        # Your regular business logic
        user = await self.service.get_user(user_id)

        return user

Next Steps

Explore the docs: For more on how requests are being handled, controllers work, etc. check out the official Mitsuki Controllers documentation.

Happy coding! ❀

Guide to Path Variables and Query Parameters in Mitsuki Controllers

David Landup — Sat, 29 Nov 2025 13:34:02 +0000

How to Handle URL Parameters in Mitsuki Controllers

A guide to reading path variables and query parameters with automatic type conversion.

Mitsuki

Mitsuki, is a web development framework, focused on bringing enterprise strength without the enterprise pain to Python.

It's opinionated, lightweight and performant.

Introduction

REST APIs usually involve reading dynamic data directly from the URL.

This data generally comes in two forms:

Path variables: which identify a specific resource (e.g., /posts/some_id)
Query parameters: which are usually used for sorting or filtering collections (e.g., /posts?popular=true).

Mitsuki makes handling both of these trivial and type-safe, allowing you to define complex URL-based data retrieval with minimal code.

Setup

For the purposes of this guide, let's create a simple Post entity, a @CrudRepository to access its data, and a @Service to contain our business logic:

from dataclasses import dataclass
from typing import List, Optional
from mitsuki import Entity, Id, CrudRepository, Service

# 1. The Entity: Defines the Post data model
@Entity()
@dataclass
class Post:
    id: int = Id()
    title: str = ""
    published: bool = True

# 2. The Repository: The data access layer for Posts
@CrudRepository(entity=Post)
class PostRepository:
    # Gets standard CRUD methods automatically.
    async def find_by_published(self, published: bool) -> List[Post]: ...

    # For custom filtering with pagination, we use @Query
    @Query("""
        SELECT p FROM Post p
        WHERE p.title LIKE :title_pattern AND p.published = :published
        ORDER BY p.id DESC
    """)
    async def search_paginated(self, title_pattern: str, published: bool, limit: int, offset: int) -> List[Post]: ...

# 3. The Service: Contains business logic
@Service()
class PostService:
    def __init__(self, repo: PostRepository):
        # Mitsuki automatically injects the PostRepository here
        self.repo = repo

    async def get_by_id(self, post_id: int) -> Optional[Post]:
        return await self.repo.find_by_id(post_id)

    async def search(self, q: str, published: bool, page: int, size: int) -> List[Post]:
        offset = page * size
        return await self.repo.search_paginated(
            title_pattern=f"%{q}%",
            published=published,
            limit=size,
            offset=offset
        )

Path Variables: Accessing Resource Identifiers

Path variables are used to capture parts of the URL path, most commonly for resource IDs. You define a path variable in your mapping decorator with curly braces {} and accept it as an argument in your controller method.

Let's create an endpoint to fetch a specific post by its ID:

from mitsuki import RestController, GetMapping
from typing import Optional
from ..services.post_service import PostService
from ..domain.post import Post

@RestController("/api/posts")
class PostController:
    def __init__(self, post_service: PostService):
        # The PostService is automatically injected by Mitsuki's DI container
        self.post_service = post_service

    @GetMapping("/{post_id}")
    async def get_post_by_id(self, post_id: int) -> Optional[Post]:
        print(f"Fetching post with ID: {post_id}")
        print(f"Type of post_id is: {type(post_id)}")

        post = await self.post_service.get_by_id(post_id)

        return post

There are two key things happening here:

Name Matching: Mitsuki automatically matches the {post_id} in the path to the post_id argument in the method.
Automatic Type Conversion: By type-hinting post_id: int, you tell Mitsuki to convert the incoming path segment from a string to an integer. If a client requests /api/posts/abc, Mitsuki will automatically respond with a 400 Bad Request error because "abc" cannot be converted to an integer.

Query Parameters: Filtering and Sorting

Query parameters are the key-value pairs that appear after the ? in a URL. They are ideal for optional filters, search queries, or pagination controls.

To read query parameters, you use the @QueryParam decorator on your method arguments. Let's add a search endpoint to our PostController.

# Continuing inside the PostController class...
from mitsuki import QueryParam

@GetMapping("/search")
async def search_posts(
    self,
    q: str = QueryParam(required=True),
    published: bool = QueryParam(default=True),
    page: int = QueryParam(default=0),
    size: int = QueryParam(default=10)
) -> List[Post]:
    # Example Request: /api/posts/search?q=mitsuki&published=true&page=0&size=20
    #
    # q: "mitsuki" (required string)
    # published: True (boolean, defaults to True)
    # page: 0 (integer, defaults to 0)
    # size: 20 (integer, defaults to 10)

    posts = await self.post_service.search(q, published, page, size)
    return posts

Let's break down how @QueryParam works:

Required Parameter: q: str = QueryParam(required=True) explicitly marks the q parameter as mandatory. If a client makes a request without it, Mitsuki will automatically return a 400 Bad Request error with a descriptive message.
Optional Parameter with Default: published: bool = QueryParam(default=True) is optional. If the published parameter is missing from the URL, its value will be True. The type hint ensures that values like "true", "1", or "on" are correctly converted to the boolean True.
Optional Parameter defaulting to None: If you were to define a parameter as size: Optional[int] = QueryParam(default=None), it would default to None if not provided. A plain QueryParam() also creates an optional parameter that defaults to None.

Next Steps

You've seen how Mitsuki's declarative approach makes it easy to read and validate data from a URL. This type-safe mechanism helps prevent common bugs and cleans up your controller logic.

Explore the docs: For a full list of available decorators and validation options, see the official Controllers & Request Handling documentation.
Handle Request Bodies: Now that you can read data from a URL, learn how to handle data sent in the request body by reading our guide to @RequestBody.

Happy coding! ❀

Guide to Mitsuki Repositories: From Zero to Full CRUD

David Landup — Sat, 29 Nov 2025 06:34:42 +0000

Introduction

The data access layer is the foundation of web applications, sitting at the bottom of the layered architectures that power enterprise code.

Mitsuki's data layer, centered around the @CrudRepository decorator, is designed to help you prototype in early stages as well as orchestrate more complex queries down the line, maximizing your time to focus on business logic. It allows you to define a complete, high-performance data access layer by writing a simple interface, letting the framework handle the implementation.

In this guide, you'll learn how to go from a simple interface to a data access layer, starting with zero-code CRUD and progressively adding more complex logic with Query DSL, native SQL, and custom SQLAlchemy queries.

Mitsuki

Mitsuki, is a web development framework, focused on bringing enterprise strength without the enterprise pain to Python.

It's opinionated, lightweight and performant.

Domain Objects - `@Entity`

Before any repository, we need a data model for that repository (i.e. an object that maps onto a database table).

In Mitsuki, this is a @dataclass decorated with @Entity. It tells Mitsuki that this object maps to a database table:

from dataclasses import dataclass
from mitsuki import Entity, Id, Column

@Entity()
@dataclass
class User:
    id: int = Id()
    name: str = ""
    email: str = Column(unique=True, default="")
    active: bool = True

This simple class is all we need to define the users table, complete with an auto-incrementing id, a name, a unique email, and an active status.

Alternatively, you can use a UUID:

from dataclasses import dataclass
from mitsuki import Entity, UUIDv7, Column

@Entity()
@dataclass
class User:
    id: int = UUIDv7()
    name: str = ""
    email: str = Column(unique=True, default="")
    active: bool = True

With support for UUIDv1, UUIDv4, UUIDv5, UUIDv7.

Level 1: Zero-Code CRUD with `@CrudRepository`

To get a fully functional (basic) data access layer for our User entity, we only need to define an empty class decorated with @CrudRepository:

from mitsuki import CrudRepository
from ..models.user import User

@CrudRepository(entity=User)
class UserRepository:
    """
    A complete data access layer for the User entity.
    All methods listed below are automatically implemented.
    """
    pass

By simply defining this interface, UserRepository is instantly equipped with a full suite of asynchronous CRUD methods:

Method Signature	Description
`save(entity: User) -> User`	Creates or updates a user.
`find_by_id(id: int) -> User	None`
`find_all(page, size, sort_by, sort_desc)`	Finds all users, with optional pagination and sorting.
`delete(entity: User) -> None`	Deletes a user object from the database.
`delete_by_id(id: int) -> None`	Deletes a user by their primary key.
`count() -> int`	Returns the total number of users.
`exists_by_id(id: int) -> bool`	Checks if a user with the given ID exists.

You can inject this repository into a service and use these methods immediately, with no implementation required:

from mitsuki import Service
from ..repositories.user_repository import UserRepository

@Service()
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create_new_user(self, name: str, email: str) -> User:
        new_user = User(name=name, email=email)
        # The .save() method is provided by @CrudRepository
        return await self.repo.save(new_user)

Level 2: Query DSL

What about custom queries? For common WHERE clauses, you don't need to write SQL. Mitsuki includes a Query DSL (Domain-Specific Language) that generates queries by parsing method names.

You just define the method signature in your repository; Mitsuki provides the implementation:

from typing import List, Optional
from mitsuki import CrudRepository
from ..models.user import User

@CrudRepository(entity=User)
class UserRepository:
    # Translates to: SELECT * FROM user WHERE email = ?
    async def find_by_email(self, email: str) -> Optional[User]: ...

    # Translates to: SELECT * FROM user WHERE active = ?
    async def find_by_active(self, active: bool) -> List[User]: ...

    # You can combine fields and operators
    # Translates to: SELECT * FROM user WHERE active = ? AND age > ?
    async def find_by_active_and_age_greater_than(
        self, active: bool, age: int
    ) -> List[User]: ...

    # Also works for counting and checking existence
    # Translates to: SELECT COUNT(*) FROM user WHERE active = ?
    async def count_by_active(self, active: bool) -> int: ...

This declarative approach is simple, readable, and covers a wide range of common query patterns without a single line of SQL.

Level 3: Taking Control with `@Query`

For queries that are too complex or long for the Query DSL, the @Query decorator gives you full control to write your own SQL.

Mitsuki supports two syntaxes:

SQLAlchemy ORM-style
Native SQL

SQLAlchemy ORM-style Syntax (Default)

This syntax queries your Entities and attributes, not your database tables and columns. It's the recommended approach as it's database-agnostic and less prone to errors if you happen to change your column names:

from mitsuki import Query

@CrudRepository(entity=User)
class UserRepository:
    # ... other methods

    @Query("""
        SELECT u FROM User u
        WHERE u.age BETWEEN :min_age AND :max_age
        AND u.active = :active
        ORDER BY u.name
    """)
    async def find_active_in_age_range(
        self, min_age: int, max_age: int, active: bool = True
    ) -> List[User]: ...

FROM User u: We select from the User entity, aliased as u.
WHERE u.age: We filter on the .age attribute of the entity.
:min_age: Named parameters in the query are automatically mapped to the method's arguments.

Native SQL Syntax

If you need to use a database-specific feature or prefer writing raw SQL, you can use native=True. This queries the database tables and columns directly:

from mitsuki import Query

@CrudRepository(entity=User)
class UserRepository:
    # ... other methods

    # Example using a PostgreSQL-specific function
    @Query("""
        SELECT * FROM "user"
        WHERE email ILIKE :pattern
        AND created_at > NOW() - INTERVAL '30 days'
    """, native=True)
    async def find_by_email_pattern(self, pattern: str) -> List[User]: ...

FROM "user": We select from the actual database table named user.
ILIKE: A PostgreSQL-specific operator for case-insensitive pattern matching.

Level 4: Custom SQLAlchemy Methods

For the highest level of control, especially for dynamically constructed queries, you can simply write a complete method implementation using SQLAlchemy Core, similar to how you would in Flask or FastAPI.

Every repository provides a self.get_connection() method that gives you direct, managed access to a SQLAlchemy connection:

from sqlalchemy import select, and_
from mitsuki.data import get_database_adapter

@CrudRepository(entity=User)
class UserRepository:
    # ... other methods

    async def find_dynamically(
        self, name_pattern: Optional[str], min_age: Optional[int]
    ) -> List[User]:
        """Builds a query dynamically based on which filters are provided."""
        adapter = get_database_adapter()
        user_table = adapter.get_table(User)

        query = select(user_table)
        conditions = []

        if name_pattern:
            conditions.append(user_table.c.name.like(f"%{name_pattern}%"))
        if min_age is not None:
            conditions.append(user_table.c.age >= min_age)

        if conditions:
            query = query.where(and_(*conditions))

        async with self.get_connection() as conn:
            result = await conn.execute(query)
            # Manually map the results back to User objects
            return [User(**dict(row._mapping)) for row in result.fetchall()]

This approach gives you programmatic access to SQLAlchemy's powerful query builder, allowing for complex, conditional logic that would be difficult to express in a static string.

Conclusion

You've seen how Mitsuki's repositories provide a layered approach to data access.

You can start with zero-effort CRUD, graduate to the expressive Query DSL, and take full control with custom SQL or SQLAlchemy when you need it.

This allows you to be highly productive while never sacrificing power.

Next Steps

Explore the docs: For a full list of supported keywords and operators, check out the official Repositories & Data Layer documentation.
Build on your work: Now that you have a powerful repository, learn how to use it in a @RestController to expose your data via a REST API.

Happy coding! ❀

How to Create Your First Mitsuki REST API from Scratch

David Landup — Sat, 29 Nov 2025 05:48:59 +0000

A step-by-step guide to using the mitsuki init CLI, understanding the generated project, and running your first high-performance web server.

Mitsuki

Mitsuki, is a web development framework, focused on bringing enterprise strength without the enterprise pain to Python.

It's opinionated, lightweight and performant.

Bootstrapping Mitsuki Projects

When you get a spark of inspiration, the first thing that comes to mind isn't boilerplate setup. It's standard, usually the same, and you might even have a project starter skeleton sitting on your device already.

When starting something new - I want to focus on the system design, business logic and value proposition. Thus, I heavily use my own templates.

Let's go from an empty folder to a fully functional, high-performance API in a single minute!

Step 1: Install Mitsuki

Mitsuki is available on PyPI. Installing it will also install the CLI tool:

$ pip install mitsuki

Step 2: mitsuki init 🪄

Let's use the built-in CLI to scaffold a complete project structure.
Run the following command in your terminal:

$ mitsuki init

The CLI will launch an interactive wizard to guide you. Let's create a simple API for a blog:

$ mitsuki init
Application name: blog_api
Description (optional): API for my new blog
Database type [sqlite/postgresql/mysql] (sqlite):
Create starter domain? [Y/n]: y
Domain name (e.g., User, Product): Post
Add another domain? [y/N]: n

Successfully created Mitsuki application: blog_api

You can immediately spin it up and fire requests to it:

python3 -m blog_api.src.app

Starts the server:

2025-11-29 10:30:15,132 - mitsuki - INFO     -
2025-11-29 10:30:15,132 - mitsuki - INFO     -     ♡ ｡ ₊°༺❤︎༻°₊ ｡ ♡
2025-11-29 10:30:15,132 - mitsuki - INFO     -               _ __             __   _
2025-11-29 10:30:15,132 - mitsuki - INFO     -    ____ ___  (_) /________  __/ /__(_)
2025-11-29 10:30:15,132 - mitsuki - INFO     -   / __ `__ \/ / __/ ___/ / / / //_/ /
2025-11-29 10:30:15,132 - mitsuki - INFO     -  / / / / / / / /_(__  ) /_/ / ,< / /
2025-11-29 10:30:15,132 - mitsuki - INFO     - /_/ /_/ /_/_/\__/____/\__,_/_/|_/_/
2025-11-29 10:30:15,132 - mitsuki - INFO     -     °❀˖ ° °❀⋆.ೃ࿔*:･  ° ❀˖°
2025-11-29 10:30:15,132 - mitsuki - INFO     -
2025-11-29 10:30:15,132 - mitsuki - INFO     - :: Mitsuki ::                (v0.1.3)
2025-11-29 10:30:15,132 - mitsuki - INFO     -
2025-11-29 10:30:15,133 - _granian - INFO     - Starting granian...
2025-11-29 10:30:15,139 - _granian - INFO     - Listening at: http://127.0.0.1:8000

And you can just send requests to it:

curl -X POST http://localhost:8000/api/post \
-H "Content-Type: application/json" \
-d '{}'

The API will respond with the newly created post, complete with a uuid, created_at, and updated_at timestamp:

{
  "id": "018ecb71-d556-7000-8000-000000000001",
  "created_at": "2025-11-29T10:35:00.123456",
  "updated_at": "2025-11-29T10:35:00.123456"
}

However - it's worth taking a look at what's inside first, customizing your Post domain object, and understanding the structure first.

Step 3: Anatomy of a Mitsuki Project

While you can really have any structure you want in a Mitsuki project, a sensible default structure is baked into the scaffold.

Let's pop the hood and see what mitsuki init created for us inside the blog_api directory:

blog_api/
  blog_api/
    src/
      controller/       # 🚪 The front door for your API requests
      domain/           # 🏗️ The blueprint for your data (Post)
      repository/       # 🗄️ The bridge that talk to your database
      service/          # 🧠 The brain where your business logic lives
      app.py            # ❤️ The heart of your application
      __init__.py
    __init__.py
  application.yml       # ⚙️ The main control panel for settings
  application-dev.yml   # 🛠️ Settings for your development environment
  application-prod.yml  # 🚀 Settings for production
  .gitignore
  README.md

What's Inside? Fully Functional API

This isn't just a collection of empty folders; it's a fully wired application.

The Post domain you specified in the beginning was created, with a corresponding RestController, Service, and CrudRepository -giving you a working CRUD API right out of the box.

Your PostController controller imports your PostService and exposes endpoints to create posts, return existing ones, and delete them:

import uuid
from typing import List
from mitsuki import RestController, GetMapping, PostMapping, DeleteMapping, QueryParam
from ..service.post_service import PostService


@RestController("/api/posts")
class PostController:
    """REST API controller for Post resources."""

    def __init__(self, service: PostService):
        self.service = service

    @GetMapping("")
    async def list_all(
        self,
        page: int = QueryParam(default=0),
        size: int = QueryParam(default=100)
    ):
        """GET /api/posts?page=0&size=100"""
        return await self.service.get_all(page=page, size=size)

    @GetMapping("/{id}")
    async def get_by_id(self, id: str):
        """GET /api/posts/123e4567-e89b-12d3-a456-426614174000"""
        entity = await self.service.get_by_id(uuid.UUID(id))
        if not entity:
            return {"error": "Not found"}, 404
        return entity

    @PostMapping("")
    async def create(self, body: dict):
        """POST /api/posts"""
        entity = await self.service.create()
        return entity, 201

    @DeleteMapping("/{id}")
    async def delete(self, id: str):
        """DELETE /api/posts/123e4567-e89b-12d3-a456-426614174000"""
        deleted = await self.service.delete(uuid.UUID(id))
        if not deleted:
            return {"error": "Not found"}, 404
        return {"success": True}

Your Post @Entity, contains only the ID, created_at and updated_at fields. Don't forget to add other fields you want here, like a body, or title:

import uuid
from dataclasses import dataclass
from datetime import datetime
from mitsuki import Entity, UUIDv7, Field


@Entity()
@dataclass
class Post:
    """Post domain object."""
    id: uuid.UUID = UUIDv7()
    created_at: datetime = Field(update_on_create=True)
    updated_at: datetime = Field(update_on_save=True)

Your PostRepository contains all the CRUD implementations by default in Mitsuki, as well as pointers for Query DSL, and custom query methods:

from typing import Optional, List
from mitsuki import CrudRepository, Query, Modifying
from ..domain.post import Post


@CrudRepository(entity=Post)
class PostRepository:
    """
    Repository for Post entities.

    Auto-implemented CRUD methods (you can just call these as-is):
    - save(entity) - Create or update
    - find_by_id(id) - Find by primary key
    - find_all(page, size, sort_by, sort_desc) - Paginated query
    - delete(entity) - Delete entity
    - delete_by_id(id) - Delete by ID
    - count() - Count all entities
    - exists_by_id(id) - Check existence

    Auto-generated query methods (define signature only):
    Examples:
        async def find_by_name(self, name: str) -> Optional[Post]: ...
        async def find_by_status(self, status: str) -> List[Post]: ...
        async def find_by_created_at_greater_than(self, created_at) -> List[Post]: ...

    Custom @Query methods:
        @Query("SELECT e FROM Post e WHERE e.status = :status ORDER BY e.created_at DESC")
        async def find_active_sorted(self, status: str) -> List[Post]: ...

    Custom @Modifying queries (UPDATE/DELETE):
        @Modifying
        @Query("UPDATE Post SET status = :status WHERE id = :id")
        async def update_status(self, id: int, status: str) -> int: ...

    Custom methods with direct database access:
        async def custom_complex_query(self, param: str) -> List[dict]:
            async with self.connection as conn:
                result = await conn.execute("SELECT * FROM table WHERE field = :param", {"param": param})
                return [dict(row) for row in result]
    """

    # Add your custom query methods here
    pass

And your PostService, where your business logic lives, currently just wraps the repository for database operations. Add your business-specific logic here:

import uuid
from typing import List, Optional
from mitsuki import Service
from ..domain.post import Post
from ..repository.post_repository import PostRepository


@Service()
class PostService:
    """Service layer for Post business logic."""

    def __init__(self, repo: PostRepository):
        self.repo = repo

    async def get_all(self, page: int = 0, size: int = 100) -> List[Post]:
        """Get all posts with pagination"""
        return await self.repo.find_all(page=page, size=size)

    async def get_by_id(self, id: uuid.UUID) -> Optional[Post]:
        """Get post by ID"""
        return await self.repo.find_by_id(id)

    async def create(self) -> Post:
        """Create new post"""
        entity = Post(id=None)
        return await self.repo.save(entity)

    async def delete(self, id: uuid.UUID) -> bool:
        """Delete post by ID"""
        exists = await self.repo.exists_by_id(id)
        if not exists:
            return False
        await self.repo.delete_by_id(id)
        return True

Step 4: Liftoff! Running Your Server 🚀

Time to spin up the server:

cd blog_api
python3 -m blog_api.src.app

You'll be greeted by the Mitsuki startup log:

2025-11-29 10:30:15,132 - mitsuki - INFO     -
2025-11-29 10:30:15,132 - mitsuki - INFO     -     ♡ ｡ ₊°༺❤︎༻°₊ ｡ ♡
2025-11-29 10:30:15,132 - mitsuki - INFO     -               _ __             __   _
2025-11-29 10:30:15,132 - mitsuki - INFO     -    ____ ___  (_) /________  __/ /__(_)
2025-11-29 10:30:15,132 - mitsuki - INFO     -   / __ `__ \/ / __/ ___/ / / / //_/ /
2025-11-29 10:30:15,132 - mitsuki - INFO     -  / / / / / / / /_(__  ) /_/ / ,< / /
2025-11-29 10:30:15,132 - mitsuki - INFO     - /_/ /_/ /_/_/\__/____/\__,_/_/|_/_/
2025-11-29 10:30:15,132 - mitsuki - INFO     -     °❀˖ ° °❀⋆.ೃ࿔*:･  ° ❀˖°
2025-11-29 10:30:15,132 - mitsuki - INFO     -
2025-11-29 10:30:15,132 - mitsuki - INFO     - :: Mitsuki ::                (v0.1.3)
2025-11-29 10:30:15,132 - mitsuki - INFO     -
2025-11-29 10:30:15,133 - _granian - INFO     - Starting granian...
2025-11-29 10:30:15,139 - _granian - INFO     - Listening at: http://127.0.0.1:8000

Step 5: See Your API in Action

Mitsuki automatically generates OpenAPI documentation for your API. No work needed.

Open your browser and navigate to http://localhost:8000/docs:

You'll find an interactive API documentation page where you can see all your endpoints and even test them directly from the browser.

Either through the terminal, or through the API UI, try creating a new post:

curl -X POST http://localhost:8000/api/post \
-H "Content-Type: application/json" \
-d '{}'

The API will respond with the newly created post:

{
  "id": "018ecb71-d556-7000-8000-000000000001",
  "created_at": "2025-11-29T10:35:00.123456",
  "updated_at": "2025-11-29T10:35:00.123456"
}

Your Journey Has Just Begun

In just a minute, you've bootstrapped a complete API with Mitsuki.

Not a "Hello World" app, you have a functional application starter with a database model, a REST API, and interactive documentation.

This is the benefit of convention over configuration. Mitsuki lets you focus on what truly matters: your business logic, and the value you want to deliver.

Next Steps

Ready to learn more?

Dive into the Official Documentation to explore custom queries, dependency injection, and more.
Add new fields to your Post entity in src/domain/post.py.
Implement unique business logic in your PostService at src/service/post_service.py.

Happy coding! ❀

Introducing Mitsuki - Spring Boot Enterprise Patterns and Speed with Python's Flexibility

David Landup — Sat, 29 Nov 2025 05:14:58 +0000

Introduction

Introducing Mitsuki: Spring Boot enterprise patterns meeting Python's flexibility and ecosystem! 🚀🐍

Get started in a single file with just a few lines, or have mitsuki init create your file structure, domain objects, and auto-implement repositories with CRUD operations. ✨

It's a structured way to build web applications/REST APIs, and benefits from assumptions on that structure. You already know the patterns - Mitsuki brings them to your fingertips. 💻

Performance-wise - it brings your Python applications into the range of Java/JS applications on consumer hardware. 📊

Check it out on GitHub :)

Bootstrapping Projects in a Minute - from Zero to Functional Starter

Okay, let's go beyond "Hello World" - how long does it take to go from zero to something more functional? Something with a database connection, a domain object, a repository with CRUD capabilities, service and controller?

About a single minute. ✨

Here's a live example, of starting a Mitsuki project, which includes:

Project setup
Domain object
Entity controller, service and repository with functional CRUD

Gifs don't work here :(
Here's a link.

Automatic OpenAPI Generation

If you like Swagger, Redoc or Scalar - Mitsuki auto-generates all of these for you. You can disable any one of them, have them all, and choose a default UI to show at /docs.

Not a single line is required to get bootstrapped, and you can customize them easily:

Supercharging Data Layer Repositories: Auto-CRUD and Query DSL

Basic CRUD functionality is oftentimes boilerplate, especially in the first couple of steps of development. Basic SQL queries are also highly predictable.

Mitsuki includes Query DSL - which lets you write function names, and infers the SQL from them. Of course, you can also write your own queries in SQL (native), SQL (SQLAlchemy ORM syntax) or SQLAlchemy itself.

Given an @Entity, any @CrudRepository aware of an entity will allow you to just... run CRUD operations on it. save() it, find_by_id(), delete() them - no implementation needed:

@Entity()
@dataclass
class Post:
    id: int = Id()
    title: str = ""
    author: str = ""
    views: int = 0

@CrudRepository(entity=Post)
class PostRepository:
    # Built-in methods (auto-implemented):
    # - find_by_id(id) -> Post | None
    # - find_all(page=0, size=10) -> List[Post]
    # - save(entity: Post) -> Post
    # - delete(entity: Post) -> None
    # - count() -> int
    # - exists_by_id(id) -> bool

Or, you want to find_by_field()? find_by_active_true()? Just define the method names - they get converted into SQL:

@CrudRepository(entity=Post)
class PostRepository:
    # Mitsuki turns these into SQL queries
    async def find_by_email(self, email: str): ...
    async def find_by_username(self, username: str): ...

find_by_email(email) → SELECT * WHERE email = ?
find_by_age_greater_than(age) → SELECT * WHERE age > ?
count_by_status(status) → SELECT COUNT(*) WHERE status = ?

Or just write @Query implementations yourself:

@CrudRepository(entity=Post)
class PostRepository:

    @Query("""
            SELECT u FROM User u
            WHERE u.active = :active
            ORDER BY u.created_at DESC
        """)
    async def find_active_users(self, active: bool, limit: int, offset: int): ...

Or use SQLALchemy:

@CrudRepository(entity=Post)
class PostRepository:
    # Or write custom SQLAlchemy
    async def find_popular_posts(self):
        async with self.get_connection() as conn:
            query = select(Post).where(Post.views > 1000)
            result = await conn.execute(query)
            return [dict(row._mapping) for row in result.fetchall()]

Why Another Web Dev Framework?

I worked in both research and enterprise and have built a lot of services in Python and Java, and they converge to using similar patterns, regardless of how different the architectures and domains are (web apps, ML research, distributed systems, etc.)

After writing a lot of Python, I was missing a framework that strongly supported some of these patterns formally, and find that this structure lets you make assumptions that can really boost dev experience on long term projects.

Microframeworks are great. They let you get started with a single file of a few lines, but (in my opinion) lack the structure you want on long term projects, working with teams, so you end up making that structure yourself anyway. In doing so, you get a small initial boost in productivity, but at the cost of your productivity in the future.

Mitsuki tries to both allow you to start quick and easy with a single file in a few lines, but also be more friendly to you and your team through time, by giving structure to your development process.

Thus, I made an early version of a framework heavily inspired by Spring Boot. The core idea is that you can do enterprise apps without the enterprise pain, in Python, with high performance.

Want a simple REST API? app.py with a few lines.
Want a decent starter with auto-implemented CRUD? mitsuki init to get a starter project with domain classes, services, controllers and repositories.
Performance? Similar to Express and Spring Boot (in Docker, on an M1 MacBook Pro, 8GB of RAM), out of the box, no configuration needed.

Lightweight

Despite the "fancy sounding" terminology, Mitsuki itself is very lightweight, and only adds a very small overhead (10%) over the components that power it (namely, Starlette and Granian). I don't want to commit to ASGI only, and a future version will likely rewrite this core logic to leverage granian further.

There is a lot of ground left to cover, lots of docs to write, examples to explore, features to expand. I'm also planning to write a few tools that leverage the structure of the framework to increase DX within enterprise teams.

But before any of that, I'm looking for feedback.
Yay or nay? :)

Benchmarks

P.S. On the topic of performance and benchmarks, there are a few remarks in the repo's /benchmarks directory. Yes, most benchmarks are arbitrary, heavily gameable, and your bottleneck is likely going to be your business logic, not the framework, anyways. Yes, Spring Boot and Elysia will likely have higher ceilings, so running on a stronger CPU will likely change the order of the benchmark. Yes, there's a million variables that affect these. Yes, granian is written in Rust, not Python.

The point of the benchmark is threefold:

This is the sort of experience you get out of the box, on your device and where you'll deploy it (Dockerized on small instance most likely)
Python web apps can stand shoulder to shoulder with JS/Java performance-wise
Despite the seeming complexity around dependency injection, state tracking, etc., Mitsuki is pretty lightweight.

DEV Community: David Landup

Guide to Database Transactions in Python with Mitsuki

Mitsuki

What are Transactions?

When are Transactions Useful?

Example Application

Making a Transfer Atomic with @Transactional

How Transactions Work in Mitsuki

Propagation: When Transactional Methods Call Each Other

REQUIRED: Join the Caller

REQUIRES_NEW: An Audit Log That Survives Rollbacks

NESTED: A Batch That Skips Failures

Rollback Rules

Isolation

Limitations

Next Steps

Mitsuki API Observability with Grafana and Prometheus

Mitsuki

Running the Stack

The Compose File

Configuring Prometheus

Generating Traffic

The Framework Dashboard

How the Panels Are Built

Scheduled Tasks

The Demo Dashboard

Using the Dashboard in Your Own Application

Cleaning Up

Next Steps

Introducing Automatic Python Web Application Instrumentation in Mitsuki

Mitsuki

Installing

Enabling It

What Gets Recorded

Human Readable and Prometheus Format

Securing the Endpoints

Limitations

Next Steps

How to Access the Request Object in Mitsuki Controllers

The Request Type Hint

Combining with Other Parameters

Next Steps

Guide to Path Variables and Query Parameters in Mitsuki Controllers

How to Handle URL Parameters in Mitsuki Controllers

Mitsuki

Introduction

Setup

Path Variables: Accessing Resource Identifiers

Query Parameters: Filtering and Sorting

Next Steps

Guide to Mitsuki Repositories: From Zero to Full CRUD

Introduction

Mitsuki

Domain Objects - @Entity

Level 1: Zero-Code CRUD with @CrudRepository

Level 2: Query DSL

Level 3: Taking Control with @Query

SQLAlchemy ORM-style Syntax (Default)

Native SQL Syntax

Level 4: Custom SQLAlchemy Methods

Conclusion

Next Steps

How to Create Your First Mitsuki REST API from Scratch

Mitsuki

Bootstrapping Mitsuki Projects

Step 1: Install Mitsuki

Step 2: mitsuki init 🪄

Step 3: Anatomy of a Mitsuki Project

What's Inside? Fully Functional API

Step 4: Liftoff! Running Your Server 🚀

Step 5: See Your API in Action

Your Journey Has Just Begun

Next Steps

Introducing Mitsuki - Spring Boot Enterprise Patterns and Speed with Python's Flexibility

Introduction

Bootstrapping Projects in a Minute - from Zero to Functional Starter

Automatic OpenAPI Generation

Supercharging Data Layer Repositories: Auto-CRUD and Query DSL

Why Another Web Dev Framework?

Lightweight

Making a Transfer Atomic with `@Transactional`

`REQUIRED`: Join the Caller

`REQUIRES_NEW`: An Audit Log That Survives Rollbacks

`NESTED`: A Batch That Skips Failures

The `Request` Type Hint

Domain Objects - `@Entity`

Level 1: Zero-Code CRUD with `@CrudRepository`

Level 3: Taking Control with `@Query`