Add mypy

Author: diofeher

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
 -   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.7.1
+    rev: v0.7.2
     hooks:
     # Run the linter.
     - id: ruff
@@ -9,3 +9,8 @@ repos:
     # Run the formatter.
     - id: ruff-format
       types_or: [ python, pyi ]
+-   repo: https://github.com/pre-commit/mirrors-mypy
+    rev: 'v1.13.0'  # Use the sha / tag you want to point at
+    hooks:
+    -   id: mypy
+        args: [--strict, --ignore-missing-imports]

backend/app/__init__.py

@@ -1,3 +1,3 @@
 from .app import app
 
-__all__ = [app]
+__all__ = ["app"]

backend/app/app.py

@@ -1,3 +1,4 @@
+from typing import AsyncGenerator
 from .tasks.routers import router as task_router
 from .users.routers import router as user_router
 from fastapi import FastAPI
@@ -9,7 +10,7 @@
 
 
 @asynccontextmanager
-async def lifespan(app: FastAPI):
+async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
     # Load the ML model
     create_db_and_tables()
     yield

backend/app/auth.py

@@ -1,11 +1,13 @@
 import jwt
-from typing import Annotated
+from typing import Dict, Annotated, cast, Any
 from passlib.hash import pbkdf2_sha256
 from jwt.exceptions import InvalidTokenError
+
 from datetime import datetime, timedelta, timezone
 
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.sql import ColumnElement
 from sqlalchemy.orm import Session
 from sqlalchemy import select
 
@@ -20,26 +22,28 @@
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/token")
 
 
-def create_access_token(data: dict, expires_delta: timedelta | None = None):
+def create_access_token(
+    data: Dict[str, str | datetime], expires_delta: timedelta | None = None
+) -> str:
     to_encode = data.copy()
     if expires_delta:
         expire = datetime.now(timezone.utc) + expires_delta
     else:
         expire = datetime.now(timezone.utc) + timedelta(minutes=15)
     to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    encoded_jwt = str(jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM))
     return encoded_jwt
 
 
 def hash_password(password: str) -> str:
-    return pbkdf2_sha256.hash(password)
+    return str(pbkdf2_sha256.hash(password))
 
 
 async def get_current_user(
     *,
     session: Session = Depends(get_session),
     token: Annotated[str, Depends(oauth2_scheme)],
-):
+) -> Any:
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
         detail="Could not validate credentials",
@@ -54,16 +58,18 @@ async def get_current_user(
     except InvalidTokenError:
         raise credentials_exception
 
-    stmt = select(User).where(User.username == token_data.username)
-    db_user = session.exec(stmt).first()[0]
+    stmt = select(User).where(
+        cast("ColumnElement[bool]", User.username == token_data.username)
+    )
+    db_user = session.execute(stmt).one()
     if not db_user:
         raise credentials_exception
     return db_user
 
 
 async def get_current_active_user(
     current_user: Annotated[User, Depends(get_current_user)],
-):
+) -> User:
     # if current_user.disabled:
     # raise HTTPException(status_code=400, detail="Inactive user")
     return current_user

backend/app/db.py

@@ -1,23 +1,23 @@
 from sqlalchemy_utils import database_exists, create_database
-from sqlmodel import Session, create_engine, SQLModel
+from sqlmodel import Session, create_engine, SQLModel, Engine
 from .settings import settings
 
 
-def get_engine(url=settings.database_url):
+def get_engine(url: str = settings.database_url) -> Engine:
     engine = create_engine(url)
     if not database_exists(engine.url):
         create_database(engine.url)
     return engine
 
 
-def get_session():
+def get_session() -> Session:
     with Session(get_engine()) as session:
         yield session
 
 
-def create_db_and_tables(engine=get_engine()):
+def create_db_and_tables(engine: Engine = get_engine()) -> None:
     SQLModel.metadata.create_all(engine)
 
 
-def drop_db_and_tables(engine=get_engine()):
+def drop_db_and_tables(engine: Engine = get_engine()) -> None:
     SQLModel.metadata.drop_all(engine)

