Implement access token strategy db adapter

Author: frankie567

fastapi_users_db_tortoise/access_token.py

@@ -0,0 +1,63 @@
+from datetime import datetime
+from typing import Generic, Optional, Type, cast
+
+from fastapi_users.authentication.strategy.db import A, AccessTokenDatabase
+from tortoise import fields, models
+from tortoise.contrib.pydantic import PydanticModel
+
+
+class TortoiseBaseAccessTokenModel(models.Model):
+    token = fields.CharField(pk=True, max_length=43)
+    created_at = fields.DatetimeField(
+        null=False,
+        auto_now_add=True,
+    )
+
+
+class TortoiseAccessTokenDatabase(AccessTokenDatabase, Generic[A]):
+    """
+    Access token database adapter for Tortoise ORM.
+
+    :param access_token_model: Pydantic model of a DB representation of an access token.
+    :param model: Tortoise ORM model.
+    """
+
+    def __init__(
+        self, access_token_model: Type[A], model: Type[TortoiseBaseAccessTokenModel]
+    ):
+        self.access_token_model = access_token_model
+        self.model = model
+
+    async def get_by_token(
+        self, token: str, max_age: Optional[datetime] = None
+    ) -> Optional[A]:
+        query = self.model.filter(token=token)
+        if max_age is not None:
+            query = query.filter(created_at__gte=max_age)
+
+        access_token = await query.first()
+        if access_token is not None:
+            return await self._model_to_pydantic(access_token)
+        return None
+
+    async def create(self, access_token: A) -> A:
+        model = self.model(**access_token.dict())
+        await model.save()
+        await model.refresh_from_db()
+        return await self._model_to_pydantic(model)
+
+    async def update(self, access_token: A) -> A:
+        model = await self.model.get(token=access_token.token)
+        for field, value in access_token.dict().items():
+            setattr(model, field, value)
+        await model.save()
+        return await self._model_to_pydantic(model)
+
+    async def delete(self, access_token: A) -> None:
+        await self.model.filter(token=access_token.token).delete()
+
+    async def _model_to_pydantic(self, model: TortoiseBaseAccessTokenModel) -> A:
+        pydantic_access_token = await cast(
+            PydanticModel, self.access_token_model
+        ).from_tortoise_orm(model)
+        return cast(A, pydantic_access_token)

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",
     "tortoise-orm >=0.17.6,<0.18.0"
 ]
 

requirements.txt

@@ -1,2 +1,2 @@
-fastapi-users >= 6.1.2
+fastapi-users >= 9.1.0
 tortoise-orm >= 0.17.6,<0.18.0

tests/test_access_token.py

@@ -0,0 +1,124 @@
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import AsyncGenerator
+
+import pytest
+from fastapi_users.authentication.strategy.db.models import BaseAccessToken
+from pydantic import UUID4
+from tortoise import Tortoise, fields
+from tortoise.contrib.pydantic import PydanticModel
+from tortoise.exceptions import IntegrityError
+
+from fastapi_users_db_tortoise import TortoiseBaseUserModel
+from fastapi_users_db_tortoise.access_token import (
+    TortoiseAccessTokenDatabase,
+    TortoiseBaseAccessTokenModel,
+)
+from tests.conftest import UserDB as BaseUserDB
+
+
+class UserModel(TortoiseBaseUserModel):
+    pass
+
+
+class UserDB(BaseUserDB, PydanticModel):
+    class Config:
+        orm_mode = True
+        orig_model = UserModel
+
+
+class AccessTokenModel(TortoiseBaseAccessTokenModel):
+    user = fields.ForeignKeyField("models.UserModel", related_name="access_tokens")
+
+
+class AccessToken(BaseAccessToken, PydanticModel):
+    class Config:
+        orm_mode = True
+        orig_model = AccessTokenModel
+
+
+@pytest.fixture
+def user_id() -> UUID4:
+    return uuid.uuid4()
+
+
+@pytest.fixture
+async def tortoise_access_token_db(
+    user_id: UUID4,
+) -> AsyncGenerator[TortoiseAccessTokenDatabase, None]:
+    DATABASE_URL = "sqlite://./test-tortoise-access-token.db"
+
+    await Tortoise.init(
+        db_url=DATABASE_URL,
+        modules={"models": ["tests.test_access_token"]},
+    )
+    await Tortoise.generate_schemas()
+
+    user = UserModel(
+        id=user_id,
+        email="[email protected]",
+        hashed_password="guinevere",
+        is_active=True,
+        is_verified=True,
+        is_superuser=False,
+    )
+    await user.save()
+
+    yield TortoiseAccessTokenDatabase(AccessToken, AccessTokenModel)
+
+    await AccessTokenModel.all().delete()
+    await UserModel.all().delete()
+    await Tortoise.close_connections()
+
+
+@pytest.mark.asyncio
+@pytest.mark.db
+async def test_queries(
+    tortoise_access_token_db: TortoiseAccessTokenDatabase[AccessToken],
+    user_id: UUID4,
+):
+    access_token = AccessToken(token="TOKEN", user_id=user_id)
+
+    # Create
+    access_token_db = await tortoise_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 tortoise_access_token_db.update(access_token_db)
+
+    # Get by token
+    access_token_by_token = await tortoise_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 tortoise_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 tortoise_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 tortoise_access_token_db.get_by_token(
+        "NOT_EXISTING_TOKEN"
+    )
+    assert access_token_by_token is None
+
+    # Exception when inserting existing token
+    with pytest.raises(IntegrityError):
+        await tortoise_access_token_db.create(access_token_db)
+
+    # Delete token
+    await tortoise_access_token_db.delete(access_token_db)
+    deleted_access_token = await tortoise_access_token_db.get_by_token(
+        access_token_db.token
+    )
+    assert deleted_access_token is None

tests/test_fastapi_users_db_tortoise.py renamed to tests/test_users.py

@@ -10,7 +10,8 @@
     TortoiseBaseUserModel,
     TortoiseUserDatabase,
 )
-from tests.conftest import UserDB as BaseUserDB, UserDBOAuth as BaseUserDBOAuth
+from tests.conftest import UserDB as BaseUserDB
+from tests.conftest import UserDBOAuth as BaseUserDBOAuth
 
 
 class User(TortoiseBaseUserModel):
@@ -39,7 +40,7 @@ async def tortoise_user_db() -> AsyncGenerator[TortoiseUserDatabase, None]:
 
     await Tortoise.init(
         db_url=DATABASE_URL,
-        modules={"models": ["tests.test_fastapi_users_db_tortoise"]},
+        modules={"models": ["tests.test_users"]},
     )
     await Tortoise.generate_schemas()
 
@@ -55,7 +56,7 @@ async def tortoise_user_db_oauth() -> AsyncGenerator[TortoiseUserDatabase, None]
 
     await Tortoise.init(
         db_url=DATABASE_URL,
-        modules={"models": ["tests.test_fastapi_users_db_tortoise"]},
+        modules={"models": ["tests.test_users"]},
     )
     await Tortoise.generate_schemas()