Replace global config with registry and add Flask example

Author: script3r

README.md

@@ -14,13 +14,9 @@ pip install sqlmodel-encrypted-fields
 from sqlalchemy import Column
 from sqlmodel import Field, SQLModel
 
-from sqlmodel_encrypted_fields import (
-    configure_keysets,
-    EncryptedString,
-    DeterministicEncryptedString,
-)
+from sqlmodel_encrypted_fields import KeysetRegistry
 
-configure_keysets(
+registry = KeysetRegistry(
     {
         "default": {"path": "/path/to/aead_keyset.json", "cleartext": True},
         "searchable": {"path": "/path/to/daead_keyset.json", "cleartext": True},
@@ -30,8 +26,28 @@ configure_keysets(
 
 class Customer(SQLModel, table=True):
     id: int | None = Field(default=None, primary_key=True)
-    email: str = Field(sa_column=Column(EncryptedString()))
-    email_lookup: str = Field(sa_column=Column(DeterministicEncryptedString(keyset="searchable")))
+    email: str = Field(sa_column=Column(registry.encrypted_string()))
+    email_lookup: str = Field(sa_column=Column(registry.deterministic_encrypted_string(keyset="searchable")))
+```
+
+## Example Apps
+
+- FastAPI example: `example_app_fastapi/`
+- Flask example: `example_app_flask/`
+
+### Flask Example Snippet
+
+```python
+from flask import Flask
+
+from example_app_flask.database import init_db
+from example_app_flask.models import Customer
+
+app = Flask(__name__)
+
+@app.before_first_request
+def _init_db() -> None:
+    init_db()
 ```
 
 ## Notes
@@ -54,7 +70,13 @@ Regular AEAD fields change ciphertext on every write. Use deterministic fields o
 ```python
 from datetime import date
 
-from sqlmodel_encrypted_fields import EncryptedType
+from sqlmodel_encrypted_fields import KeysetRegistry
+
+registry = KeysetRegistry(
+    {
+        "default": {"path": "/path/to/aead_keyset.json", "cleartext": True},
+    }
+)
 
 
 def serialize_date(value: date) -> str:
@@ -67,5 +89,7 @@ def deserialize_date(value: str) -> date:
 
 class User(SQLModel, table=True):
     id: int | None = Field(default=None, primary_key=True)
-    birthday: date = Field(sa_column=EncryptedType(serializer=serialize_date, deserializer=deserialize_date))
+    birthday: date = Field(
+        sa_column=registry.encrypted_type(serializer=serialize_date, deserializer=deserialize_date)
+    )
 ```

example_app/models.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from sqlmodel_encrypted_fields import KeysetRegistry
+
+
+def _default_config() -> dict[str, dict[str, object]]:
+    root = Path(__file__).resolve().parents[1]
+    return {
+        "default": {"path": str(root / "tests" / "fixtures" / "aead_keyset.json"), "cleartext": True},
+        "deterministic": {
+            "path": str(root / "tests" / "fixtures" / "daead_keyset.json"),
+            "cleartext": True,
+        },
+    }
+
+
+registry = KeysetRegistry(_default_config())

example_app/__init__.py example_app_fastapi/__init__.pyexample_app/__init__.py renamed to example_app_fastapi/__init__.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from sqlmodel import Session, SQLModel, create_engine
+
+engine = create_engine("sqlite:///./example_app_fastapi.db", echo=False)
+
+
+def init_db() -> None:
+    SQLModel.metadata.create_all(engine)
+
+
+def get_session() -> Session:
+    return Session(engine)

example_app_fastapi/crypto.py

@@ -5,18 +5,11 @@
 from fastapi import Depends, FastAPI, HTTPException
 from sqlmodel import Session, select
 
-from example_app.database import get_session, init_db
-from example_app.models import Customer
-from sqlmodel_encrypted_fields import configure_keysets
+from example_app_fastapi.database import get_session, init_db
+from example_app_fastapi.models import Customer
 
 @asynccontextmanager
 async def lifespan(_: FastAPI):
-    configure_keysets(
-        {
-            "default": {"path": "./tests/fixtures/aead_keyset.json", "cleartext": True},
-            "deterministic": {"path": "./tests/fixtures/daead_keyset.json", "cleartext": True},
-        }
-    )
     init_db()
     yield
 

example_app_fastapi/database.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from sqlalchemy import Column
+from sqlmodel import Field, SQLModel
+
+from example_app_fastapi.crypto import registry
+
+
+class Customer(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    email: str = Field(sa_column=Column(registry.encrypted_string()))
+    email_lookup: str = Field(sa_column=Column(registry.deterministic_encrypted_string(keyset="deterministic")))

example_app/main.py example_app_fastapi/main.pyexample_app/main.py renamed to example_app_fastapi/main.py

@@ -5,33 +5,16 @@
 from fastapi.testclient import TestClient
 from sqlmodel import Session, SQLModel, create_engine, select
 
-from example_app.database import get_session
-from example_app.main import app
-from example_app.models import Customer
-from sqlmodel_encrypted_fields import configure_keysets
-
-
-def _project_root() -> Path:
-    return Path(__file__).resolve().parents[2]
-
-
-def _configure_keysets() -> None:
-    root = _project_root()
-    configure_keysets(
-        {
-            "default": {"path": str(root / "tests" / "fixtures" / "aead_keyset.json"), "cleartext": True},
-            "deterministic": {"path": str(root / "tests" / "fixtures" / "daead_keyset.json"), "cleartext": True},
-        }
-    )
+from example_app_fastapi.database import get_session
+from example_app_fastapi.main import app
+from example_app_fastapi.models import Customer
 
 
 def _test_engine(tmp_path: Path):
     return create_engine(f"sqlite:///{tmp_path / 'test.db'}", echo=False)
 
 
 def test_customer_create_and_lookup(tmp_path: Path) -> None:
-    _configure_keysets()
-
     engine = _test_engine(tmp_path)
     SQLModel.metadata.create_all(engine)
 

example_app_fastapi/models.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from flask import Flask, jsonify, request
+from sqlmodel import Session, select
+
+from example_app_flask.database import get_session, init_db
+from example_app_flask.models import Customer
+
+
+def create_app() -> Flask:
+    app = Flask(__name__)
+
+    @app.before_first_request
+    def _init_db() -> None:
+        init_db()
+
+    @app.post("/customers")
+    def create_customer():
+        payload = request.get_json(force=True)
+        customer = Customer(**payload)
+        with get_session() as session:
+            session.add(customer)
+            session.commit()
+            session.refresh(customer)
+            return jsonify(customer.model_dump())
+
+    @app.get("/customers/<int:customer_id>")
+    def get_customer(customer_id: int):
+        with get_session() as session:
+            customer = session.get(Customer, customer_id)
+            if customer is None:
+                return jsonify({"detail": "Customer not found"}), 404
+            return jsonify(customer.model_dump())
+
+    @app.get("/customers/by-email/<string:email>")
+    def get_customer_by_email(email: str):
+        with get_session() as session:
+            statement = select(Customer).where(Customer.email_lookup == email)
+            customer = session.exec(statement).first()
+            if customer is None:
+                return jsonify({"detail": "Customer not found"}), 404
+            return jsonify(customer.model_dump())
+
+    return app
+
+
+if __name__ == "__main__":
+    create_app().run(debug=True)