backend/app/settings.py

@@ -1,7 +1,7 @@
 from pydantic_settings import BaseSettings
 
 
-class Settings(BaseSettings):
+class Settings(BaseSettings):  # type: ignore
     # TODO: Remove this hardcoded default from here
     database_url: str = (
         # "postgresql+psycopg2://myuser:mypassword@db:5432/mydatabase"

backend/app/tasks/models.py

@@ -11,9 +11,9 @@ class TaskStatus(str, enum.Enum):
     created = "created"
 
 
-class TaskBase(SQLModel):
+class TaskBase(SQLModel):  # type: ignore
     title: str
-    description: str
+    description: str | None
     due_date: datetime | None = Field(nullable=True, default=None)
     status: TaskStatus = Field(
         sa_column=Column(Enum(TaskStatus)), default=TaskStatus.created
@@ -22,6 +22,6 @@ class TaskBase(SQLModel):
 
 
 class Task(TaskBase, table=True):
-    model_config = ConfigDict(validate_assignment=True)
+    model_config = ConfigDict(validate_assignment=True)  # type: ignore
 
     id: int | None = Field(default=None, primary_key=True, index=True)

backend/app/tasks/repository.py

@@ -1,4 +1,5 @@
-from typing import List, Optional
+from typing import List, cast
+from sqlalchemy.sql import ColumnElement
 from sqlalchemy.orm import Session
 from sqlalchemy import select, update
 from dataclasses import dataclass
@@ -9,7 +10,7 @@
 @dataclass
 class TaskRepository:
     session: Session
-    current_user_id: int
+    current_user_id: int | None
 
     def create(self, data: TaskInput) -> TaskOutput:
         task = Task(**data.model_dump(exclude_none=True))
@@ -18,25 +19,32 @@ def create(self, data: TaskInput) -> TaskOutput:
         self.session.refresh(task)
         return TaskOutput(**dict(task))
 
-    def list(self, offset: int, limit: int) -> List[Optional[TaskOutput]]:
+    def list(self, offset: int, limit: int) -> List[TaskOutput]:
+        # import pdb; pdb.set_trace()
         stmt = (
             select(Task)
             .where(
-                Task.status != TaskStatus.deleted,
-                Task.user_id == self.current_user_id,
+                cast("ColumnElement[bool]", Task.status != TaskStatus.deleted)
+            )
+            .where(
+                cast(
+                    "ColumnElement[bool]", Task.user_id == self.current_user_id
+                )
             )
             .offset(offset)
             .limit(limit)
         )
         tasks = self.session.execute(stmt).all()
         return [TaskOutput(**dict(task[0])) for task in tasks]
 
-    def get_by_id(self, id: int) -> Task:
-        return self.session.get(Task, id)
+    def get_by_id(self, id: int) -> Task | None:
+        return self.session.get(Task, id)  # type: ignore
 
     def update(self, task: Task) -> TaskOutput:
         self.session.execute(
-            update(Task).where(Task.id == task.id).values(**dict(task))
+            update(Task)
+            .where(cast("ColumnElement[bool]", Task.id == task.id))
+            .values(**dict(task))
         )
         self.session.commit()
         self.session.refresh(task)

backend/app/tasks/routers.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import List, Annotated
 from fastapi import Depends, Query, APIRouter
 from sqlalchemy.orm import Session
 from ..users.models import User
@@ -11,49 +11,49 @@
 router = APIRouter(prefix="/tasks")
 
 
-@router.get("/")
+@router.get("/")  # type: ignore
 def read_tasks(
     *,
     current_user: Annotated[User, Depends(get_current_active_user)],
     session: Session = Depends(get_session),
     offset: int = 0,
     limit: int = Query(default=100, le=100),
-):
+) -> List[TaskOutput]:
     return TaskService(session, current_user.id).list(offset, limit)
 
 
-@router.post("/", response_model=TaskOutput)
+@router.post("/", response_model=TaskOutput)  # type: ignore
 def create_task(
     *,
     current_user: Annotated[User, Depends(get_current_active_user)],
     session: Session = Depends(get_session),
     data: TaskInput,
-):
+) -> TaskOutput:
     return TaskService(session, current_user.id).create(data)
 
 
-@router.get("/{task_id}")
+@router.get("/{task_id}")  # type: ignore
 def read_task(
     *,
     session: Session = Depends(get_session),
     current_user: Annotated[User, Depends(get_current_active_user)],
     task_id: int,
-):
+) -> TaskOutput:
     return TaskService(session, current_user.id).read(task_id)
 
 
-@router.patch("/{task_id}")
+@router.patch("/{task_id}")  # type: ignore
 def update_task(
     *,
     session: Session = Depends(get_session),
     current_user: Annotated[User, Depends(get_current_active_user)],
     task_id: int,
     task: TaskInput,
-):
+) -> TaskOutput:
     return TaskService(session, current_user.id).update(task_id, task)
 
 
-@router.delete("/{task_id}")
+@router.delete("/{task_id}")  # type: ignore
 def delete_task(
     *,
     current_user: Annotated[User, Depends(get_current_active_user)],

backend/app/tasks/schema.py

@@ -4,15 +4,15 @@
 from .models import TaskStatus
 
 
-class TaskInput(BaseModel):
+class TaskInput(BaseModel):  # type: ignore
     title: str = Field(min_length=1, max_length=50)
     description: str | None = None
     user_id: int | None = None
     due_date: datetime.datetime | None = None
     status: TaskStatus = TaskStatus.created
 
 
-class TaskOutput(BaseModel):
+class TaskOutput(BaseModel):  # type: ignore
     id: int
     title: str
     description: Optional[str] = ""

backend/app/tasks/service.py

@@ -10,9 +10,9 @@
 @dataclass
 class TaskService:
     session: Session
-    current_user_id: int
+    current_user_id: int | None
 
-    def __post_init__(self):
+    def __post_init__(self) -> None:
         self.repository: TaskRepository = TaskRepository(
             self.session, self.current_user_id
         )
@@ -49,4 +49,4 @@ def delete(self, task_id: int) -> TaskOutput:
 
         task.status = TaskStatus.deleted
         self.repository.update(task)
-        return task
+        return TaskOutput(**dict(task))

backend/app/users/models.py

@@ -3,7 +3,7 @@
 from pydantic import BaseModel, ConfigDict
 
 
-class User(SQLModel, table=True):
+class User(SQLModel, table=True):  # type: ignore
     id: int | None = Field(
         default=None,
         sa_column=Column(
@@ -14,17 +14,17 @@ class User(SQLModel, table=True):
     hashed_password: str
 
 
-class UserCreate(BaseModel):
+class UserCreate(BaseModel):  # type: ignore
     model_config = ConfigDict(validate_assignment=True)
 
     username: str
     password: str
 
 
-class Token(BaseModel):
+class Token(BaseModel):  # type: ignore
     access_token: str
     token_type: str
 
 
-class TokenData(BaseModel):
+class TokenData(BaseModel):  # type: ignore
     username: str | None = None

backend/app/users/routers.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, Dict
 from datetime import timedelta
 
 from sqlmodel import Session, select
@@ -20,12 +20,12 @@
 router = APIRouter(prefix="/users")
 
 
-@router.post("/token")
+@router.post("/token")  # type: ignore
 async def login(
     *,
     session: Session = Depends(get_session),
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
-):
+) -> Dict[str, str]:
     stmt = select(User).where(User.username == form_data.username)
     db_user = session.exec(stmt).first()
     if not db_user or (
@@ -47,14 +47,14 @@ async def login(
     }
 
 
-@router.get("/me")
+@router.get("/me")  # type: ignore
 async def read_users_me(
     current_user: Annotated[User, Depends(get_current_active_user)],
-):
+) -> User:
     return current_user
 
 
-@router.post("/")
+@router.post("/")  # type: ignore
 def create_user(*, session: Session = Depends(get_session), user: UserCreate):
     stmt = select(User).where(User.username == user.username)
     db_user = session.exec(stmt).first()