kakaorm 0.4.3

Async-native ORM for PostgreSQL, SQLite, and MySQL/MariaDB

KakaORM

日本語

An async-native ORM for Python. Supports PostgreSQL (asyncpg / psycopg3), SQLite (aiosqlite), and MySQL/MariaDB (aiomysql) as backends, providing Django ORM-like model definitions and type-safe query building.

Features

• Fully async — async/await-based API that integrates naturally with asyncio
• Type-safe queries — Build queries without strings using operator overloading: User.age >= 20
• Multi-database — Supports PostgreSQL (asyncpg / psycopg3), SQLite (aiosqlite), and MySQL/MariaDB (aiomysql)
• Auto migrations — Detects diff between models and DB schema and generates ALTER TABLE; Migrator.run(models) applies pending changes in one call
• Generic descriptors — Type annotation inference via Column[T] for correct IDE completion
• Event hooks — Define before_insert / after_update etc. directly on your Model
• Relation definitions — Declare FK navigation (forward and reverse) with has_many() / has_one() / belongs_to()
• Pydantic v2 integration — Implements __get_pydantic_core_schema__ / __get_pydantic_json_schema__; use KakaORM models directly as FastAPI response_model
• Eager loading — Batch-fetch related models with prefetch() to eliminate N+1 queries
• Migration autogenerate — Auto-generate diff files with autogenerate(); manage with run_files() + downgrade()
• CTE (WITH clause) — Structure complex queries with with_cte(name, queryset)
• Deletion strategies — SoftDeleteModel (logical deletion) and ArchiveModel (archive deletion) base classes; switch delete() behavior simply by changing inheritance
• Validation — Attach validators to columns (min_length, max_length, min_value, max_value, regex, one_of); save() auto-validates and raises ValidationError before any DB write
• Upsert — get_or_create() and update_or_create() return (instance, created: bool)
• Bulk update — bulk_update(instances, fields=[...]) flushes many instances in a single executemany call

Installation

# SQLite (development / testing)
pip install "kakaorm[aiosqlite]"

# PostgreSQL (asyncpg)
pip install "kakaorm[asyncpg]"

# PostgreSQL (psycopg3)
pip install "kakaorm[psycopg3]"

# MySQL / MariaDB
pip install "kakaorm[aiomysql]"

# All drivers
pip install "kakaorm[all]"

Quickstart

import asyncio
import kakaorm
from kakaorm import Model, IntColumn, StrColumn, BoolColumn

class Task(Model):
    title = StrColumn(nullable=False)
    done  = BoolColumn(nullable=False, default=False)

    class Meta:
        table_name = "task"

async def main():
    engine = await kakaorm.connect("sqlite+aiosqlite:///:memory:")
    await engine.create_table(Task)

    task = await Task.create(title="Try KakaORM")
    print(task.id, task.title, task.done)  # 1 Try KakaORM False

    task.done = True
    await task.save()

    tasks = await Task.where(Task.done == True)
    print(tasks)  # [<Task id=1>]

    await engine.disconnect()

asyncio.run(main())

Model Definition

from kakaorm import Model, IntColumn, StrColumn, FloatColumn, BoolColumn, DateTimeColumn, ForeignKey

class Author(Model):
    name  = StrColumn(nullable=False)
    email = StrColumn(unique=True, nullable=False)
    bio   = StrColumn(nullable=True)

    class Meta:
        table_name = "author"

class Post(Model):
    title     = StrColumn(nullable=False)
    body      = StrColumn(nullable=True)
    published = BoolColumn(nullable=False, default=False)
    views     = IntColumn(nullable=False, default=0)
    author_id = ForeignKey(Author, nullable=True)

    class Meta:
        table_name = "post"

An id column is added automatically as the primary key.

Custom Primary Keys

Set primary_key=True on any column to make it the primary key. No auto-increment is applied.

class Country(Model):
    code = StrColumn(primary_key=True, nullable=False)  # e.g. "JP" / "US"
    name = StrColumn(nullable=False)

    class Meta:
        table_name = "country"

# Explicit primary key on INSERT
jp = await Country.create(code="JP", name="Japan")
jp.name = "Japan (updated)"
await jp.save()  # UPDATE WHERE code = 'JP'

Composite Indexes

Declare indexes in Meta.indexes as a list of tuples. CREATE INDEX is issued automatically when create_table() runs.

