ref(backend): Centralize deps and add mypy

Extract reusable API dependencies and activity logging helpers.\nAdd mypy configuration and dev dependency for type checking.\n\nCo-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Abhimanyu Saharan
2026-02-04 03:57:19 +05:30
parent 7fd079e4f1
commit b24e3e1dcd
12 changed files with 205 additions and 189 deletions

View File

@@ -2,45 +2,36 @@ from __future__ import annotations
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends
from sqlmodel import Session, select
from app.core.auth import get_auth_context
from app.api.deps import get_board_or_404, get_task_or_404, require_admin_auth
from app.core.auth import AuthContext
from app.db.session import get_session
from app.models.activity_events import ActivityEvent
from app.models.boards import Board
from app.models.tasks import Task
from app.schemas.tasks import TaskCreate, TaskRead, TaskUpdate
from app.services.admin_access import require_admin
from app.services.activity_log import record_activity
router = APIRouter(prefix="/boards/{board_id}/tasks", tags=["tasks"])
@router.get("", response_model=list[TaskRead])
def list_tasks(
board_id: str,
board: Board = Depends(get_board_or_404),
session: Session = Depends(get_session),
auth=Depends(get_auth_context),
auth: AuthContext = Depends(require_admin_auth),
) -> list[Task]:
require_admin(auth)
board = session.get(Board, board_id)
if board is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return list(session.exec(select(Task).where(Task.board_id == board.id)))
@router.post("", response_model=TaskRead)
def create_task(
board_id: str,
payload: TaskCreate,
board: Board = Depends(get_board_or_404),
session: Session = Depends(get_session),
auth=Depends(get_auth_context),
auth: AuthContext = Depends(require_admin_auth),
) -> Task:
require_admin(auth)
board = session.get(Board, board_id)
if board is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
task = Task.model_validate(payload)
task.board_id = board.id
if task.created_by_user_id is None and auth.user is not None:
@@ -49,32 +40,23 @@ def create_task(
session.commit()
session.refresh(task)
event = ActivityEvent(
record_activity(
session,
event_type="task.created",
task_id=task.id,
message=f"Task created: {task.title}.",
)
session.add(event)
session.commit()
return task
@router.patch("/{task_id}", response_model=TaskRead)
def update_task(
board_id: str,
task_id: str,
payload: TaskUpdate,
task: Task = Depends(get_task_or_404),
session: Session = Depends(get_session),
auth=Depends(get_auth_context),
auth: AuthContext = Depends(require_admin_auth),
) -> Task:
require_admin(auth)
board = session.get(Board, board_id)
if board is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
task = session.get(Task, task_id)
if task is None or task.board_id != board.id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
previous_status = task.status
updates = payload.model_dump(exclude_unset=True)
for key, value in updates.items():
@@ -91,25 +73,17 @@ def update_task(
else:
event_type = "task.updated"
message = f"Task updated: {task.title}."
event = ActivityEvent(event_type=event_type, task_id=task.id, message=message)
session.add(event)
record_activity(session, event_type=event_type, task_id=task.id, message=message)
session.commit()
return task
@router.delete("/{task_id}")
def delete_task(
board_id: str,
task_id: str,
session: Session = Depends(get_session),
auth=Depends(get_auth_context),
task: Task = Depends(get_task_or_404),
auth: AuthContext = Depends(require_admin_auth),
) -> dict[str, bool]:
require_admin(auth)
board = session.get(Board, board_id)
if board is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
task = session.get(Task, task_id)
if task and task.board_id == board.id:
session.delete(task)
session.commit()
session.delete(task)
session.commit()
return {"ok": True}