● Zion Boggan
repos/Sightline/backend/app/api/routes/apikeys.py
zionboggan.com ↗
56 lines · python
History for this file →
1
from __future__ import annotations
2
 
3
from datetime import datetime, timezone
4
 
5
from fastapi import APIRouter, Depends, HTTPException
6
from pydantic import BaseModel
7
from sqlalchemy import select
8
from sqlalchemy.orm import Session
9
 
10
from app.api.deps import Perm, get_current_user, get_db, require
11
from app.core.security import generate_api_key
12
from app.models import ApiKey, AuditLog, User
13
 
14
router = APIRouter(prefix="/api-keys", tags=["api-keys"])
15
 
16
 
17
class ApiKeyCreate(BaseModel):
18
    name: str
19
    scopes: list[str] = []
20
 
21
 
22
@router.post("", status_code=201)
23
def create_key(payload: ApiKeyCreate, db: Session = Depends(get_db),
24
               user: User = Depends(get_current_user)):
25
    raw, key_hash = generate_api_key()
26
    key = ApiKey(user_id=user.id, name=payload.name, key_hash=key_hash,
27
                 scopes={"scopes": payload.scopes})
28
    db.add(key)
29
    db.flush()
30
    db.add(AuditLog(actor_user_id=user.id, action="apikey.create",
31
                    entity_type="api_key", entity_id=key.id, meta={"name": payload.name}))
32
    db.commit()
33
    db.refresh(key)
34
    return {"id": key.id, "name": key.name, "api_key": raw,
35
            "note": "Store this now - it will not be shown again."}
36
 
37
 
38
@router.get("")
39
def list_keys(db: Session = Depends(get_db), user: User = Depends(get_current_user)):
40
    rows = db.scalars(select(ApiKey).where(ApiKey.user_id == user.id)
41
                      .order_by(ApiKey.created_at.desc())).all()
42
    return [{"id": k.id, "name": k.name, "scopes": (k.scopes or {}).get("scopes", []),
43
             "last_used_at": k.last_used_at, "revoked_at": k.revoked_at,
44
             "created_at": k.created_at} for k in rows]
45
 
46
 
47
@router.delete("/{key_id}")
48
def revoke_key(key_id: str, db: Session = Depends(get_db), user: User = Depends(get_current_user)):
49
    key = db.get(ApiKey, key_id)
50
    if not key or key.user_id != user.id:
51
        raise HTTPException(404, "api key not found")
52
    key.revoked_at = datetime.now(timezone.utc)
53
    db.add(AuditLog(actor_user_id=user.id, action="apikey.revoke",
54
                    entity_type="api_key", entity_id=key.id))
55
    db.commit()
56
    return {"id": key_id, "revoked": True}