import secrets from datetime import datetime, timedelta from typing import Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlmodel import Session, select from app.auth import get_current_user, hash_token from app.db import get_session from app.models.models import APIToken, APITokenCreate, APITokenRead router = APIRouter(prefix="/tokens", tags=["tokens"]) class TokenCreatedResponse(BaseModel): token: str name: str expires_at: Optional[datetime] @router.post("/", response_model=TokenCreatedResponse, status_code=201) def create_token( data: APITokenCreate, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): plaintext = secrets.token_urlsafe(32) expires_at = None if data.expires_days: expires_at = datetime.utcnow() + timedelta(days=data.expires_days) api_token = APIToken( name=data.name, token_hash=hash_token(plaintext), expires_at=expires_at, ) session.add(api_token) session.commit() session.refresh(api_token) return TokenCreatedResponse(token=plaintext, name=api_token.name, expires_at=api_token.expires_at) @router.get("/", response_model=list[APITokenRead]) def list_tokens( session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): return list(session.exec( select(APIToken).where(APIToken.is_active == True).order_by(APIToken.created_at.desc()) ).all()) @router.delete("/{token_id}", status_code=204) def revoke_token( token_id: int, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): token = session.get(APIToken, token_id) if not token: raise HTTPException(status_code=404, detail="Token not found") token.is_active = False session.add(token) session.commit()