class Product(Model):
    name     = StrColumn(nullable=False)
    category = StrColumn(nullable=False)
    price    = IntColumn(nullable=False)

    class Meta:
        table_name = "product"
        indexes = [
            ("category", "price"),  # composite index
            ("name",),              # single-column index
        ]

Column Types

Class	Python type	SQL type
IntColumn	int	INTEGER
StrColumn	str	TEXT / VARCHAR(n)
FloatColumn	float	DOUBLE PRECISION
BoolColumn	bool	BOOLEAN
DateTimeColumn	datetime	TIMESTAMP WITH TIME ZONE
DateColumn	date	DATE
TimeColumn	time	TIME
DecimalColumn	Decimal	NUMERIC(p, s)
ForeignKey	int	INTEGER REFERENCES ...

For column options (nullable, default, unique, primary_key, index, check, auto_increment, auto_now, on_delete, etc.) see the full reference below.

→ Full API reference: docs/REFERENCE.md

CRUD

Create

author = await Author.create(name="Alice", email="[e-mail]")
print(author.id)  # DB-generated ID is set

Read

# All records
authors = await Author.all()

# Single record (raises NotFound if not found)
author = await Author.get(Author.email == "[e-mail]")

# Single record by primary key (returns None if not found)
author = await Author.find(1)

# Single record by condition (returns None if not found)
author = await Author.get_or_none(Author.id == 1)

# First / last
first = await Author.first()
last  = await Author.last()

# As a dict
author = await Author.get(Author.id == 1)
data = author.to_dict()         # {"id": 1, "name": "Alice", "email": "..."}

Update

author.name = "Alicia"
await author.save()

Delete

await author.delete()

Validation

from kakaorm.validators import min_length, max_length, min_value, regex

class User(Model):
    name  = StrColumn(nullable=False, validators=[min_length(2), max_length(50)])
    age   = IntColumn(nullable=True,  validators=[min_value(0)])
    email = StrColumn(nullable=False, validators=[
        regex(r"^[^@]+@[^@]+\.[^@]+$", message="Enter a valid email address.")
    ])

    class Meta:
        table_name = "user"

try:
    await User.create(name="A", age=-1, email="bad")
except ValidationError as e:
    print(e.errors)
    # {"name": ["..."], "age": ["..."], "email": ["..."]}

Upsert

# get_or_create — find or create; returns (instance, created: bool)
author, created = await Author.get_or_create(
    email="[e-mail]",
    defaults={"name": "Alice"},
)

# update_or_create — find and update, or create
post, created = await Post.update_or_create(
    slug="hello-world",
    defaults={"title": "Hello World", "published": True},
)

Bulk Operations

# Bulk INSERT (batched into minimal SQL statements)
posts = [Post(title=f"Post {i}", views=0) for i in range(1000)]
await Post.bulk_create(posts)

# Bulk UPDATE — flush many instances at once (uses executemany)
for post in posts:
    post.views = 0
await Post.bulk_update(posts, fields=["views"])

# QuerySet-level bulk UPDATE / DELETE
await Post.where(Post.published == False).update(published=True)
await Post.where(Post.views == 0).delete()

# TRUNCATE (also resets sequences)
await Post.truncate()

Database Connections

# SQLite (development / testing)
pip install fastapi uvicorn kakaorm aiosqlite

# PostgreSQL (asyncpg)
pip install fastapi uvicorn kakaorm asyncpg

# PostgreSQL (psycopg3)
pip install fastapi uvicorn kakaorm "psycopg[binary]" psycopg-pool

# MySQL / MariaDB
pip install fastapi uvicorn kakaorm aiomysql

DB	URL format
SQLite (file)	sqlite+aiosqlite:///./app.db
SQLite (in-memory)	sqlite+aiosqlite:///:memory:
PostgreSQL (asyncpg)	postgresql+asyncpg://user:password@localhost/dbname
PostgreSQL (psycopg3)	postgresql+psycopg3://user:password@localhost/dbname
MySQL / MariaDB	mysql+aiomysql://user:password@localhost:3306/dbname

# Always use await — omitting it raises a RuntimeWarning with a fix hint
engine = await kakaorm.connect("sqlite+aiosqlite:///:memory:")
engine = await kakaorm.connect("sqlite+aiosqlite:///./dev.db")

# PostgreSQL (asyncpg)
engine = await kakaorm.connect("postgresql+asyncpg://user:password@localhost/dbname")

# PostgreSQL (psycopg3)
engine = await kakaorm.connect("postgresql+psycopg3://user:password@localhost/dbname")

# MySQL / MariaDB (aiomysql)
engine = await kakaorm.connect("mysql+aiomysql://user:password@localhost:3306/dbname")

# Also usable as a context manager
async with await kakaorm.connect("sqlite+aiosqlite:///:memory:") as engine:
    ...

Migrations

from kakaorm.migration import Migrator

# Apply any pending schema changes in one call (no-op if schema is up to date)
await Migrator(engine).run([Author, Post])

# Or use the two-step API for more control
migrator = Migrator(engine)
plan = await migrator.plan([Author, Post])
if not plan.is_empty():
    await plan.apply()

Framework Integration

• FastAPI — See FastAPI Integration Guide / 日本語
• Flask — See Flask Integration Guide / 日本語

Project Structure

kakaorm/
├── .github/
│   └── workflows/
│       └── ci.yml           # GitHub Actions CI (lint + test matrix + MySQL + build)
├── docs/
│   ├── FASTAPI.md           # FastAPI integration guide (English)
│   ├── FASTAPI.ja.md        # FastAPI integration guide (Japanese)
│   ├── FLASK.md             # Flask integration guide (English)
│   ├── FLASK.ja.md          # Flask integration guide (Japanese)
│   ├── REFERENCE.md         # Full API reference (English)
│   └── REFERENCE.ja.md      # Full API reference (Japanese)
├── kakaorm/                 # Package source
│   ├── __init__.py          # Public API re-exports
│   ├── py.typed             # PEP 561 type marker
│   ├── engine.py            # Engine base class + AsyncpgEngine / AioSQLiteEngine / AioMySQLEngine / Psycopg3Engine, connect()
│   ├── model.py             # Model base class, AsyncORMMeta metaclass
│   ├── query.py             # QuerySet (lazy query builder)
│   ├── validators.py        # ValidationError + built-in validators (min_length, max_length, …)
│   ├── soft_delete.py       # SoftDeleteModel / SoftDeleteQuerySet (logical deletion)
│   ├── archive.py           # ArchiveModel / ArchiveQuerySet (archive deletion)
│   ├── relationship.py      # has_many / has_one / belongs_to descriptors
│   ├── columns/
│   │   ├── base.py          # Column[T] base class, ColumnMeta, WhereClause
│   │   └── types.py         # IntColumn, StrColumn, FloatColumn, BoolColumn,
│   │                        # DateTimeColumn, DateColumn, TimeColumn, DecimalColumn, ForeignKey
│   └── migration/
│       └── __init__.py      # Migrator, VersionedMigrator, MigrationPlan
├── examples/
│   ├── blog_example.py      # Blog system usage example
│   ├── fastapi_todo.py      # FastAPI TODO list API
│   └── flask_todo.py        # Flask TODO list API
│
│   Larger sample apps live in a separate repository:
│   → https://github.com/AyumuTakai/kakaorm-samples
├── tests/
│   ├── conftest.py
│   ├── test_crud.py
│   ├── test_joins.py
│   ├── test_aggregates.py
│   ├── test_transaction.py
│   ├── test_bulk_create.py
│   ├── test_bulk_update.py  # bulk_update()
│   ├── test_validation.py   # Column validators + ValidationError
│   ├── test_upsert.py       # get_or_create / update_or_create
│   ├── test_raw_sql.py
│   ├── test_migration.py
│   ├── test_indexes.py      # Composite indexes
│   ├── test_custom_pk.py    # Custom primary keys
│   ├── test_hooks.py        # Event hooks
│   ├── test_relationship.py # Relation definitions
│   └── test_security.py     # Security regression tests
├── CHANGELOG.md             # Version history
├── LICENSE                  # MIT License
├── pyproject.toml           # Package metadata and build configuration
└── ruff.toml                # Ruff configuration

Running Tests

pip install -e ".[aiosqlite,dev]"
pytest

# MySQL tests (requires a running MySQL server)
# MySQL 8.0 uses caching_sha2_password auth, which requires the cryptography package
pip install -e ".[aiomysql,dev]" cryptography
export KAKAORM_MYSQL_URL="mysql+aiomysql://root:password@localhost:3306/test_db"
pytest tests/test_mysql.py

Requirements

• Python 3.11+
• The appropriate driver for your database (aiosqlite / asyncpg / psycopg[binary] / aiomysql)
• For Pydantic v2 integration: pip install pydantic (optional — the ORM core works without it)

License

MIT License