Spaces:
Sleeping
Sleeping
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() | |
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 | |
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) | |
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 | |
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) | |