Sun Jun 8 16:41:37 CST 2025

app/database.py

@@ -38,7 +38,7 @@ class KeyCategory(Base):
     name = Column(String, unique=True, index=True, nullable=False)
     type = Column(String, nullable=False)
     tags = Column(JSONText) # Stored as JSON string
-    metadata_ = Column(JSONText, name="metadata") # Use metadata_ to avoid conflict with SQLAlchemy metadata
+    metadata_ = Column(JSONText, name="metadata") # Map to 'metadata' column in DB
 
     api_keys = relationship("APIKey", back_populates="category")
 
@@ -55,7 +55,7 @@ class APIKey(Base):
     status = Column(String, default="active", nullable=False) # 'active', 'inactive'
     usage_count = Column(Integer, default=0, nullable=False)
     last_used = Column(DateTime)
-    metadata_ = Column(JSONText, name="metadata") # Use metadata_
+    metadata_ = Column(JSONText, name="metadata") # Map to 'metadata' column in DB
 
     category = relationship("KeyCategory", back_populates="api_keys")
 

app/routers/key_management.py

@@ -0,0 +1,74 @@
+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)

app/schemas.py

@@ -0,0 +1,74 @@
+from typing import List, Optional, Dict, Any
+from pydantic import BaseModel, Field # Import Field
+from datetime import datetime # Import datetime
+
+# Pydantic models for reading data from the database
+
+class KeyCategoryBase(BaseModel):
+    name: str
+    type: str
+    tags: Optional[List[str]] = None
+    # metadata: Optional[Dict[str, Any]] = None # Removed from base
+
+class KeyCategory(KeyCategoryBase):
+    id: int
+    # Map metadata_ from SQLAlchemy model to metadata in schema
+    metadata: Optional[Dict[str, Any]] = Field(None, alias="metadata_")
+    created_at: Optional[datetime] = None # Use datetime type
+    updated_at: Optional[datetime] = None # Use datetime type
+
+    class Config:
+        orm_mode = True # Enable ORM mode
+        by_alias = True # Use aliases for serialization
+        # Configure JSON encoding for datetime objects if needed (orm_mode might handle this)
+        # json_encoders = {
+        #     datetime: lambda v: v.isoformat() if v else None
+        # }
+
+
+class APIKeyBase(BaseModel):
+    value: str
+    category_id: int
+    status: str = "active"
+    # metadata: Optional[Dict[str, Any]] = None # Removed from base
+
+class APIKey(APIKeyBase):
+    id: int
+    usage_count: int
+    last_used: Optional[datetime] = None # Use datetime type
+    # Map metadata_ from SQLAlchemy model to metadata in schema
+    metadata: Optional[Dict[str, Any]] = Field(None, alias="metadata_")
+    created_at: Optional[datetime] = None # Use datetime type
+    updated_at: Optional[datetime] = None # Use datetime type
+
+    class Config:
+        orm_mode = True # Enable ORM mode
+        by_alias = True # Use aliases for serialization
+        # Configure JSON encoding for datetime objects if needed
+        # json_encoders = {
+        #     datetime: lambda v: v.isoformat() if v else None
+        # }
+
+
+# Pydantic models for creating data (less fields, e.g., no ID, timestamps)
+
+class KeyCategoryCreate(KeyCategoryBase):
+    metadata: Optional[Dict[str, Any]] = None # Add metadata back to create schema
+
+class APIKeyCreate(APIKeyBase):
+    metadata: Optional[Dict[str, Any]] = None # Add metadata back to create schema
+
+
+# Pydantic models for updating data (all fields optional)
+
+class KeyCategoryUpdate(BaseModel):
+    name: Optional[str] = None
+    type: Optional[str] = None
+    tags: Optional[List[str]] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+class APIKeyUpdate(BaseModel):
+    value: Optional[str] = None
+    category_id: Optional[int] = None
+    status: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None

main.py

@@ -6,6 +6,8 @@ import os, json
 import re  # 用于URL路径处理
 from key_selector import KeySelector # 自动选择key
 
+from app.routers import key_management # Import the new router
+
 def get_target_url(url: str) -> str:
     """将url参数变了转换为合法的目标url；from http/ or https/ to http:// or https://"""
     url = re.sub(r"^http/", "http://", url)
@@ -14,6 +16,9 @@ def get_target_url(url: str) -> str:
 
 app = FastAPI()
 
+# Include the new key management router
+app.include_router(key_management.router, prefix="/api/keys", tags=["Key Management"])
+
 # 配置日志
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("uvicorn.error")

sql/schema.sql

@@ -0,0 +1,3 @@
+CREATE TABLE key_categories ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT UNIQUE NOT NULL, type TEXT NOT NULL, tags TEXT, metadata TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL );
+CREATE TABLE sqlite_sequence(name,seq);
+CREATE TABLE api_keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT NOT NULL, category_id INTEGER NOT NULL, status TEXT DEFAULT 'active' NOT NULL, usage_count INTEGER DEFAULT 0 NOT NULL, last_used DATETIME, metadata TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL, FOREIGN KEY (category_id) REFERENCES key_categories (id) ON DELETE CASCADE );