Ormar

0.23.1 · active · verified Wed Apr 15

Ormar is an async ORM for Python, designed with FastAPI and Pydantic validation in mind, supporting Postgres, MySQL, and SQLite. It provides a single model definition that acts as both an ORM model and a Pydantic model. Currently at version 0.23.1, it maintains an active development and release cadence, frequently pushing updates including vulnerability fixes and new features.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to set up an Ormar model, connect to an in-memory SQLite database, create tables, and perform basic CRUD operations. It uses the modern `OrmarConfig` and `DatabaseConnection` patterns. Note that for persistent databases, `alembic` is recommended for migrations, and `create_all` is typically run only once during setup or testing.

import asyncio
import sqlalchemy
import ormar

# 1. Define Database Connection and Metadata
DATABASE_URL = "sqlite+aiosqlite:///test.db"

# This assumes a base config for all models. For complex apps, use `base_ormar_config.copy()`
base_ormar_config = ormar.OrmarConfig(
    metadata=sqlalchemy.MetaData(),
    database=ormar.DatabaseConnection(DATABASE_URL),
)

# 2. Define an Ormar Model
class User(ormar.Model):
    ormar_config = base_ormar_config.copy(tablename="users")

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    is_active: bool = ormar.Boolean(default=True)

async def main():
    # 3. Connect to the database and create tables
    if not base_ormar_config.database.is_connected:
        await base_ormar_config.database.connect()
    
    # Create tables (only once, usually in a migration or startup script)
    # For a persistent DB, use alembic. For quickstart, create all.
    print("Creating tables...")
    engine = sqlalchemy.create_engine(DATABASE_URL.replace('+aiosqlite', ''))
    base_ormar_config.metadata.create_all(engine)
    print("Tables created.")

    # 4. Create a new user
    print("Creating user Jane Doe...")
    jane = await User.objects.create(name="Jane Doe")
    print(f"Created user: {jane.id} - {jane.name} (active: {jane.is_active})")

    # 5. Retrieve all users
    print("Retrieving all users...")
    users = await User.objects.all()
    for user in users:
        print(f"Found user: {user.id} - {user.name} (active: {user.is_active})")

    # 6. Disconnect from the database
    print("Disconnecting from database...")
    if base_ormar_config.database.is_connected:
        await base_ormar_config.database.disconnect()
    print("Disconnected.")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →