Add API tokens support (#186)

* Add API tokens management in backend

* Generate client

* Add registration token generation

Author: ArchiDevil

backend/alembic/versions/56cc00f2d285_add_api_token_table.py

@@ -0,0 +1,42 @@
+"""Add API token table
+
+Revision ID: 56cc00f2d285
+Revises: fc5e85011fea
+Create Date: 2026-04-18 15:49:06.026121
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# pylint: disable=E1101
+
+# revision identifiers, used by Alembic.
+revision: str = '56cc00f2d285'
+down_revision: Union[str, None] = 'fc5e85011fea'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "api_token",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("token_hash", sa.String(), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("created_by", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("expires_at", sa.DateTime(), nullable=True),
+        sa.Column("last_used_at", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("token_hash"),
+        sa.ForeignKeyConstraint(["user_id"], ["user.id"]),
+        sa.ForeignKeyConstraint(["created_by"], ["user.id"]),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("api_token")

backend/app/__init__.py

@@ -1,3 +1,4 @@
+from app.api_token.models import ApiToken
 from app.comments.models import Comment
 from app.db import Base
 from app.documents.models import (
@@ -19,6 +20,7 @@
 from app.translation_memory.models import TranslationMemory, TranslationMemoryRecord
 
 __all__ = [
+    "ApiToken",
     "Base",
     "Comment",
     "DocumentTask",

backend/app/api_token/__init__.py

@@ -0,0 +1,3 @@
+from app.api_token.models import ApiToken
+
+__all__ = ["ApiToken"]

backend/app/api_token/models.py

@@ -0,0 +1,30 @@
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.db import Base
+
+if TYPE_CHECKING:
+    from app.schema import User
+
+
+def utc_time():
+    return datetime.now(UTC)
+
+
+class ApiToken(Base):
+    __tablename__ = "api_token"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column()
+    token_hash: Mapped[str] = mapped_column(unique=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"))
+    created_by: Mapped[int] = mapped_column(ForeignKey("user.id"))
+    created_at: Mapped[datetime] = mapped_column(default=utc_time)
+    expires_at: Mapped[datetime | None] = mapped_column(default=None)
+    last_used_at: Mapped[datetime | None] = mapped_column(default=None)
+
+    user: Mapped["User"] = relationship("User", foreign_keys=[user_id])
+    created_by_user: Mapped["User"] = relationship("User", foreign_keys=[created_by])

backend/app/api_token/query.py

@@ -0,0 +1,69 @@
+from datetime import UTC, datetime
+from typing import Sequence
+
+from sqlalchemy import select, update
+from sqlalchemy.orm import Session
+
+from app.api_token.models import ApiToken
+from app.base.exceptions import BaseQueryException
+
+
+class ApiTokenNotFoundExc(BaseQueryException):
+    pass
+
+
+class ApiTokenQuery:
+    def __init__(self, db: Session) -> None:
+        self.__db = db
+
+    def create(
+        self,
+        name: str,
+        token_hash: str,
+        user_id: int,
+        created_by: int,
+        expires_at: datetime | None = None,
+    ) -> ApiToken:
+        token = ApiToken(
+            name=name,
+            token_hash=token_hash,
+            user_id=user_id,
+            created_by=created_by,
+            expires_at=expires_at,
+        )
+        self.__db.add(token)
+        self.__db.commit()
+        self.__db.refresh(token)
+        return token
+
+    def get_by_hash(self, token_hash: str) -> ApiToken | None:
+        return self.__db.execute(
+            select(ApiToken).where(ApiToken.token_hash == token_hash)
+        ).scalar_one_or_none()
+
+    def get_all(self) -> Sequence[ApiToken]:
+        return (
+            self.__db.execute(select(ApiToken).order_by(ApiToken.created_at.desc()))
+            .scalars()
+            .all()
+        )
+
+    def get_by_id(self, token_id: int) -> ApiToken | None:
+        return self.__db.execute(
+            select(ApiToken).where(ApiToken.id == token_id)
+        ).scalar_one_or_none()
+
+    def delete(self, token_id: int) -> None:
+        token = self.get_by_id(token_id)
+        if not token:
+            raise ApiTokenNotFoundExc()
+        self.__db.delete(token)
+        self.__db.commit()
+
+    def update_last_used(self, token_id: int) -> None:
+        self.__db.execute(
+            update(ApiToken)
+            .where(ApiToken.id == token_id)
+            .values(last_used_at=datetime.now(UTC))
+        )
+        self.__db.commit()

backend/app/api_token/schema.py

@@ -0,0 +1,26 @@
+import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from app.base.schema import Identified
+
+
+class ApiTokenCreateRequest(BaseModel):
+    name: str
+    user_id: int
+    expires_at: datetime.datetime | None = None
+
+
+class ApiTokenResponse(Identified):
+    name: str
+    user_id: int
+    created_by: int
+    created_at: datetime.datetime
+    expires_at: datetime.datetime | None
+    last_used_at: datetime.datetime | None
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class ApiTokenCreatedResponse(ApiTokenResponse):
+    token: str

backend/app/routers/api_tokens.py

@@ -0,0 +1,52 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.api_token.schema import (
+    ApiTokenCreatedResponse,
+    ApiTokenCreateRequest,
+    ApiTokenResponse,
+)
+from app.db import get_db
+from app.models import StatusMessage
+from app.permissions import P, PermissionChecker
+from app.services.api_token_service import ApiTokenService
+from app.user.depends import get_current_user_id
+
+router = APIRouter(
+    prefix="/api_tokens",
+    tags=["apitokens"],
+    dependencies=[Depends(PermissionChecker(P.USER_MANAGE))],
+)
+
+
+def get_service(db: Annotated[Session, Depends(get_db)]) -> ApiTokenService:
+    return ApiTokenService(db)
+
+
+@router.post("/")
+def create_token(
+    data: ApiTokenCreateRequest,
+    current_user: Annotated[int, Depends(get_current_user_id)],
+    service: Annotated[ApiTokenService, Depends(get_service)],
+) -> ApiTokenCreatedResponse:
+    return service.create_token(data, current_user)
+
+
+@router.get("/")
+def list_tokens(
+    service: Annotated[ApiTokenService, Depends(get_service)],
+) -> list[ApiTokenResponse]:
+    return service.list_tokens()
+
+
+@router.delete("/{token_id}")
+def delete_token(
+    token_id: int,
+    service: Annotated[ApiTokenService, Depends(get_service)],
+) -> StatusMessage:
+    try:
+        return service.delete_token(token_id)
+    except Exception as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

backend/app/services/__init__.py

@@ -1,5 +1,6 @@
 """Service layer module for business logic operations."""
 
+from app.services.api_token_service import ApiTokenService
 from app.services.auth_service import AuthService
 from app.services.comment_service import CommentService
 from app.services.document_service import DocumentService
@@ -9,6 +10,7 @@
 from app.services.user_service import UserService
 
 __all__ = [
+    "ApiTokenService",
     "AuthService",
     "CommentService",
     "DocumentService",

backend/app/services/api_token_service.py

@@ -0,0 +1,52 @@
+import hashlib
+import secrets
+
+from sqlalchemy.orm import Session
+
+from app.api_token.query import ApiTokenQuery
+from app.api_token.schema import (
+    ApiTokenCreatedResponse,
+    ApiTokenCreateRequest,
+    ApiTokenResponse,
+)
+from app.base.exceptions import EntityNotFound
+from app.models import StatusMessage
+
+
+class ApiTokenService:
+    def __init__(self, db: Session) -> None:
+        self.__query = ApiTokenQuery(db)
+
+    def create_token(
+        self, data: ApiTokenCreateRequest, created_by: int
+    ) -> ApiTokenCreatedResponse:
+        raw_token = f"hat_{secrets.token_hex(32)}"
+        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+        token = self.__query.create(
+            name=data.name,
+            token_hash=token_hash,
+            user_id=data.user_id,
+            created_by=created_by,
+            expires_at=data.expires_at,
+        )
+        return ApiTokenCreatedResponse(
+            id=token.id,
+            name=token.name,
+            user_id=token.user_id,
+            created_by=token.created_by,
+            created_at=token.created_at,
+            expires_at=token.expires_at,
+            last_used_at=token.last_used_at,
+            token=raw_token,
+        )
+
+    def list_tokens(self) -> list[ApiTokenResponse]:
+        tokens = self.__query.get_all()
+        return [ApiTokenResponse.model_validate(t) for t in tokens]
+
+    def delete_token(self, token_id: int) -> StatusMessage:
+        try:
+            self.__query.delete(token_id)
+        except Exception:
+            raise EntityNotFound("Api Token", token_id)
+        return StatusMessage(message="ok")

backend/app/user/depends.py

@@ -1,20 +1,40 @@
+import hashlib
+from datetime import UTC, datetime
 from typing import Annotated
 
-from fastapi import Cookie, HTTPException, status
+from fastapi import Cookie, Depends, Header, HTTPException, status
 from itsdangerous import URLSafeTimedSerializer
+from sqlalchemy.orm import Session
 
+from app.api_token.query import ApiTokenQuery
+from app.db import get_db
 from app.settings import settings
 
 
 def get_current_user_id(
+    db: Annotated[Session, Depends(get_db)],
     session: Annotated[str | None, Cookie(include_in_schema=False)] = None,
+    authorization: Annotated[str | None, Header(include_in_schema=False)] = None,
 ) -> int:
-    if not session:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+    if session:
+        serializer = URLSafeTimedSerializer(secret_key=settings.secret_key)
+        data = serializer.loads(session)
+        if "user_id" in data:
+            return data["user_id"]
 
-    serializer = URLSafeTimedSerializer(secret_key=settings.secret_key)
-    data = serializer.loads(session)
-    if "user_id" in data:
-        return data["user_id"]
+    if authorization and authorization.startswith("Bearer "):
+        query = ApiTokenQuery(db)
+        raw_token = authorization[7:]
+        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
+        api_token = query.get_by_hash(token_hash)
+        if api_token:
+            if api_token.expires_at:
+                exp = api_token.expires_at
+                if exp.tzinfo is None:
+                    exp = exp.replace(tzinfo=UTC)
+                if exp < datetime.now(UTC):
+                    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+            query.update_last_used(api_token.id)
+            return api_token.user_id
 
     raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)