-
-
Notifications
You must be signed in to change notification settings - Fork 773
📝 Add tutorial and advanced page for EncryptedType
#1574
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from 3 commits
00b594f
854314c
04a1ebf
cb84d05
692ae8f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,141 @@ | ||
# Use SQLAlchemy columns 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`. | ||
|
||
/// warning | Deprecation | ||
|
||
`EncryptedType` is deprecated in SQLAlchemy‑Utils since version `0.36.6`. Use `StringEncryptedType` instead. | ||
|
||
<a href="https://sqlalchemy-utils.readthedocs.io/en/latest/data_types.html#encryptedtype" class="external-link" target="_blank">See the upstream deprecation note</a>. | ||
|
||
/// | ||
|
||
Note: `StringEncryptedType` provides explicit string type handling and better compatibility with SQLAlchemy 2.x. | ||
ravishan16 marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
|
||
{* ./docs_src/advanced/sa_column/tutorial001.py *} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It would be nice to explain this in a bit more details, make it a bit more entertaining and step-by-step.
An then add a part of code example (only related lines with most important lines highlighted) like this
Then describe in short what is happening in |
||
|
||
### 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 uses `StringEncryptedType` 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: | ||
|
||
```bash | ||
python -m venv .venv | ||
source .venv/bin/activate | ||
pip install sqlmodel sqlalchemy-utils cryptography | ||
Comment on lines
+39
to
+42
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we should use standardized text for installation part (see how it's done for fastapi-related section) and move it to the beginning (inside
|
||
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 SQLAlchemy `Column(...)` when using `sa_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__` and `UniqueConstraint`. | ||
|
||
{* ./docs_src/advanced/sa_column/tutorial002.py *} | ||
|
||
### Key points | ||
|
||
- Single‑column unique can be declared with `Field(unique=True)` (simple case) or on the SQLAlchemy `Column(..., unique=True)` via `sa_column` when you need full control over type/nullable/name. `Field(unique=True)` is shorthand for setting `unique=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: | ||
|
||
```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 protected]) | ||
Adding Hero 2: Roy Kent ([email protected]) | ||
Adding Hero 3: Keeley Jones ([email protected]) | ||
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='[email protected]' | ||
|
||
Selecting by composite key (name, secret_name): | ||
Hero 2: name='Roy Kent' id=2 age=None secret_name='Roy' email='[email protected]' | ||
``` | ||
|
||
## Important considerations | ||
|
||
- **Prefer** built‑in `Field()` parameters (like `unique=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 other `Field()` 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`, or `UUID` from `sqlalchemy.dialects.postgresql` when appropriate. | ||
|
||
## See also | ||
|
||
- SQLAlchemy Column docs: <a href="https://docs.sqlalchemy.org/en/20/core/metadata.html#sqlalchemy.schema.Column" class="external-link" target="_blank">`Column`</a> | ||
- Advanced SQLModel topics: <a href="./index.md" class="internal-link">Advanced User Guide</a> |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,88 @@ | ||
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() |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,83 @@ | ||
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="[email protected]", name="Ted Lasso", secret_name="Coach") | ||
hero_2 = Hero(email="[email protected]", name="Roy Kent", secret_name="Roy") | ||
hero_3 = Hero( | ||
email="[email protected]", name="Keeley Jones", secret_name="Keeley" | ||
) | ||
|
||
print("Adding Hero 1: Ted Lasso ([email protected])") | ||
print("Adding Hero 2: Roy Kent ([email protected])") | ||
print("Adding Hero 3: Keeley Jones ([email protected])") | ||
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="[email protected]", 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 == "[email protected]") | ||
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() |
Uh oh!
There was an error while loading. Please reload this page.