Use SQLAlchemy's Column
with sa_column
¶
Sometimes you need full control over how a database column is defined — beyond what Field()
options provide.
SQLModel lets you pass a fully configured SQLAlchemy Column(...)
using the sa_column
parameter.
This allows you to use advanced SQLAlchemy features and third‑party column types directly while keeping the simplicity of SQLModel models.
Info
sa_column
provides a low-level hook to supply a complete SQLAlchemy Column(...)
object for a field. SQLModel will use the column's type, options, and constraints as-is.
What sa_column
enables¶
- Fine‑grained control over column definitions (e.g.
ForeignKey
,CheckConstraint
,UniqueConstraint
,Index
,server_default
,server_onupdate
). - Custom/third‑party SQLAlchemy types (for example, encrypted strings, PostgreSQL
JSONB
, etc.). - Easier migration from or integration with existing SQLAlchemy models.
Use case: encrypted field with a custom type¶
Use a third‑party SQLAlchemy type from sqlalchemy-utils
to encrypt a string field. The key idea is that the field uses a full SQLAlchemy Column(...)
via sa_column
.
import os
from typing import Optional
from sqlalchemy import Column
from sqlalchemy_utils.types.encrypted.encrypted_type import (
AesEngine,
StringEncryptedType,
)
from sqlmodel import Field, Session, SQLModel, create_engine, select
# In a real application, load this from a secure source (e.g., environment variable or secrets manager)
ENCRYPTION_KEY = os.getenv("SQLMODEL_ENCRYPTION_KEY", "a-super-secret-key")
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
# Because the secret name should stay a secret
secret_name: str = Field(
sa_column=Column(
StringEncryptedType(
key=ENCRYPTION_KEY,
engine=AesEngine,
padding="pkcs5",
)
)
)
age: Optional[int] = None
sqlite_file_name = "database_encrypted.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url)
def create_db_and_tables() -> None:
# Reset DB for demo so decryption key changes don't break runs
if os.path.exists(sqlite_file_name):
os.remove(sqlite_file_name)
SQLModel.metadata.create_all(engine)
def create_heroes() -> None:
hero_1 = Hero(name="Ted Lasso", secret_name="Coach")
hero_2 = Hero(name="Roy Kent", secret_name="Roy")
hero_3 = Hero(name="Keeley Jones", secret_name="Keeley", age=29)
with Session(engine) as session:
print("Adding Hero 1: Ted Lasso")
print("Adding Hero 2: Roy Kent")
print("Adding Hero 3: Keeley Jones")
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
print("Inserted 3 heroes.\n")
def select_heroes() -> None:
with Session(engine) as session:
print("Selecting by name: Ted Lasso")
statement = select(Hero).where(Hero.name == "Ted Lasso")
hero_1 = session.exec(statement).one()
print("Hero 1:", hero_1)
print("Hero 1 secret_name (decrypted in Python):", hero_1.secret_name)
# Read the raw encrypted value directly from the DB (bypassing type decryption)
with engine.connect() as conn:
raw_encrypted = conn.exec_driver_sql(
"SELECT secret_name FROM hero WHERE name = ?",
("Ted Lasso",),
).scalar_one()
print("Hero 1 secret_name (stored in DB, encrypted):", raw_encrypted)
print("\nSelecting by name: Roy Kent")
statement = select(Hero).where(Hero.name == "Roy Kent")
hero_2 = session.exec(statement).one()
print("Hero 2:", hero_2)
print("Hero 2 secret_name (decrypted in Python):", hero_2.secret_name)
def main() -> None:
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
Key points¶
- The field uses
sa_column=Column(StringEncryptedType(...))
, which gives full control over the SQLAlchemy column while keeping a SQLModel model. EncryptedType
is deprecated; the example usesStringEncryptedType
instead.- The type is initialized with keyword args (
key=...
,engine=...
,padding=...
) to match the installed package signature and avoid runtime errors. - The key is read from an environment variable. Don’t hard‑code secrets; use a secrets manager or environment variables, and ensure the same key is available for decryption.
- In the DB, the value is stored encrypted (you’ll see ciphertext in the SQL echo and database); in Python it’s transparently decrypted when you access the field.
- Indexing or filtering on encrypted ciphertext is typically not useful; design queries accordingly.
Run it¶
To try the encrypted type example locally:
python -m venv .venv
source .venv/bin/activate
pip install sqlmodel sqlalchemy-utils cryptography
export SQLMODEL_ENCRYPTION_KEY="change-me"
# Copy the code from docs_src/advanced/sa_column/tutorial001.py into app.py
python app.py
After running, you should see "Database and tables created." and a database_encrypted.db
SQLite file created in your working directory.
Tip
If you change the encryption key between runs, delete database_encrypted.db
first so existing ciphertext doesn’t fail to decrypt with the new key.
Output¶
Adding Hero 1: Ted Lasso
Adding Hero 2: Roy Kent
Adding Hero 3: Keeley Jones
Inserted 3 heroes.
Selecting by name: Ted Lasso
Hero 1: id=1 name='Ted Lasso' secret_name='Coach' age=None
Hero 1 secret_name (decrypted in Python): Coach
Hero 1 secret_name (stored in DB, encrypted): omSF3WBuflYmqx2+Dz6PgQ==
Selecting by name: Roy Kent
Hero 2: id=2 name='Roy Kent' secret_name='Roy' age=None
Hero 2 secret_name (decrypted in Python): Roy
Use case: enforcing uniqueness¶
- Single‑column unique: You can express this using
Field(unique=True)
in SQLModel or directly on the SQLAlchemyColumn(...)
when usingsa_column
for full control (e.g., to set a specific SQL type or name). - Composite unique (multiple columns): Prefer the idiomatic SQLAlchemy approach with
__table_args__
andUniqueConstraint
.
import os
from typing import Optional
from sqlalchemy import Column, String, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
# Single-column unique via sa_column
email: str = Field(sa_column=Column(String(255), unique=True, nullable=False))
name: str
secret_name: str
age: Optional[int] = None
# Composite unique constraint
__table_args__ = (
UniqueConstraint("name", "secret_name", name="uq_hero_name_secret"),
)
sqlite_file_name = "database_unique.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url)
def create_db_and_tables() -> None:
# Reset DB for demo
if os.path.exists(sqlite_file_name):
os.remove(sqlite_file_name)
SQLModel.metadata.create_all(engine)
def create_heroes() -> None:
with Session(engine) as session:
hero_1 = Hero(email="ted@richmond.afc", name="Ted Lasso", secret_name="Coach")
hero_2 = Hero(email="roy@richmond.afc", name="Roy Kent", secret_name="Roy")
hero_3 = Hero(
email="keeley@richmond.afc", name="Keeley Jones", secret_name="Keeley"
)
print("Adding Hero 1: Ted Lasso (email=ted@richmond.afc)")
print("Adding Hero 2: Roy Kent (email=roy@richmond.afc)")
print("Adding Hero 3: Keeley Jones (email=keeley@richmond.afc)")
session.add_all([hero_1, hero_2, hero_3])
session.commit()
print("Inserted 3 heroes.\n")
# Duplicate (name, secret_name) should fail (different email)
hero_4 = Hero(email="roy2@richmond.afc", name="Roy Kent", secret_name="Roy")
try:
print("Attempting to insert a duplicate (name, secret_name) ...")
session.add(hero_4)
session.commit()
except IntegrityError as e:
session.rollback()
print("Composite unique constraint enforced:", str(e.orig))
def select_heroes() -> None:
with Session(engine) as session:
print("\nSelecting by email (unique column):")
statement = select(Hero).where(Hero.email == "ted@richmond.afc")
hero_1 = session.exec(statement).one()
print("Hero 1:", hero_1)
print("\nSelecting by composite key (name, secret_name):")
statement = select(Hero).where(
(Hero.name == "Roy Kent") & (Hero.secret_name == "Roy")
)
hero_2 = session.exec(statement).one()
print("Hero 2:", hero_2)
def main() -> None:
create_db_and_tables()
create_heroes()
select_heroes()
if __name__ == "__main__":
main()
Key points¶
- Single‑column unique can be declared with
Field(unique=True)
(simple case) or on the SQLAlchemyColumn(..., unique=True)
viasa_column
when you need full control over type/nullable/name.Field(unique=True)
is shorthand for settingunique=True
on the underlying SQLAlchemy column. - Composite unique constraints across multiple columns use
__table_args__ = (UniqueConstraint(...),)
. Naming the constraint helps during migrations and debugging. - Nullability matters: a unique, nullable column can usually store multiple NULLs (DB‑specific). Set
nullable=False
for strict uniqueness. - The example uses a separate DB file (
database_unique.db
) to avoid colliding with other tutorials. - Attempting to insert a duplicate
email
or the same(name, secret_name)
pair will raise an integrity error.
Run it¶
To try the unique constraints example locally on macOS with bash:
python -m venv .venv
source .venv/bin/activate
pip install sqlmodel
# Copy the code from docs_src/advanced/sa_column/tutorial002.py into app.py
python app.py
After running, you should see the selected rows printed, with a database created at database_unique.db
. Attempting to insert a duplicate email
(single‑column unique) or a duplicate pair of (name, secret_name)
(composite unique) would raise an integrity error.
Output¶
Adding Hero 1: Ted Lasso (email=ted@richmond.afc)
Adding Hero 2: Roy Kent (email=roy@richmond.afc)
Adding Hero 3: Keeley Jones (email=keeley@richmond.afc)
Inserted 3 heroes.
Attempting to insert a duplicate (name, secret_name) ...
Composite unique constraint enforced: UNIQUE constraint failed: hero.name, hero.secret_name
Selecting by email (unique column):
Hero 1: name='Ted Lasso' id=1 age=None secret_name='Coach' email='ted@richmond.afc'
Selecting by composite key (name, secret_name):
Hero 2: name='Roy Kent' id=2 age=None secret_name='Roy' email='roy@richmond.afc'
Important considerations¶
- Prefer built‑in
Field()
parameters (likeunique=True
,index=True
,default=...
) when they are sufficient. - Use
sa_column
only when you need full SQLAlchemy control over the column. - Avoid conflicts between
sa_column
and otherField()
arguments that also affect the underlying column. - Match your backend: ensure the SQLAlchemy
Column(...)
you pass is compatible with your target database. - PostgreSQL: import and use types like
JSONB
,ARRAY
, orUUID
fromsqlalchemy.dialects.postgresql
when appropriate.
See also¶
- SQLAlchemy Column docs:
Column
- Advanced SQLModel topics: Advanced User Guide