Tables and Columns
SQLAlchemy’s schema definition system provides programmatic ways to define database tables, columns, and metadata. All schema constructs are in sqlalchemy.schema.
MetaData is a container object that holds table definitions and schema information.
from sqlalchemy import MetaData
# Create metadata container
metadata = MetaData()
# With schema name
metadata = MetaData( schema = 'myschema' )
Signature: MetaData(schema: Optional[str] = None, naming_convention: Optional[Dict] = None)Container for database schema elements like tables, columns, and constraints.
# Create all tables
metadata.create_all(engine)
# Drop all tables
metadata.drop_all(engine)
# Create specific tables
metadata.create_all(engine, tables = [users, orders])
# Check if table exists
metadata.create_all(engine, checkfirst = True )
# Reflect existing database
metadata.reflect( bind = engine)
# Access reflected tables
users_table = metadata.tables[ 'users' ]
Table Definition
Basic 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 ),
Column( 'age' , Integer)
)
Signature: Table(name: str, metadata: MetaData, *columns: Column, schema: Optional[str] = None, **kwargs)Represents a database table with columns and constraints.
Table with Schema
# Explicit schema
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'name' , String( 50 )),
schema = 'public'
)
# Use metadata default schema
metadata = MetaData( schema = 'myapp' )
users = Table( 'users' , metadata, Column( 'id' , Integer)) # myapp.users
Table Options
Common Options
Database-Specific
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'name' , String( 50 )),
# Table options
schema = 'public' ,
comment = 'User accounts table' ,
info = { 'version' : '1.0' },
# Existence checks
keep_existing = True , # Don't override if exists
extend_existing = False , # Add to existing definition
autoload_with = engine, # Reflect from database
)
# MySQL options
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
mysql_engine = 'InnoDB' ,
mysql_charset = 'utf8mb4' ,
mysql_collate = 'utf8mb4_unicode_ci' ,
mysql_comment = 'User table'
)
# PostgreSQL options
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
postgresql_tablespace = 'fast_storage' ,
postgresql_partition_by = 'RANGE (created_at)'
)
# SQLite options
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
sqlite_autoincrement = True
)
Column Definition
Basic Columns
from sqlalchemy import Column, Integer, String, Boolean, DateTime
users = Table(
'users' ,
metadata,
Column( 'id' , Integer),
Column( 'username' , String( 50 )),
Column( 'email' , String( 100 )),
Column( 'is_active' , Boolean),
Column( 'created_at' , DateTime)
)
Signature: Column(name: Optional[str], type_: TypeEngine, *constraints, **kwargs)Represents a table column with type and constraints.
Column Constraints
from sqlalchemy import Column, Integer, String, Boolean
users = Table(
'users' ,
metadata,
# Primary key
Column( 'id' , Integer, primary_key = True ),
# Not null
Column( 'username' , String( 50 ), nullable = False ),
# Unique
Column( 'email' , String( 100 ), unique = True ),
# Default value
Column( 'is_active' , Boolean, default = True ),
# Server default (SQL expression)
Column( 'created_at' , DateTime, server_default = func.now()),
# Auto-increment
Column( 'id' , Integer, autoincrement = True ),
# Column comment
Column( 'notes' , String( 500 ), comment = 'User notes' ),
)
Column Options
Nullability
Default Values
Update Defaults
Indexes
# Nullable (default for most columns)
Column( 'middle_name' , String( 50 ), nullable = True )
# Not nullable
Column( 'email' , String( 100 ), nullable = False )
# Primary keys are automatically NOT NULL
Column( 'id' , Integer, primary_key = True ) # Implicitly NOT NULL
Column Names and Keys
# Column with different SQL name
Column( 'user_name' , String( 50 ), key = 'username' )
# Access as: users.c.username
# SQL generates: user_name
# Quoted names (preserve case, special chars)
from sqlalchemy import quoted_name
Column(quoted_name( 'User Name' , quote = True ), String( 50 ))
# Generates: "User Name" (quoted)
Accessing Columns
# Access table columns
users.c.id
users.c.name
users.columns.email
# Iterate columns
for col in users.c:
print (col.name, col.type)
# Get column by name
id_col = users.c[ 'id' ]
# Check if column exists
if 'email' in users.c:
print (users.c.email)
# All column names
column_names = users.c.keys()
Composite Primary Keys
# Multiple column primary key
order_items = Table(
'order_items' ,
metadata,
Column( 'order_id' , Integer, primary_key = True ),
Column( 'product_id' , Integer, primary_key = True ),
Column( 'quantity' , Integer),
Column( 'price' , Numeric( 10 , 2 ))
)
# Using PrimaryKeyConstraint
from sqlalchemy import PrimaryKeyConstraint
order_items = Table(
'order_items' ,
metadata,
Column( 'order_id' , Integer),
Column( 'product_id' , Integer),
Column( 'quantity' , Integer),
PrimaryKeyConstraint( 'order_id' , 'product_id' , name = 'pk_order_items' )
)
Generated Columns
Computed Columns
from sqlalchemy import Computed
# PostgreSQL, MySQL 5.7+, SQLite 3.31+
products = Table(
'products' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'price' , Numeric( 10 , 2 )),
Column( 'tax_rate' , Numeric( 4 , 3 )),
Column(
'price_with_tax' ,
Numeric( 10 , 2 ),
Computed( 'price * (1 + tax_rate)' )
)
)
# Stored vs. virtual
Column( 'total' , Numeric, Computed( 'price * quantity' , persisted = True )) # Stored
Column( 'total' , Numeric, Computed( 'price * quantity' , persisted = False )) # Virtual
Identity Columns
from sqlalchemy import Identity
# PostgreSQL IDENTITY, SQL Server IDENTITY
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, Identity( start = 1 , increment = 1 ), primary_key = True ),
Column( 'name' , String( 50 ))
)
# Custom options
Column(
'id' ,
Integer,
Identity(
start = 100 ,
increment = 1 ,
minvalue = 1 ,
maxvalue = 999999 ,
cycle = False ,
cache = 20
),
primary_key = True
)
Table Reflection
Reflect Existing Tables
# Reflect single table
users = Table( 'users' , metadata, autoload_with = engine)
# Access reflected columns
print (users.c.keys())
# Reflect all tables
metadata.reflect( bind = engine)
# Access reflected tables
for table_name in metadata.tables:
table = metadata.tables[table_name]
print ( f "Table: { table_name } " )
for col in table.c:
print ( f " Column: { col.name } ( { col.type } )" )
# Reflect specific schema
metadata.reflect( bind = engine, schema = 'myschema' )
# Reflect with views
metadata.reflect( bind = engine, views = True )
Selective Reflection
# Reflect only specific tables
metadata.reflect( bind = engine, only = [ 'users' , 'orders' ])
# Extend existing table definition
users = Table(
'users' ,
metadata,
Column( 'custom_field' , String( 100 )), # Add custom column
extend_existing = True ,
autoload_with = engine
)
Declarative Table Copies
Cloning Tables
# Copy to different metadata
new_metadata = MetaData()
users_copy = users.to_metadata(new_metadata)
# Copy with new name
users_archive = users.to_metadata(metadata, name = 'users_archive' )
# Copy with schema change
users_test = users.to_metadata(metadata, schema = 'test' )
# Add custom info
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
info = { 'version' : '2.0' , 'owner' : 'admin' }
)
# Access info
print (users.info[ 'version' ])
# Column info
Column( 'data' , String( 100 ), info = { 'encrypted' : True , 'pii' : True })
Common Patterns
Timestamp Columns
from sqlalchemy import func
from datetime import datetime
# Standard timestamp pattern
Table(
'posts' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'title' , String( 200 )),
Column( 'created_at' , DateTime, nullable = False , server_default = func.now()),
Column( 'updated_at' , DateTime, onupdate = func.now())
)
Soft Delete
Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'name' , String( 50 )),
Column( 'deleted_at' , DateTime, nullable = True ),
Column( 'is_deleted' , Boolean, default = False , server_default = 'false' )
)
Versioning
Table(
'documents' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'content' , Text),
Column( 'version' , Integer, default = 1 , nullable = False ),
Column( 'updated_at' , DateTime, onupdate = func.now())
)
Multi-tenant
Table(
'data' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'tenant_id' , Integer, nullable = False , index = True ),
Column( 'value' , String( 100 ))
)
Inspecting Schema
from sqlalchemy import inspect
inspector = inspect(engine)
# Get table names
table_names = inspector.get_table_names()
view_names = inspector.get_view_names()
# Get columns
columns = inspector.get_columns( 'users' )
for col in columns:
print ( f " { col[ 'name' ] } : { col[ 'type' ] } " )
# Get primary keys
pk = inspector.get_pk_constraint( 'users' )
print (pk[ 'constrained_columns' ])
# Get foreign keys
fks = inspector.get_foreign_keys( 'orders' )
# Get indexes
indexes = inspector.get_indexes( 'users' )
Next Steps
Constraints Learn about foreign keys and constraints
DDL Operations Create and modify database schemas