Sun Jun 8 15:02:12 CST 2025

.gitignore

@@ -1,3 +1,4 @@
 **/__pycache__/
 README.md
-.env
+.env
+api_proxy.db

Dockerfile

@@ -13,4 +13,4 @@ COPY --chown=user ./requirements.txt requirements.txt
 RUN pip install --no-cache-dir --upgrade -r requirements.txt
 
 COPY --chown=user . /app
-CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

app/crud.py

@@ -0,0 +1,114 @@
+from sqlalchemy.orm import Session
+from .database import Base, KeyCategory, APIKey # Import models directly from database module
+# from . import schemas # Schemas will be created later
+
+# --- CRUD for KeyCategory ---
+
+def create_key_category(db: Session, name: str, type: str, tags: list = None, metadata: dict = None):
+    db_category = KeyCategory(name=name, type=type, tags=tags, metadata_=metadata)
+    db.add(db_category)
+    db.commit()
+    db.refresh(db_category)
+    return db_category
+
+def get_key_category(db: Session, category_id: int):
+    return db.query(KeyCategory).filter(KeyCategory.id == category_id).first()
+
+def get_key_category_by_name(db: Session, name: str):
+    return db.query(KeyCategory).filter(KeyCategory.name == name).first()
+
+def get_key_categories(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(KeyCategory).offset(skip).limit(limit).all()
+
+def update_key_category(db: Session, category_id: int, name: str = None, type: str = None, tags: list = None, metadata: dict = None):
+    db_category = get_key_category(db, category_id)
+    if db_category:
+        if name is not None:
+            db_category.name = name
+        if type is not None:
+            db_category.type = type
+        if tags is not None:
+            db_category.tags = tags
+        if metadata is not None:
+            db_category.metadata_ = metadata
+        db.commit()
+        db.refresh(db_category)
+    return db_category
+
+def delete_key_category(db: Session, category_id: int):
+    db_category = get_key_category(db, category_id)
+    if db_category:
+        db.delete(db_category)
+        db.commit()
+    return db_category
+
+# --- CRUD for APIKey ---
+
+def create_api_key(db: Session, value: str, category_id: int, status: str = "active", metadata: dict = None):
+    db_key = APIKey(value=value, category_id=category_id, status=status, metadata_=metadata)
+    db.add(db_key)
+    db.commit()
+    db.refresh(db_key)
+    return db_key
+
+def get_api_key(db: Session, key_id: int):
+    return db.query(APIKey).filter(APIKey.id == key_id).first()
+
+def get_api_keys(db: Session, category_id: int = None, status: str = None, skip: int = 0, limit: int = 100):
+    query = db.query(APIKey)
+    if category_id is not None:
+        query = query.filter(APIKey.category_id == category_id)
+    if status is not None:
+        query = query.filter(APIKey.status == status)
+    return query.offset(skip).limit(limit).all()
+
+def update_api_key(db: Session, key_id: int, value: str = None, category_id: int = None, status: str = None, metadata: dict = None):
+    db_key = get_api_key(db, key_id)
+    if db_key:
+        if value is not None:
+            db_key.value = value
+        if category_id is not None:
+            db_key.category_id = category_id
+        if status is not None:
+            db_key.status = status
+        if metadata is not None:
+            db_key.metadata_ = metadata
+        db.commit()
+        db.refresh(db_key)
+    return db_key
+
+def delete_api_key(db: Session, key_id: int):
+    db_key = get_api_key(db, key_id)
+    if db_key:
+        db.delete(db_key)
+        db.commit()
+    return db_key
+
+# --- Key Selection Logic (Basic Placeholder) ---
+# This will be expanded later based on selection strategy (round-robin, etc.)
+
+def get_available_keys_for_category(db: Session, category_id: int):
+    """Get all active keys for a given category."""
+    return db.query(APIKey).filter(
+        APIKey.category_id == category_id,
+        APIKey.status == "active"
+    ).all()
+
+# Placeholder for selection strategy - will need to implement round-robin, etc.
+def select_key_from_pool(db: Session, category_id: int):
+    """Select one key from the available pool for a category."""
+    available_keys = get_available_keys_for_category(db, category_id)
+    if not available_keys:
+        return None # No keys available
+
+    # Basic selection: just return the first active key for now
+    # TODO: Implement round-robin or other selection strategy
+    selected_key = available_keys[0]
+
+    # Optional: Update usage stats (e.g., increment usage_count, update last_used)
+    # selected_key.usage_count += 1
+    # selected_key.last_used = func.now()
+    # db.commit()
+    # db.refresh(selected_key)
+
+    return selected_key

app/database.py

@@ -0,0 +1,75 @@
+import json
+from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey, func
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker, relationship
+from sqlalchemy.types import TypeDecorator, TEXT
+
+# Custom type for JSON storage
+class JSONText(TypeDecorator):
+    impl = TEXT
+
+    def process_bind_param(self, value, dialect):
+        if value is not None:
+            return json.dumps(value)
+        return None
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            return json.loads(value)
+        return None
+
+# Database connection URL (using SQLite)
+DATABASE_URL = "sqlite:///./api_proxy.db"
+
+# Create SQLAlchemy engine
+engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
+
+# Create a configured "Session" class
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# Base class for declarative models
+Base = declarative_base()
+
+# Define the KeyCategory model
+class KeyCategory(Base):
+    __tablename__ = "key_categories"
+
+    id = Column(Integer, primary_key=True, index=True)
+    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
+
+    api_keys = relationship("APIKey", back_populates="category")
+
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+# Define the APIKey model
+class APIKey(Base):
+    __tablename__ = "api_keys"
+
+    id = Column(Integer, primary_key=True, index=True)
+    value = Column(String, nullable=False)
+    category_id = Column(Integer, ForeignKey("key_categories.id"), nullable=False)
+    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_
+
+    category = relationship("KeyCategory", back_populates="api_keys")
+
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+
+# Function to create database tables
+def create_db_tables():
+    Base.metadata.create_all(bind=engine)
+
+# Dependency to get DB session
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

key_selector.py

@@ -6,7 +6,7 @@ class KeySelector():
     def __init__(self):
         pass
     
-    def get_api_key(self):
+    def get_api_key_info(self):
         api_key_info = {
             "key_at": "at_headers",
             "key_name": "X_Goog_Api_Key",

app.py → main.py

@@ -36,7 +36,7 @@ async def proxy(request: Request, path: str):
         if k.lower() not in ["host", "content-length"]}
     
     key_selector = KeySelector()
-    headers["X-Goog-Api-Key"] = key_selector.get_api_key()['key_value'] # 从数据库获取API密钥
+    headers["X-Goog-Api-Key"] = key_selector.get_api_key_info()['key_value'] # 从数据库获取API密钥
 
     try:
         # 关键修复：禁用KeepAlive防止连接冲突
@@ -133,3 +133,9 @@ async def proxy(request: Request, path: str):
     except Exception as e:
         logger.exception("Unexpected proxy error")
         raise HTTPException(500, f"内部服务器错误: {str(e)}")
+
+if __name__ == "__main__":
+    # In a real application, you would typically run uvicorn here:
+    # import uvicorn
+    # uvicorn.run(app, host="0.0.0.0", port=8000)
+    pass # Placeholder to fix indentation error

memory-bank/activeContext.md

@@ -1,25 +1,36 @@
 # 当前背景 (Active Context)
 
 ## 当前工作重点
-初始化项目记忆库。
+实现基于数据库的 API Key 管理和代理规则配置，并集成到核心代理转发逻辑中。
 
 ## 最近的变更
-- 创建了 `projectbrief.md` 文件，定义了项目的核心目标和范围。
-- 创建了 `productContext.md` 文件，描述了项目的目的、解决的问题和用户体验目标。
-- 创建了 `systemPatterns.md` 文件，概述了系统架构、设计模式和关键组件。
-- 创建了 `techContext.md` 文件，记录了使用的技术、开发环境设置和技术约束。
+- 讨论并设计了使用 SQLite 数据库存储 Key 类别、API Key 实例和代理规则的 Schema (三表设计: `key_categories`, `api_keys`, `proxy_rules`)。
+- 安装了 SQLAlchemy 和 aiosqlite 库。
+- 创建了 `app/database.py` 文件，定义了 Key 类别 (`KeyCategory`) 和 API Key 实例 (`APIKey`) 的 SQLAlchemy 模型。
+- 使用 `sqlite3` 命令成功创建了数据库表 (`api_proxy.db`)。
+- 创建了 `app/crud.py` 文件，实现了 Key 类别和 API Key 实例的基本 CRUD (创建、读取、更新、删除) 操作函数。
+- 创建了 `tests/test_crud.py` 文件，为 CRUD 函数编写了单元测试。
+- 成功运行了所有单元测试 (14 个通过)。
+- 解决了 Python 包导入 (`ModuleNotFoundError`, `NameError`) 和文件创建 (`__init__.py` 未成功创建) 等环境和代码问题。
 
 ## 下一步计划
-- 创建 `progress.md` 文件，记录项目的当前状态、已完成和待完成的工作。
-- 根据项目需求，开始实现 FastAPI 应用的核心代理转发逻辑。
-- 定义和实现代理规则的配置加载机制。
-- 集成 HTTP 客户端库 (`httpx`) 进行后端请求转发。
+- 实现 Key 管理相关的 API 路由 (`/api/keys/...`)，使用 `app/crud.py` 中的函数与数据库交互。
+- 完善 Key 选择逻辑 (`app/crud.py` 中的 `select_key_from_pool`)，实现 Key 池的轮询或其他策略。
+- 设计并实现代理规则的数据库存储和加载逻辑（`proxy_rules` 表的 CRUD 和查询）。
+- 将 Key 选择和代理规则查找逻辑集成到核心代理转发路由 (`app/routers/proxy.py`) 中。
+- 实现前端/静态文件路由 (`app/routers/frontend.py`) (如果需要)。
+- 编写其他功能的单元测试和集成测试。
 
 ## 重要的模式和偏好
 - 优先使用异步编程。
 - 保持代码简洁和模块化。
 - 遵循 FastAPI 的最佳实践。
+- 使用数据库进行配置管理以提高灵活性。
+- 通过单元测试验证核心逻辑。
 
 ## 学习和项目洞察
-- 项目处于初期阶段，主要任务是搭建基本框架和配置。
-- 需要仔细设计配置文件的结构，以便于灵活地添加和管理代理规则。
+- 确认了使用 SQLite 和 SQLAlchemy 进行数据库操作的可行性。
+- 解决了在当前环境中进行 Python 包导入和文件创建时遇到的具体问题。
+- 明确了 Key 的应用方式应与代理规则关联，而非 Key 本身。
+- 数据库三表设计 (`key_categories`, `api_keys`, `proxy_rules`) 能够更好地组织 Key 和规则信息。
+</textarea>

memory-bank/progress.md

@@ -1,24 +1,29 @@
 # 项目进度 (Progress)
 
 ## 当前状态
-项目记忆库已初始化。核心记忆文件（projectbrief.md, productContext.md, systemPatterns.md, techContext.md, activeContext.md）已创建。
+数据库已设置，Key 类别和 API Key 实例的基本 CRUD 操作已实现并通过单元测试。项目记忆库已更新。
 
 ## 已完成的工作
-- 创建了 Memory Bank 目录。
-- 创建并填充了以下记忆文件：
-    - `projectbrief.md`
-    - `productContext.md`
-    - `systemPatterns.md`
-    - `techContext.md`
-    - `activeContext.md`
+- 创建了 Memory Bank 目录并初始化了核心记忆文件。
+- 讨论并设计了基于 SQLite 的 Key 管理和代理规则数据库 Schema (三表设计)。
+- 安装了必要的 Python 依赖 (SQLAlchemy, aiosqlite)。
+- 创建了 `app/database.py` 文件，定义了 SQLAlchemy 模型和数据库连接。
+- 使用 `sqlite3` 命令成功创建了数据库表 (`api_proxy.db`)。
+- 创建了 `app/crud.py` 文件，实现了 Key 类别和 API Key 实例的 CRUD 函数。
+- 创建了 `tests/test_crud.py` 文件，为 CRUD 函数编写了单元测试。
+- 成功运行了所有单元测试 (14 个通过)。
+- 解决了环境配置和 Python 导入问题 (`__init__.py` 创建、导入路径修正)。
+- 更新了 Memory Bank 文件 (`systemPatterns.md`, `techContext.md`, `activeContext.md`, `progress.md`)。
 
 ## 待完成的工作
-- 实现 FastAPI 应用的核心代理转发逻辑。
-- 设计和实现代理规则的配置加载机制。
-- 集成 `httpx` 库进行后端请求转发。
+- 实现 Key 管理相关的 API 路由 (`/api/keys/...`)。
+- 完善 Key 选择逻辑 (`app/crud.py` 中的 `select_key_from_pool`)，实现 Key 池的轮询或其他策略。
+- 设计并实现代理规则的数据库存储和加载逻辑（`proxy_rules` 表的 CRUD 和查询）。
+- 将 Key 选择和代理规则查找逻辑集成到核心代理转发路由 (`app/routers/proxy.py`) 中。
+- 实现前端/静态文件路由 (`app/routers/frontend.py`) (如果需要)。
 - 添加基本的日志记录功能。
 - 根据需求实现请求/响应转换功能。
-- 编写单元测试和集成测试。
+- 编写剩余功能的单元测试和集成测试。
 - 完善 Dockerfile。
 - 编写详细的 README.md 文档。
 
@@ -28,4 +33,6 @@
 ## 项目决策演变
 - 决定使用 FastAPI 作为后端框架，因为它提供了高性能和易于使用的异步支持。
 - 决定使用 `httpx` 作为 HTTP 客户端，因为它支持异步请求。
-- 决定使用配置文件来管理代理规则，以提高灵活性。
+- **重要决策**: 决定使用 SQLite 数据库 (配合 SQLAlchemy 和 aiosqlite) 来管理 API Key 和代理规则，取代了最初使用静态配置文件的想法，以支持更灵活的动态管理和 Key 池化。
+- 数据库 Schema 经过讨论，从单表设计演变为三表设计 (`key_categories`, `api_keys`, `proxy_rules`)，以更好地组织数据和处理 Key 应用方式与规则关联的需求。
+</textarea>

memory-bank/systemPatterns.md

@@ -1,32 +1,39 @@
 # 系统模式 (System Patterns)
 
 ## 架构概述
-本项目将采用基于 FastAPI 的微服务架构。核心是一个 FastAPI 应用，负责接收所有进来的请求，并根据配置将请求转发到相应的后端服务。
+本项目将采用基于 FastAPI 的微服务架构。核心是一个 FastAPI 应用，负责接收所有进来的请求，并根据数据库中的配置将请求转发到相应的后端服务，并管理 API Key。
 
 ```mermaid
 graph LR
     Client --> Proxy[FastAPI Proxy]
+    Proxy --> Database[SQLite Database]
     Proxy --> BackendA[Backend Service A]
     Proxy --> BackendB[Backend Service B]
     Proxy --> BackendC[Backend Service C]
+    Database --> Proxy
 ```
 
 ## 设计模式
 - **API Gateway Pattern**: FastAPI 应用充当 API 网关，作为客户端访问后端服务的单一入口点。
-- **Configuration Pattern**: 代理规则将通过配置文件进行管理，以便于修改和扩展。
+- **Database Pattern**: 使用 SQLite 数据库存储 Key 类别、Key 实例和代理规则配置。
+- **Configuration Pattern**: 代理规则和 Key 信息通过数据库进行管理，而非静态配置文件。
 - **Middleware Pattern**: 利用 FastAPI 的中间件功能处理请求和响应的通用逻辑（如日志、认证等）。
 
 ## 关键组件
-- **FastAPI Application**: 核心应用，处理路由和请求转发。
-- **Configuration Loader**: 负责加载和解析代理规则配置文件。
+- **FastAPI Application**: 核心应用，处理路由、请求转发和 Key 管理 API。
+- **SQLite Database**: 存储 Key 类别 (`key_categories`)、API Key 实例 (`api_keys`) 和代理规则 (`proxy_rules`)。
+- **SQLAlchemy Models**: 定义与数据库表对应的 Python 模型 (`KeyCategory`, `APIKey`)。
+- **CRUD Operations**: 实现对数据库中 Key 类别和 Key 实例的增删改查逻辑 (`app/crud.py`)。
+- **Key Selection Logic**: 根据代理规则从 Key 池中选择合适 Key 的逻辑。
 - **HTTP Client**: 用于向后端服务发起请求（例如使用 `httpx` 库）。
 - **Request/Response Transformer (Optional)**: 根据配置修改请求和响应。
 
 ## 数据流
 1. 客户端发起请求到 FastAPI Proxy。
 2. FastAPI Proxy 接收请求。
-3. Configuration Loader 解析配置，查找匹配的代理规则。
-4. 如果找到匹配规则，HTTP Client 根据规则向后端服务发起请求。
-5. 后端服务返回响应。
-6. Request/Response Transformer (如果存在) 修改响应。
-7. FastAPI Proxy 将响应返回给客户端。
+3. **Proxy 查找数据库**: 根据请求查找数据库 (`proxy_rules` 表) 获取匹配的代理规则。
+4. **Key 选择 (如果需要)**: 如果匹配规则需要 Key，Proxy 查询数据库 (`api_keys` 表) 获取 Key 池，并根据策略选择一个 Key。
+5. **应用 Key 并转发**: Proxy 根据规则中定义的应用方式，将选定的 Key 应用到转发请求中。使用 HTTP Client 向后端服务发起请求。
+6. 后端服务返回响应。
+7. Request/Response Transformer (如果存在) 修改响应。
+8. FastAPI Proxy 将响应返回给客户端。

memory-bank/techContext.md

@@ -2,8 +2,11 @@
 
 ## 使用的技术
 - **后端框架**: FastAPI (Python)
+- **数据库**: SQLite
+- **ORM**: SQLAlchemy
+- **异步 SQLite 驱动**: aiosqlite
 - **异步 HTTP 客户端**: httpx
-- **配置管理**: 可能使用 `python-dotenv` 或其他配置库来加载配置文件。
+- **配置管理**: 数据库用于存储代理规则和 Key 信息。
 - **依赖管理**: pip 和 requirements.txt
 - **容器化**: Docker
 
@@ -11,16 +14,19 @@
 1. 克隆仓库。
 2. 创建并激活 Conda 虚拟环境 (`conda create -n api-proxy python=3.9` 或使用现有环境，然后 `conda activate api-proxy`)。
 3. 安装依赖 (`pip install -r requirements.txt`)。
-4. 运行应用 (`uvicorn app:app --reload`)。
+4. **创建数据库**: 运行数据库初始化脚本或命令（例如 `sqlite3 ./api_proxy.db "..."`）。
+5. 运行应用 (`uvicorn app:app --reload`)。
 
 ## 技术约束
 - 需要 Python 3.7+。
-- 依赖于 FastAPI 和 httpx 库。
+- 依赖于 FastAPI, httpx, SQLAlchemy, aiosqlite 库。
 
 ## 依赖关系
-- `requirements.txt` 文件列出了所有必要的 Python 依赖。
+- `requirements.txt` 文件列出了所有必要的 Python 依赖，包括 FastAPI, httpx, SQLAlchemy, aiosqlite。
 
 ## 工具使用模式
 - 使用 `uvicorn` 作为 ASGI 服务器运行 FastAPI 应用。
 - 使用 `pip` 管理 Python 包。
 - 使用 Docker 构建和运行容器。
+- 使用 `sqlite3` 命令或 SQLAlchemy 进行数据库管理。
+- 使用 `pytest` 运行单元测试。

requirements.txt

@@ -1,3 +1,5 @@
 fastapi
 uvicorn[standard]
 httpx
+SQLAlchemy
+aiosqlite

tests/test_crud.py

@@ -0,0 +1,152 @@
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session # Import Session
+from app.database import Base, get_db, KeyCategory, APIKey # Import Base, get_db, and models from database
+from app import crud # Import crud from app.crud
+
+# Use an in-memory SQLite database for testing
+SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# Override the get_db dependency for testing
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+# Fixture to create and drop tables for each test
+@pytest.fixture(scope="function")
+def db_session():
+    Base.metadata.create_all(bind=engine)
+    db = TestingSessionLocal()
+    yield db
+    db.close()
+    Base.metadata.drop_all(bind=engine)
+
+# --- Test KeyCategory CRUD ---
+
+def test_create_key_category(db_session: Session):
+    category = crud.create_key_category(db_session, name="test_llm", type="llm", tags=["llm", "test"])
+    assert category.id is not None
+    assert category.name == "test_llm"
+    assert category.type == "llm"
+    assert category.tags == ["llm", "test"]
+
+def test_get_key_category(db_session: Session):
+    created_category = crud.create_key_category(db_session, name="get_test", type="test")
+    retrieved_category = crud.get_key_category(db_session, created_category.id)
+    assert retrieved_category is not None
+    assert retrieved_category.name == "get_test"
+
+def test_get_key_category_by_name(db_session: Session):
+    crud.create_key_category(db_session, name="get_by_name_test", type="test")
+    retrieved_category = crud.get_key_category_by_name(db_session, "get_by_name_test")
+    assert retrieved_category is not None
+    assert retrieved_category.name == "get_by_name_test"
+
+def test_get_key_categories(db_session: Session):
+    crud.create_key_category(db_session, name="list_test_1", type="test")
+    crud.create_key_category(db_session, name="list_test_2", type="test")
+    categories = crud.get_key_categories(db_session)
+    assert len(categories) >= 2 # Account for potential other categories if not in-memory db
+
+def test_update_key_category(db_session: Session):
+    created_category = crud.create_key_category(db_session, name="update_test", type="old_type")
+    updated_category = crud.update_key_category(db_session, created_category.id, name="updated_test", type="new_type", tags=["updated"])
+    assert updated_category.name == "updated_test"
+    assert updated_category.type == "new_type"
+    assert updated_category.tags == ["updated"]
+
+def test_delete_key_category(db_session: Session):
+    created_category = crud.create_key_category(db_session, name="delete_test", type="test")
+    deleted_category = crud.delete_key_category(db_session, created_category.id)
+    assert deleted_category is not None
+    retrieved_category = crud.get_key_category(db_session, created_category.id)
+    assert retrieved_category is None
+
+# --- Test APIKey CRUD ---
+
+def test_create_api_key(db_session: Session):
+    category = crud.create_key_category(db_session, name="key_cat_for_key", type="test")
+    api_key = crud.create_api_key(db_session, value="test_key_value", category_id=category.id)
+    assert api_key.id is not None
+    assert api_key.value == "test_key_value"
+    assert api_key.category_id == category.id
+    assert api_key.status == "active"
+
+def test_get_api_key(db_session: Session):
+    category = crud.create_key_category(db_session, name="get_key_cat", type="test")
+    created_key = crud.create_api_key(db_session, value="get_test_key", category_id=category.id)
+    retrieved_key = crud.get_api_key(db_session, created_key.id)
+    assert retrieved_key is not None
+    assert retrieved_key.value == "get_test_key"
+
+def test_get_api_keys(db_session: Session):
+    category1 = crud.create_key_category(db_session, name="list_key_cat_1", type="test")
+    category2 = crud.create_key_category(db_session, name="list_key_cat_2", type="test")
+    crud.create_api_key(db_session, value="list_key_1", category_id=category1.id)
+    crud.create_api_key(db_session, value="list_key_2", category_id=category1.id)
+    crud.create_api_key(db_session, value="list_key_3", category_id=category2.id)
+
+    all_keys = crud.get_api_keys(db_session)
+    assert len(all_keys) >= 3
+
+    category1_keys = crud.get_api_keys(db_session, category_id=category1.id)
+    assert len(category1_keys) == 2
+
+    active_keys = crud.get_api_keys(db_session, status="active")
+    assert len(active_keys) >= 3 # All created keys are active by default
+
+    inactive_keys = crud.get_api_keys(db_session, status="inactive")
+    assert len(inactive_keys) == 0
+
+def test_update_api_key(db_session: Session):
+    category1 = crud.create_key_category(db_session, name="update_key_cat_1", type="test")
+    category2 = crud.create_key_category(db_session, name="update_key_cat_2", type="test")
+    created_key = crud.create_api_key(db_session, value="old_key_value", category_id=category1.id, status="active")
+
+    updated_key = crud.update_api_key(db_session, created_key.id, value="new_key_value", category_id=category2.id, status="inactive")
+    assert updated_key.value == "new_key_value"
+    assert updated_key.category_id == category2.id
+    assert updated_key.status == "inactive"
+
+def test_delete_api_key(db_session: Session):
+    category = crud.create_key_category(db_session, name="delete_key_cat", type="test")
+    created_key = crud.create_api_key(db_session, value="delete_test_key", category_id=category.id)
+    deleted_key = crud.delete_api_key(db_session, created_key.id)
+    assert deleted_key is not None
+    retrieved_key = crud.get_api_key(db_session, created_key.id)
+    assert retrieved_key is None
+
+# --- Test Key Selection Logic Placeholder ---
+
+def test_get_available_keys_for_category(db_session: Session):
+    category = crud.create_key_category(db_session, name="available_key_cat", type="test")
+    crud.create_api_key(db_session, value="key1", category_id=category.id, status="active")
+    crud.create_api_key(db_session, value="key2", category_id=category.id, status="inactive")
+    crud.create_api_key(db_session, value="key3", category_id=category.id, status="active")
+
+    available_keys = crud.get_available_keys_for_category(db_session, category.id)
+    assert len(available_keys) == 2
+    assert all(key.status == "active" for key in available_keys)
+    assert {key.value for key in available_keys} == {"key1", "key3"}
+
+def test_select_key_from_pool_no_keys(db_session: Session):
+    category = crud.create_key_category(db_session, name="empty_pool_cat", type="test")
+    selected_key = crud.select_key_from_pool(db_session, category.id)
+    assert selected_key is None
+
+def test_select_key_from_pool_basic(db_session: Session):
+    category = crud.create_key_category(db_session, name="basic_pool_cat", type="test")
+    key1 = crud.create_api_key(db_session, value="key1", category_id=category.id, status="active")
+    key2 = crud.create_api_key(db_session, value="key2", category_id=category.id, status="active")
+
+    # Basic selection returns the first one found
+    selected_key = crud.select_key_from_pool(db_session, category.id)
+    assert selected_key is not None
+    # The order might depend on DB implementation, but it should be one of the active keys
+    assert selected_key.value in ["key1", "key2"]