create access tokens table

alembic/versions/2024_10_02_1319-16aa19936ef4_add_access_token_table.py

@@ -0,0 +1,55 @@
+"""add access token table
+
+Revision ID: 16aa19936ef4
+Revises: 756bbba724e8
+Create Date: 2024-10-02 13:19:48.314416
+
+"""
+
+from typing import Sequence, Union
+
+import fastapi_users_db_sqlalchemy
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "16aa19936ef4"
+down_revision: Union[str, None] = "756bbba724e8"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "access_tokens",
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("token", sa.String(length=43), nullable=False),
+        sa.Column(
+            "created_at",
+            fastapi_users_db_sqlalchemy.generics.TIMESTAMPAware(timezone=True),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["users.id"],
+            name=op.f("fk_access_tokens_user_id_users"),
+            ondelete="cascade",
+        ),
+        sa.PrimaryKeyConstraint("token", name=op.f("pk_access_tokens")),
+    )
+    op.create_index(
+        op.f("ix_access_tokens_created_at"),
+        "access_tokens",
+        ["created_at"],
+        unique=False,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f("ix_access_tokens_created_at"), table_name="access_tokens")
+    op.drop_table("access_tokens")
+    # ### end Alembic commands ###

authentication/transport.py

@@ -0,0 +1,8 @@
+from fastapi_users.authentication import BearerTransport
+
+from config import settings
+
+bearer_transport = BearerTransport(
+    # TODO: update url
+    tokenUrl="auth/jwt/login",
+)

models/__init__.py

@@ -3,9 +3,12 @@ from .base import Base
 
 from .user import User
 
+from .access_token import AccessToken
+
 
 __all__ = (
     "db_helper",
     "Base",
     "User",
+    "AccessToken",
 )

models/access_token.py

@@ -0,0 +1,33 @@
+from typing import TYPE_CHECKING
+
+from fastapi_users_db_sqlalchemy.access_token import (
+    SQLAlchemyAccessTokenDatabase,
+    SQLAlchemyBaseAccessTokenTable,
+)
+from sqlalchemy import (
+    Integer,
+    ForeignKey,
+)
+from sqlalchemy.orm import (
+    Mapped,
+    mapped_column,
+)
+
+
+from config import settings
+from .base import Base
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+class AccessToken(Base, SQLAlchemyBaseAccessTokenTable[settings.type.UserIdType]):
+    user_id: Mapped[settings.type.UserIdType] = mapped_column(
+        Integer,
+        ForeignKey("users.id", ondelete="cascade"),
+        nullable=False,
+    )
+
+    @classmethod
+    def get_db(cls, session: "AsyncSession"):
+        return SQLAlchemyAccessTokenDatabase(session, cls)