fastapi-proxy-test / app /routers /key_management.py
tanbushi's picture
Sun Jun 8 16:41:37 CST 2025
7183ec8
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List, Optional # Import Optional
from .. import crud, schemas # Use relative imports
from ..database import get_db # Use relative import
router = APIRouter()
@router.get("/categories/", response_model=List[schemas.KeyCategory])
def read_key_categories(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=0, le=100),
db: Session = Depends(get_db)
):
"""
Retrieve a list of key categories.
"""
categories = crud.get_key_categories(db, skip=skip, limit=limit)
return categories
@router.post("/categories/", response_model=schemas.KeyCategory)
def create_key_category(
category: schemas.KeyCategoryCreate,
db: Session = Depends(get_db)
):
"""
Create a new key category.
"""
# Check if category with the same name already exists
db_category = crud.get_key_category_by_name(db, name=category.name)
if db_category:
raise HTTPException(status_code=400, detail="Key Category with this name already exists")
# Create the category using the CRUD function
return crud.create_key_category(db=db, name=category.name, type=category.type, tags=category.tags, metadata=category.metadata)
@router.get("/instances/", response_model=List[schemas.APIKey])
def read_api_keys(
category_id: Optional[int] = Query(None, ge=1), # Optional filter by category ID
status: Optional[str] = Query(None, pattern="^(active|inactive)$"), # Optional filter by status
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=0, le=100),
db: Session = Depends(get_db)
):
"""
Retrieve a list of API key instances, optionally filtered by category or status.
"""
keys = crud.get_api_keys(db, category_id=category_id, status=status, skip=skip, limit=limit)
return keys
@router.post("/instances/", response_model=schemas.APIKey)
def create_api_key(
api_key: schemas.APIKeyCreate,
db: Session = Depends(get_db)
):
"""
Create a new API key instance under a specific category.
"""
# Check if the category_id exists
category = crud.get_key_category(db, category_id=api_key.category_id)
if not category:
raise HTTPException(status_code=404, detail=f"Key Category with ID {api_key.category_id} not found")
# Create the API key instance using the CRUD function
return crud.create_api_key(
db=db,
value=api_key.value,
category_id=api_key.category_id,
status=api_key.status,
metadata=api_key.metadata
)
# TODO: Add other CRUD endpoints for key categories and API keys (GET by ID, PUT, DELETE)