Implement access token strategy db adapter

Author: frankie567

fastapi_users_db_sqlalchemy/__init__.py

@@ -1,55 +1,16 @@
 """FastAPI Users database adapter for SQLAlchemy + encode/databases."""
-import uuid
 from typing import Mapping, Optional, Type
 
 from databases import Database
 from fastapi_users.db.base import BaseUserDatabase
 from fastapi_users.models import UD
 from pydantic import UUID4
 from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Table, func, select
-from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.ext.declarative import declared_attr
-from sqlalchemy.types import CHAR, TypeDecorator
 
-__version__ = "1.0.0"
-
-
-class GUID(TypeDecorator):  # pragma: no cover
-    """Platform-independent GUID type.
+from fastapi_users_db_sqlalchemy.guid import GUID
 
-    Uses PostgreSQL's UUID type, otherwise uses
-    CHAR(36), storing as regular strings.
-    """
-
-    class UUIDChar(CHAR):
-        python_type = UUID4
-
-    impl = UUIDChar
-
-    def load_dialect_impl(self, dialect):
-        if dialect.name == "postgresql":
-            return dialect.type_descriptor(UUID())
-        else:
-            return dialect.type_descriptor(CHAR(36))
-
-    def process_bind_param(self, value, dialect):
-        if value is None:
-            return value
-        elif dialect.name == "postgresql":
-            return str(value)
-        else:
-            if not isinstance(value, uuid.UUID):
-                return str(uuid.UUID(value))
-            else:
-                return str(value)
-
-    def process_result_value(self, value, dialect):
-        if value is None:
-            return value
-        else:
-            if not isinstance(value, uuid.UUID):
-                value = uuid.UUID(value)
-            return value
+__version__ = "1.0.0"
 
 
 class SQLAlchemyBaseUserTable:

fastapi_users_db_sqlalchemy/access_token.py

@@ -0,0 +1,71 @@
+from datetime import datetime
+from typing import Generic, Optional, Type
+
+from databases import Database
+from fastapi_users.authentication.strategy.db import A, AccessTokenDatabase
+from sqlalchemy import Column, DateTime, ForeignKey, String, Table
+from sqlalchemy.ext.declarative import declared_attr
+
+from fastapi_users_db_sqlalchemy.guid import GUID
+
+
+class SQLAlchemyBaseAccessTokenTable:
+    """Base SQLAlchemy access token table definition."""
+
+    __tablename__ = "accesstoken"
+
+    token = Column(String(length=43), primary_key=True)
+    created_at = Column(DateTime(timezone=True), index=True, nullable=False)
+
+    @declared_attr
+    def user_id(cls):
+        return Column(GUID, ForeignKey("user.id", ondelete="cascade"), nullable=False)
+
+
+class SQLAlchemyAccessTokenDatabase(AccessTokenDatabase, Generic[A]):
+    """
+    Access token database adapter for SQLAlchemy.
+
+    :param access_token_model: Pydantic model of a DB representation of an access token.
+    :param database: `Database` instance from `encode/databases`.
+    :param access_tokens: SQLAlchemy access token table instance.
+    """
+
+    def __init__(
+        self, access_token_model: Type[A], database: Database, access_tokens: Table
+    ):
+        self.access_token_model = access_token_model
+        self.database = database
+        self.access_tokens = access_tokens
+
+    async def get_by_token(
+        self, token: str, max_age: Optional[datetime] = None
+    ) -> Optional[A]:
+        query = self.access_tokens.select().where(self.access_tokens.c.token == token)
+        if max_age is not None:
+            query = query.where(self.access_tokens.c.created_at >= max_age)
+
+        access_token = await self.database.fetch_one(query)
+        if access_token is not None:
+            return self.access_token_model(**access_token)
+        return None
+
+    async def create(self, access_token: A) -> A:
+        query = self.access_tokens.insert()
+        await self.database.execute(query, access_token.dict())
+        return access_token
+
+    async def update(self, access_token: A) -> A:
+        update_query = (
+            self.access_tokens.update()
+            .where(self.access_tokens.c.token == access_token.token)
+            .values(access_token.dict())
+        )
+        await self.database.execute(update_query)
+        return access_token
+
+    async def delete(self, access_token: A) -> None:
+        query = self.access_tokens.delete().where(
+            self.access_tokens.c.token == access_token.token
+        )
+        await self.database.execute(query)

fastapi_users_db_sqlalchemy/guid.py

@@ -0,0 +1,43 @@
+import uuid
+
+from pydantic import UUID4
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.types import CHAR, TypeDecorator
+
+
+class GUID(TypeDecorator):  # pragma: no cover
+    """Platform-independent GUID type.
+
+    Uses PostgreSQL's UUID type, otherwise uses
+    CHAR(36), storing as regular strings.
+    """
+
+    class UUIDChar(CHAR):
+        python_type = UUID4
+
+    impl = UUIDChar
+
+    def load_dialect_impl(self, dialect):
+        if dialect.name == "postgresql":
+            return dialect.type_descriptor(UUID())
+        else:
+            return dialect.type_descriptor(CHAR(36))
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return value
+        elif dialect.name == "postgresql":
+            return str(value)
+        else:
+            if not isinstance(value, uuid.UUID):
+                return str(uuid.UUID(value))
+            else:
+                return str(value)
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return value
+        else:
+            if not isinstance(value, uuid.UUID):
+                value = uuid.UUID(value)
+            return value

pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
 description-file = "README.md"
 requires-python = ">=3.7"
 requires = [
-    "fastapi-users >= 6.1.2",
+    "fastapi-users >= 9.1.0",
     "sqlalchemy >=1.4",
     "databases >=0.5"
 ]

requirements.txt

@@ -1,3 +1,3 @@
-fastapi-users >= 6.1.2
+fastapi-users >= 9.1.0
 sqlalchemy >=1.4
 databases[postgresql, sqlite] >=0.5

tests/test_access_token.py

@@ -0,0 +1,123 @@
+import sqlite3
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import AsyncGenerator
+
+import pytest
+import sqlalchemy
+from databases import Database
+from fastapi_users.authentication.strategy.db.models import BaseAccessToken
+from pydantic import UUID4
+from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
+
+from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTable
+from fastapi_users_db_sqlalchemy.access_token import (
+    SQLAlchemyAccessTokenDatabase,
+    SQLAlchemyBaseAccessTokenTable,
+)
+
+
+class AccessToken(BaseAccessToken):
+    pass
+
+
+@pytest.fixture
+def user_id() -> UUID4:
+    return uuid.uuid4()
+
+
+@pytest.fixture
+async def sqlalchemy_access_token_db(
+    user_id: UUID4,
+) -> AsyncGenerator[SQLAlchemyAccessTokenDatabase, None]:
+    Base: DeclarativeMeta = declarative_base()
+
+    class AccessTokenTable(SQLAlchemyBaseAccessTokenTable, Base):
+        pass
+
+    class UserTable(SQLAlchemyBaseUserTable, Base):
+        pass
+
+    DATABASE_URL = "sqlite:///./test-sqlalchemy-access-token.db"
+    database = Database(DATABASE_URL)
+
+    engine = sqlalchemy.create_engine(
+        DATABASE_URL, connect_args={"check_same_thread": False}
+    )
+    Base.metadata.create_all(engine)
+
+    await database.connect()
+
+    # Create user
+    query = UserTable.__table__.insert()
+    await database.execute(
+        query,
+        {
+            "id": user_id,
+            "email": "lancelot@camelot.bt",
+            "hashed_password": "guinevere",
+            "is_active": True,
+            "is_verified": False,
+            "is_superuser": False,
+        },
+    )
+
+    yield SQLAlchemyAccessTokenDatabase(
+        AccessToken, database, AccessTokenTable.__table__
+    )
+
+    Base.metadata.drop_all(engine)
+    await database.disconnect()
+
+
+@pytest.mark.asyncio
+@pytest.mark.db
+async def test_queries(
+    sqlalchemy_access_token_db: SQLAlchemyAccessTokenDatabase[AccessToken],
+    user_id: UUID4,
+):
+    access_token = AccessToken(token="TOKEN", user_id=user_id)
+
+    # Create
+    access_token_db = await sqlalchemy_access_token_db.create(access_token)
+    assert access_token_db.token == "TOKEN"
+    assert access_token_db.user_id == user_id
+
+    # Update
+    access_token_db.created_at = datetime.now(timezone.utc)
+    await sqlalchemy_access_token_db.update(access_token_db)
+
+    # Get by token
+    access_token_by_token = await sqlalchemy_access_token_db.get_by_token(
+        access_token_db.token
+    )
+    assert access_token_by_token is not None
+
+    # Get by token expired
+    access_token_by_token = await sqlalchemy_access_token_db.get_by_token(
+        access_token_db.token, max_age=datetime.now(timezone.utc) + timedelta(hours=1)
+    )
+    assert access_token_by_token is None
+
+    # Get by token not expired
+    access_token_by_token = await sqlalchemy_access_token_db.get_by_token(
+        access_token_db.token, max_age=datetime.now(timezone.utc) - timedelta(hours=1)
+    )
+    assert access_token_by_token is not None
+
+    # Get by token unknown
+    access_token_by_token = await sqlalchemy_access_token_db.get_by_token(
+        "NOT_EXISTING_TOKEN"
+    )
+    assert access_token_by_token is None
+
+    # Exception when inserting existing token
+    with pytest.raises(sqlite3.IntegrityError):
+        await sqlalchemy_access_token_db.create(access_token_db)
+
+    # Delete token
+    await sqlalchemy_access_token_db.delete(access_token_db)
+    deleted_access_token = await sqlalchemy_access_token_db.get_by_token(
+        access_token_db.token
+    )
+    assert deleted_access_token is None