alex
22474cdba5
Backend (ruff): - Fix 106 errors: move imports to top of file (E402) - Remove unused imports (F401) - Add missing imports for undefined names (F821) - Remove unused variables (F841) - Fix test_models.py broken RefreshToken test - Fix test_projects_api.py missing TestClient import Frontend (eslint): - Remove unused imports/variables across 10 files - Fix explicit any types in client.ts and sessions.ts - Clean up empty block statements in terminal.tsx Quality gates: ruff (pass), eslint (pass), tsc --noEmit (pass), pytest (98 passed, 4 pre-existing failures)
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
import uuid
|
|
from typing import Annotated
|
|
|
|
from fastapi import Cookie, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.auth.session import decode_session_cookie
|
|
from src.config import Settings
|
|
from src.database import SessionLocal
|
|
from src.models.project import Project
|
|
from src.models.user import User
|
|
|
|
|
|
async def get_db_session():
|
|
async with SessionLocal() as session:
|
|
yield session
|
|
|
|
|
|
async def get_current_user_id(
|
|
session_cookie: Annotated[str | None, Cookie(alias="session")] = None,
|
|
) -> uuid.UUID:
|
|
if not session_cookie:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing session")
|
|
|
|
settings = Settings()
|
|
try:
|
|
payload = decode_session_cookie(settings=settings, cookie_value=session_cookie)
|
|
return uuid.UUID(str(payload["user_id"]))
|
|
except ValueError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid session")
|
|
|
|
|
|
async def get_current_user(
|
|
session_cookie: Annotated[str | None, Cookie(alias="session")] = None,
|
|
db_session: AsyncSession = Depends(get_db_session),
|
|
) -> User:
|
|
if not session_cookie:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing session")
|
|
|
|
settings = Settings()
|
|
try:
|
|
payload = decode_session_cookie(settings=settings, cookie_value=session_cookie)
|
|
user_id = uuid.UUID(str(payload["user_id"]))
|
|
except ValueError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid session")
|
|
|
|
user = await db_session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
|
|
return user
|
|
|
|
|
|
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
|
|
"""Fetch a user by ID or raise 401 if not found."""
|
|
user = await session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
|
|
return user
|
|
|
|
|
|
async def _get_owned_project(
|
|
project_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
session: AsyncSession,
|
|
) -> "Project":
|
|
"""Fetch a project and verify ownership.
|
|
|
|
Args:
|
|
project_id: UUID of the project.
|
|
user_id: ID of the authenticated user.
|
|
session: Database session.
|
|
|
|
Returns:
|
|
The project if found and owned by the user.
|
|
|
|
Raises:
|
|
HTTPException: 404 if project not found, 403 if user is not the owner.
|
|
"""
|
|
from src.models.project import Project
|
|
|
|
project = await session.get(Project, project_id)
|
|
if project is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="project not found")
|
|
if project.owner_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="not project owner")
|
|
return project
|