Skip to content

Commit 945b445

Browse files
committed
fix(tests,docs): green CI — table fixtures, runnable doc examples, topmost imports
- Replace per-test create_all(tables=[...]) + drop_all in the encrypted-type integration tests with a shared fixture doing metadata.create_all(engine) (mirrors the password-hash tests). The old pattern did not create the BigIntBase PK sequence on a fresh database, causing "relation <table>_id_seq does not exist" on PostgreSQL/Oracle/MSSQL CI. - Make the new docs code examples self-contained and runnable under Sybil: build HashedPassword/TOTPProvider directly instead of referencing undefined model instances, and import HashedPassword/StoredObject from the topmost advanced_alchemy.types path. - Add TOTP secret, one-time-code, and rehash-on-verify to the README feature list. The crypto type imports remain safe when their optional libraries are absent: importing advanced_alchemy.types pulls in none of cryptography/pyotp/argon2/ passlib/pwdlib (guards fire at construction time, not import time).
1 parent e9f517c commit 945b445

5 files changed

Lines changed: 88 additions & 86 deletions

File tree

README.md

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,9 @@ offering:
4747
- Integrated support for UUID6 and UUID7 ([`uuid-utils`](https://github.com/aminalaee/uuid-utils) when installed, otherwise native in Python 3.14+, with UUID4 fallback on older versions).
4848
- Integrated support for Nano ID using [`fastnanoid`](https://github.com/oliverlambson/fastnanoid) (install with the `nanoid` extra)
4949
- Custom encrypted text type with multiple backend support including [`pgcrypto`](https://www.postgresql.org/docs/current/pgcrypto.html) for PostgreSQL and the Fernet implementation from [`cryptography`](https://cryptography.io/en/latest/) for other databases
50-
- Custom password hashing type with multiple backend support including [`Argon2`](https://github.com/P-H-C/phc-winner-argon2), [`Passlib`](https://passlib.readthedocs.io/en/stable/), and [`Pwdlib`](https://pwdlib.readthedocs.io/en/stable/) with automatic salt generation
50+
- Custom password hashing type with multiple backend support including [`Argon2`](https://github.com/P-H-C/phc-winner-argon2), [`Passlib`](https://passlib.readthedocs.io/en/stable/), and [`Pwdlib`](https://pwdlib.readthedocs.io/en/stable/) with automatic salt generation and rehash-on-verify for transparent hash upgrades on login
51+
- Custom TOTP shared-secret type that stores an authenticator seed encrypted at rest and verifies time-based one-time passwords via [`pyotp`](https://github.com/pyauth/pyotp) (install with the `pyotp` extra)
52+
- Custom one-time-code type for hashed, single-use email/SMS verification codes, reusing the password-hash backends
5153
- Pre-configured base classes with audit columns UUID or Big Integer primary keys and
5254
a [sentinel column](https://docs.sqlalchemy.org/en/20/core/connections.html#configuring-sentinel-columns).
5355
- Synchronous and asynchronous repositories featuring:

docs/usage/frameworks/fastapi.rst

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -255,9 +255,8 @@ Start by registering a backend and mapping the file column:
255255
from sqlalchemy.orm import Mapped, mapped_column
256256
257257
from advanced_alchemy.extensions.fastapi import base, repository, service
258-
from advanced_alchemy.types import FileObject, storages
258+
from advanced_alchemy.types import FileObject, storages, StoredObject
259259
from advanced_alchemy.types.file_object.backends.obstore import ObstoreBackend
260-
from advanced_alchemy.types.file_object.data_type import StoredObject
261260
262261
documents_backend = ObstoreBackend(
263262
key="documents",

docs/usage/frameworks/litestar.rst

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -996,9 +996,8 @@ The pattern below mirrors ``examples/litestar/litestar_fileobject.py``:
996996
repository,
997997
service,
998998
)
999-
from advanced_alchemy.types import FileObject, storages
999+
from advanced_alchemy.types import FileObject, storages, StoredObject
10001000
from advanced_alchemy.types.file_object.backends.obstore import ObstoreBackend
1001-
from advanced_alchemy.types.file_object.data_type import StoredObject
10021001
10031002
# Configure file storage backend
10041003
documents_backend = ObstoreBackend(

docs/usage/modeling/types.rst

Lines changed: 21 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -165,15 +165,21 @@ explicitly — each pulls a different optional dependency:
165165
166166
password: Mapped[str] = mapped_column(PasswordHash(backend=Argon2Hasher()))
167167
168-
On read, ``account.password`` is a ``HashedPassword``. Verify with ``verify``, or use
169-
``verify_and_update`` to transparently upgrade a hash that was created with weaker
170-
parameters after a successful login:
168+
On read, ``account.password`` is a :class:`HashedPassword <advanced_alchemy.types.HashedPassword>`.
169+
Verify with ``verify``, or use ``verify_and_update`` to transparently upgrade a hash created
170+
with weaker parameters after a successful login. The same wrapper can be built standalone:
171171

172172
.. code-block:: python
173173
174-
ok, new_hash = account.password.verify_and_update(submitted_password)
174+
from advanced_alchemy.types import HashedPassword
175+
from advanced_alchemy.types.password_hash.argon2 import Argon2Hasher
176+
177+
backend = Argon2Hasher()
178+
stored = HashedPassword(backend.hash("s3cret"), backend)
179+
180+
ok, new_hash = stored.verify_and_update("s3cret")
175181
if ok and new_hash is not None:
176-
account.password = new_hash # re-hashed with current parameters; persist it
182+
... # persist new_hash back to the row
177183
178184
The default column ``length`` is 255 to accommodate stacked passlib schemes.
179185

@@ -197,12 +203,17 @@ extra (``pip install advanced_alchemy[pyotp]``); a ``key`` is mandatory.
197203
198204
seed: Mapped[str] = mapped_column(TOTPSecret(key="my-secret-key", issuer="ACME"))
199205
200-
# enrollment
201-
enrollment = MfaEnrollment(seed=generate_totp_secret())
202-
uri = enrollment.seed.provisioning_uri(name="alice@example.com") # render as a QR code
206+
After loading a row, ``enrollment.seed`` is a
207+
:class:`TOTPProvider <advanced_alchemy.types.TOTPProvider>`. Build one directly to generate a
208+
provisioning URI (render it as a QR code) and verify submitted codes:
209+
210+
.. code-block:: python
211+
212+
from advanced_alchemy.types import TOTPProvider, generate_totp_secret
203213
204-
# verification (tolerates one tick of clock drift by default)
205-
is_valid = enrollment.seed.verify(submitted_code)
214+
provider = TOTPProvider(generate_totp_secret(), issuer="ACME")
215+
uri = provider.provisioning_uri(name="alice@example.com")
216+
is_valid = provider.verify("123456") # tolerates one tick of clock drift by default
206217
207218
One-Time Codes
208219
--------------
Lines changed: 62 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
"""Integration round-trip tests for the encrypted column types."""
22

3-
from typing import Optional, cast
3+
from typing import Any, Optional, cast
44

55
import pytest
6-
from sqlalchemy import Engine, String, Table, text
6+
from sqlalchemy import Engine, String, text
77
from sqlalchemy.orm import Mapped, Session, mapped_column, sessionmaker
88

99
from advanced_alchemy.base import BigIntBase
@@ -50,6 +50,14 @@ class TOTPModel(BigIntBase):
5050
__table_args__ = {"info": {"allow_eager": True}}
5151

5252

53+
@pytest.fixture()
54+
def encrypted_test_tables(engine: Engine) -> None:
55+
"""Create the encrypted-type test tables for sync engines."""
56+
name = getattr(engine.dialect, "name", "")
57+
if name != "mock" and not name.startswith("spanner"):
58+
FernetModel.metadata.create_all(engine)
59+
60+
5361
def _should_skip(engine: Engine) -> Optional[str]:
5462
name = engine.dialect.name
5563
if name == "mock":
@@ -61,51 +69,43 @@ def _should_skip(engine: Engine) -> Optional[str]:
6169
return None
6270

6371

64-
def test_fernet_round_trip(engine: Engine) -> None:
72+
def test_fernet_round_trip(engine: Engine, encrypted_test_tables: None) -> None:
6573
skip_reason = _should_skip(engine)
6674
if skip_reason is not None:
6775
pytest.skip(skip_reason)
6876

69-
FernetModel.metadata.create_all(engine, tables=[cast("Table", FernetModel.__table__)])
7077
session_factory: sessionmaker[Session] = sessionmaker(engine, expire_on_commit=False)
71-
try:
72-
with session_factory() as db_session:
73-
obj = FernetModel(name="fernet", secret="top-secret-value")
74-
db_session.add(obj)
75-
db_session.commit()
76-
db_session.refresh(obj)
77-
assert obj.secret == "top-secret-value"
78+
with session_factory() as db_session:
79+
obj = FernetModel(name="fernet", secret="top-secret-value")
80+
db_session.add(obj)
81+
db_session.commit()
82+
db_session.refresh(obj)
83+
assert obj.secret == "top-secret-value"
7884

79-
obj.secret = None
80-
db_session.commit()
81-
db_session.refresh(obj)
82-
assert obj.secret is None
83-
finally:
84-
FernetModel.metadata.drop_all(engine, tables=[cast("Table", FernetModel.__table__)])
85+
obj.secret = None
86+
db_session.commit()
87+
db_session.refresh(obj)
88+
assert obj.secret is None
8589

8690

87-
def test_pgcrypto_round_trip(engine: Engine) -> None:
91+
def test_pgcrypto_round_trip(engine: Engine, encrypted_test_tables: None) -> None:
8892
if engine.dialect.name != "postgresql":
8993
pytest.skip("pgcrypto requires a real PostgreSQL backend")
9094

9195
with engine.begin() as conn:
9296
conn.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
9397

94-
PGCryptoModel.metadata.create_all(engine, tables=[cast("Table", PGCryptoModel.__table__)])
9598
session_factory: sessionmaker[Session] = sessionmaker(engine, expire_on_commit=False)
96-
try:
97-
with session_factory() as db_session:
98-
obj = PGCryptoModel(name="pgcrypto", secret="pg-secret-value")
99-
db_session.add(obj)
100-
db_session.commit()
101-
db_session.refresh(obj)
102-
assert obj.secret == "pg-secret-value"
103-
finally:
104-
PGCryptoModel.metadata.drop_all(engine, tables=[cast("Table", PGCryptoModel.__table__)])
99+
with session_factory() as db_session:
100+
obj = PGCryptoModel(name="pgcrypto", secret="pg-secret-value")
101+
db_session.add(obj)
102+
db_session.commit()
103+
db_session.refresh(obj)
104+
assert obj.secret == "pg-secret-value"
105105

106106

107107
@pytest.mark.parametrize("model", [FernetModel, PGCryptoModel])
108-
def test_backends_are_interchangeable(engine: Engine, model: "type[BigIntBase]") -> None:
108+
def test_backends_are_interchangeable(engine: Engine, model: "type[BigIntBase]", encrypted_test_tables: None) -> None:
109109
"""Fernet and PGCrypto must be drop-in interchangeable: same API, same external behavior.
110110
111111
Run on PostgreSQL where both backends work (pgcrypto is PostgreSQL-only) so the two are
@@ -117,32 +117,27 @@ def test_backends_are_interchangeable(engine: Engine, model: "type[BigIntBase]")
117117
with engine.begin() as conn:
118118
conn.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
119119

120-
table = cast("Table", model.__table__)
121-
model.metadata.create_all(engine, tables=[table])
122120
session_factory: sessionmaker[Session] = sessionmaker(engine, expire_on_commit=False)
123-
try:
124-
with session_factory() as db_session:
125-
for value in ["plain-value", "ünîcödé 🔐 secret", "", None]:
126-
obj = model(name="x", secret=value)
127-
db_session.add(obj)
128-
db_session.commit()
129-
db_session.refresh(obj)
130-
assert obj.secret == value
131-
132-
raw = db_session.execute(
133-
text(f"SELECT secret FROM {model.__tablename__} WHERE id = :id"),
134-
{"id": obj.id},
135-
).scalar_one()
136-
if value is None:
137-
assert raw is None
138-
else:
139-
assert raw != value
140-
finally:
141-
model.metadata.drop_all(engine, tables=[table])
121+
with session_factory() as db_session:
122+
for value in ["plain-value", "ünîcödé 🔐 secret", "", None]:
123+
obj = cast("Any", model(name="x", secret=value))
124+
db_session.add(obj)
125+
db_session.commit()
126+
db_session.refresh(obj)
127+
assert obj.secret == value
128+
129+
raw = db_session.execute(
130+
text(f"SELECT secret FROM {model.__tablename__} WHERE id = :id"),
131+
{"id": obj.id},
132+
).scalar_one()
133+
if value is None:
134+
assert raw is None
135+
else:
136+
assert raw != value
142137

143138

144139
@pytest.mark.skipif(not PYOTP_INSTALLED, reason="pyotp not installed")
145-
def test_totp_secret_round_trip(engine: Engine) -> None:
140+
def test_totp_secret_round_trip(engine: Engine, encrypted_test_tables: None) -> None:
146141
import pyotp
147142

148143
from advanced_alchemy.types.totp import TOTPProvider
@@ -152,24 +147,20 @@ def test_totp_secret_round_trip(engine: Engine) -> None:
152147
pytest.skip(skip_reason)
153148

154149
secret = pyotp.random_base32()
155-
TOTPModel.metadata.create_all(engine, tables=[cast("Table", TOTPModel.__table__)])
156150
session_factory: sessionmaker[Session] = sessionmaker(engine, expire_on_commit=False)
157-
try:
158-
with session_factory() as db_session:
159-
obj = TOTPModel(name="totp", seed=secret)
160-
db_session.add(obj)
161-
db_session.commit()
162-
row_id = obj.id
163-
db_session.refresh(obj)
164-
165-
provider = cast("object", obj.seed)
166-
assert isinstance(provider, TOTPProvider)
167-
assert provider.secret == secret
168-
assert provider.verify(pyotp.TOTP(secret).now()) is True
169-
170-
stored = db_session.execute(
171-
text("SELECT seed FROM test_encrypted_totp WHERE id = :id"), {"id": row_id}
172-
).scalar_one()
173-
assert stored != secret
174-
finally:
175-
TOTPModel.metadata.drop_all(engine, tables=[cast("Table", TOTPModel.__table__)])
151+
with session_factory() as db_session:
152+
obj = TOTPModel(name="totp", seed=secret)
153+
db_session.add(obj)
154+
db_session.commit()
155+
row_id = obj.id
156+
db_session.refresh(obj)
157+
158+
provider = cast("object", obj.seed)
159+
assert isinstance(provider, TOTPProvider)
160+
assert provider.secret == secret
161+
assert provider.verify(pyotp.TOTP(secret).now()) is True
162+
163+
stored = db_session.execute(
164+
text("SELECT seed FROM test_encrypted_totp WHERE id = :id"), {"id": row_id}
165+
).scalar_one()
166+
assert stored != secret

0 commit comments

Comments
 (0)