Schema API

SQLAlchemy schema definition API reference

Complete reference for SQLAlchemy’s schema definition APIs.

MetaData

Container for database schema information.

from sqlalchemy import MetaData

metadata = MetaData()

schema · str

Default schema name for all tables

naming_convention · dict

Naming convention for constraints and indexes

Methods

create_all(bind) - Create all tables

metadata.create_all(engine)

drop_all(bind) - Drop all tables

metadata.drop_all(engine)

reflect(bind) - Reflect tables from database

metadata.reflect(engine)

Table

Define a database table.

from sqlalchemy import Table, Column, Integer, String, MetaData

metadata = MetaData()

users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50), nullable=False),
    Column("email", String(100), unique=True),
)

name · str · required

Table name

metadata · MetaData · required

MetaData instance

*args · Column | Constraint

Columns and constraints

schema · str

Schema name

Properties

c or columns - Column collection

users.c.name
users.columns.email

Column

Define a table column.

from sqlalchemy import Column, Integer, String, ForeignKey

Column("id", Integer, primary_key=True)
Column("name", String(50), nullable=False)
Column("email", String(100), unique=True, index=True)
Column("user_id", Integer, ForeignKey("users.id"))

name · str

Column name (can be inferred in ORM)

type_ · TypeEngine · required

Column data type

primary_key · bool · default: "False"

Part of primary key

nullable · bool · default: "True"

Allow NULL values

unique · bool · default: "False"

Add unique constraint

index · bool · default: "False"

Create index on column

default · Any

Python-side default value

server_default · str | FetchedValue

Server-side default

onupdate · Any

Python-side update value

autoincrement · bool | str · default: "auto"

Auto-increment behavior

Constraints

PrimaryKeyConstraint

from sqlalchemy import PrimaryKeyConstraint

Table(
    "users",
    metadata,
    Column("id", Integer),
    Column("tenant_id", Integer),
    PrimaryKeyConstraint("id", "tenant_id", name="pk_users")
)

ForeignKeyConstraint

from sqlalchemy import ForeignKeyConstraint

Table(
    "addresses",
    metadata,
    Column("user_id", Integer),
    Column("tenant_id", Integer),
    ForeignKeyConstraint(
        ["user_id", "tenant_id"],
        ["users.id", "users.tenant_id"],
        name="fk_addresses_users"
    )
)

UniqueConstraint

from sqlalchemy import UniqueConstraint

Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("email", String(100)),
    UniqueConstraint("email", name="uq_users_email")
)

CheckConstraint

from sqlalchemy import CheckConstraint

Table(
    "products",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("price", Numeric),
    CheckConstraint("price > 0", name="ck_products_price")
)

Indexes

Index

from sqlalchemy import Index

# Simple index
Index("idx_users_email", users.c.email)

# Composite index
Index("idx_users_name_email", users.c.name, users.c.email)

# Unique index
Index("idx_users_email_unique", users.c.email, unique=True)

# Partial index (PostgreSQL)
Index(
    "idx_active_users",
    users.c.email,
    postgresql_where=users.c.active == True
)

ForeignKey

Define a foreign key reference.

from sqlalchemy import ForeignKey

Column("user_id", Integer, ForeignKey("users.id"))
Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"))

column · str · required

Referenced column (table.column)

ondelete · str

ON DELETE action: CASCADE, SET NULL, RESTRICT

onupdate · str

ON UPDATE action: CASCADE, SET NULL, RESTRICT

Default Values

Python-side defaults

import datetime

Column("created_at", DateTime, default=datetime.datetime.utcnow)
Column("status", String, default="active")

Server-side defaults

from sqlalchemy import func

Column("created_at", DateTime, server_default=func.now())
Column("id", Integer, server_default=text("nextval('user_id_seq')"))

Computed columns

from sqlalchemy import Computed

Column("full_name", String, Computed("first_name || ' ' || last_name"))

Identity columns

from sqlalchemy import Identity

Column("id", Integer, Identity(start=1000, increment=1))

Sequences

from sqlalchemy import Sequence

user_id_seq = Sequence("user_id_seq", start=1000)
Column("id", Integer, user_id_seq, server_default=user_id_seq.next_value())

DDL

DDL Events

from sqlalchemy import DDL, event

event.listen(
    users,
    "after_create",
    DDL("ALTER TABLE users ADD CONSTRAINT ck_email CHECK (email LIKE '%@%')")
)

CreateTable / DropTable

from sqlalchemy.schema import CreateTable, DropTable

# Get DDL as string
print(CreateTable(users).compile(engine))

# Execute DDL
with engine.begin() as conn:
    conn.execute(CreateTable(users))
    conn.execute(DropTable(users))