masonite-orm-async 1.2.0

Async ORM for Python — fully async from connection to migration

async-orm

An async fork of Masonite ORM — the same familiar API, fully async from connection to migration.

Fork notice: This project is based on masonite-orm v2.24.0 by Joe Mancuso, licensed under MIT. All connection, query, model, relationship, scope, schema, and migration layers have been converted to use async/await.

What changed from Masonite ORM

Layer	Original	This fork
Connections	Synchronous drivers	aiosqlite, aiomysql, asyncpg, aioodbc
QueryBuilder	.get(), .first(), .count() are sync	Terminal methods return coroutines (await)
Models	User.all() returns a collection	await User.all() returns a collection
Relationships	Sync eager loading	await User.with_("posts").find(1)
Schema/Blueprint	with schema.create(...) as table:	async with schema.create(...) as table:
Migrations	def up(self):	async def up(self):
CLI commands	Direct calls	asyncio.run() bridge

Chain methods (where, order_by, select, limit, etc.) remain synchronous — only terminal methods that hit the database are async.

Installation

pip install masonite-orm-async

With optional Pydantic support:

pip install masonite-orm-async[pydantic]

Quick start

Models

from masoniteorm.models import Model
from masoniteorm.relationships import has_many

class User(Model):
    __connection__ = "sqlite"

    id: int
    name: str
    email: str

    @has_many("id", "user_id")
    def posts(self):
        return Post

class Post(Model):
    __connection__ = "sqlite"

    id: int
    user_id: int
    title: str
    body: str

Field annotations are optional but give you IDE autocomplete on user.name, user.email, etc. They are bare annotations with no default values — they don't affect runtime behavior.

Queries

# All terminal operations are awaited
users = await User.all()
user = await User.find(1)
user = await User.where("email", "[e-mail]").first()

# Chain methods are still synchronous
await User.where("active", True).order_by("name").limit(10).get()

# CRUD
user = await User.create({"name": "Alice", "email": "[e-mail]"})
user.name = "Alice Updated"
await user.save()
await user.delete()

# Eager loading
user = await User.with_("posts").find(1)

Migrations

from masoniteorm.migrations import Migration

class CreateUsersTable(Migration):
    async def up(self):
        async with self.schema.create("users") as table:
            table.increments("id")
            table.string("name")
            table.string("email").unique()
            table.timestamps()

    async def down(self):
        await self.schema.drop("users")

Running migrations programmatically

import asyncio
from masoniteorm.migrations import Migration

async def main():
    migration = Migration(
        connection="sqlite",
        migration_directory="databases/migrations",
        config_path="config",
    )
    await migration.create_table_if_not_exists()
    await migration.migrate()

asyncio.run(main())

FastAPI integration

from contextlib import asynccontextmanager
from fastapi import FastAPI
from masoniteorm.migrations import Migration

@asynccontextmanager
async def lifespan(app: FastAPI):
    migration = Migration(
        connection="sqlite",
        migration_directory="databases/migrations",
        config_path="config",
    )
    await migration.create_table_if_not_exists()
    await migration.migrate()
    yield
    from config import DB
    await DB.close_all()

app = FastAPI(lifespan=lifespan)

@app.get("/users")
async def list_users():
    users = await User.all()
    return [u.model_dump() for u in users]

Sanic integration

from sanic import Sanic, json
from masoniteorm.migrations import Migration

app = Sanic("MyApp")

@app.before_server_start
async def setup(app, loop):
    migration = Migration(
        connection="sqlite",
        migration_directory="databases/migrations",
        config_path="config",
    )
    await migration.create_table_if_not_exists()
    await migration.migrate()

@app.after_server_stop
async def teardown(app, loop):
    from config import DB
    await DB.close_all()

@app.get("/users")
async def list_users(request):
    users = await User.all()
    return json([u.serialize() for u in users])

Type hints

All generic classes are subscriptable at runtime and have .pyi stubs for full IDE/type-checker support:

from masoniteorm.collection import Collection
from masoniteorm.query import QueryBuilder

# IDE knows: users is Collection[User], user is User | None
users = await User.all()       # → Collection[User]
user = await User.find(1)      # → User | None

# Chain methods preserve the type
query = User.where("active", True).order_by("name").limit(10)  # → QueryBuilder[User]
results = await query.get()    # → Collection[User]

Pydantic integration (optional)

Auto-generate Pydantic schemas from your models:

# Get the Pydantic model class
UserSchema = User.to_schema()  # → UserSchema(id: int | None, name: str | None, email: str | None)

# Serialize a model instance through Pydantic
user = await User.find(1)
data = user.model_dump()       # → {"id": 1, "name": "Alice", "email": "[e-mail]"}

# Use the schema directly for validation
validated = UserSchema(name="Bob", email="[e-mail]")

If pydantic is not installed, model_dump() falls back to serialize().

Configuration

Create a config.py (or config/database.py) that exposes a DB object:

from masoniteorm.connections import ConnectionResolver

DATABASES = {
    "default": "sqlite",
    "sqlite": {
        "driver": "sqlite",
        "database": "app.db",
    },
}

DB = ConnectionResolver().set_connection_details(DATABASES)

Set the config path via environment variable or pass it directly:

import os
os.environ["DB_CONFIG_PATH"] = "config"

Supported databases

Database	Async driver	Connection key
SQLite	aiosqlite	sqlite
MySQL	aiomysql	mysql
PostgreSQL	asyncpg	postgres
MSSQL	aioodbc	mssql

Transactions

from config import DB

# Context manager — auto-rollback on exception
async with DB.transaction():
    user = await User.create({"name": "Alice", "email": "[e-mail]"})
    await Post.create({"title": "Hello", "user_id": user.id})

# Manual control
await DB.begin_transaction()
await User.create({"name": "Bob", "email": "[e-mail]"})
await DB.commit()  # or DB.rollback()

Connection resilience

Pool creation retries automatically with exponential backoff (configurable):

DATABASES = {
    "default": "postgres",
    "postgres": {
        "driver": "postgres",
        "host": "127.0.0.1",
        "database": "mydb",
        "user": "myuser",
        "password": "mypass",
        "port": 5432,
        "connection_retries": 3,           # retry pool creation (default: 3)
        "connection_pool_timeout": 30,     # seconds to wait for a connection (default: 30)
        "connection_pooling_min_size": 1,
        "connection_pooling_max_size": 10,
    },
}

When the pool is exhausted, a PoolExhaustedError is raised instead of hanging:

from masoniteorm.exceptions import PoolExhaustedError

Testing

# ORM unit tests (SQLite, no dependencies)
python orm/test_async.py

# FastAPI demo tests
cd fastapi_demo && PYTHONPATH="../orm/src:." python -m pytest test_app.py -v

# Integration tests (requires Docker)
docker compose up -d --wait
PYTHONPATH="orm/src:." python -m pytest tests/ -v

# Load test
PYTHONPATH="orm/src:." python tests/load_test.py --db postgres --ops 1000 --concurrency 50

License

MIT — same as the original Masonite ORM. See LICENSE.

Credits

• Masonite ORM by Joe Mancuso and contributors — the original synchronous ORM this project is forked from.