8ea9fff334
✅ Fully functional FastAPI server with comprehensive features: 🏗️ Architecture: - Complete API design documentation - Modular structure (models, auth, service, main) - OpenAPI/Swagger auto-documentation 🔧 Core Features: - Memory CRUD endpoints (POST, GET, DELETE) - User management and statistics - Search functionality with filtering - Admin endpoints with proper authorization 🔐 Security & Auth: - API key authentication (Bearer token) - Rate limiting (100 req/min configurable) - Input validation with Pydantic models - Comprehensive error handling 🧪 Testing: - Comprehensive test suite with automated server lifecycle - Simple test suite for quick validation - All functionality verified and working 🐛 Fixes: - Resolved Pydantic v2 compatibility (.dict() → .model_dump()) - Fixed missing dependencies (posthog, qdrant-client, vecs, ollama) - Fixed mem0 package version metadata issues 📊 Performance: - Async operations for scalability - Request timing middleware - Proper error boundaries - Health monitoring endpoints 🎯 Status: Phase 2 100% complete - REST API fully functional 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
145 lines
6.7 KiB
Python
145 lines
6.7 KiB
Python
"""
|
|
Pydantic models for API request/response validation
|
|
"""
|
|
|
|
from typing import Optional, List, Dict, Any, Union
|
|
from pydantic import BaseModel, Field, validator
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
|
|
class Message(BaseModel):
|
|
"""Message model for memory input"""
|
|
role: str = Field(..., description="Message role (user, assistant)")
|
|
content: str = Field(..., description="Message content", min_length=1, max_length=10000)
|
|
|
|
|
|
class AddMemoryRequest(BaseModel):
|
|
"""Request model for adding memories"""
|
|
messages: List[Message] = Field(..., description="List of messages to process")
|
|
user_id: str = Field(..., description="User identifier", min_length=1, max_length=100)
|
|
metadata: Optional[Dict[str, Any]] = Field(default={}, description="Additional metadata")
|
|
|
|
@validator('user_id')
|
|
def validate_user_id(cls, v):
|
|
if not v.strip():
|
|
raise ValueError('user_id cannot be empty')
|
|
return v.strip()
|
|
|
|
|
|
class SearchMemoriesRequest(BaseModel):
|
|
"""Request model for searching memories"""
|
|
query: str = Field(..., description="Search query", min_length=1, max_length=1000)
|
|
user_id: str = Field(..., description="User identifier", min_length=1, max_length=100)
|
|
limit: Optional[int] = Field(default=10, description="Number of results", ge=1, le=100)
|
|
threshold: Optional[float] = Field(default=0.0, description="Similarity threshold", ge=0.0, le=1.0)
|
|
|
|
|
|
class UpdateMemoryRequest(BaseModel):
|
|
"""Request model for updating memories"""
|
|
content: Optional[str] = Field(None, description="Updated memory content", max_length=10000)
|
|
metadata: Optional[Dict[str, Any]] = Field(None, description="Updated metadata")
|
|
|
|
|
|
class MemoryResponse(BaseModel):
|
|
"""Response model for memory objects"""
|
|
id: str = Field(..., description="Memory identifier")
|
|
memory: str = Field(..., description="Processed memory content")
|
|
user_id: str = Field(..., description="User identifier")
|
|
hash: Optional[str] = Field(None, description="Content hash")
|
|
score: Optional[float] = Field(None, description="Similarity score")
|
|
metadata: Optional[Dict[str, Any]] = Field(default={}, description="Memory metadata")
|
|
created_at: datetime = Field(..., description="Creation timestamp")
|
|
updated_at: Optional[datetime] = Field(None, description="Last update timestamp")
|
|
|
|
|
|
class MemoryAddResult(BaseModel):
|
|
"""Result of adding a memory"""
|
|
id: str = Field(..., description="Memory identifier")
|
|
memory: str = Field(..., description="Processed memory content")
|
|
event: str = Field(..., description="Event type (ADD, UPDATE)")
|
|
previous_memory: Optional[str] = Field(None, description="Previous memory content if updated")
|
|
|
|
|
|
class StandardResponse(BaseModel):
|
|
"""Standard API response format"""
|
|
success: bool = Field(..., description="Operation success status")
|
|
data: Optional[Union[Dict[str, Any], List[Any]]] = Field(None, description="Response data")
|
|
message: str = Field(..., description="Response message")
|
|
timestamp: datetime = Field(default_factory=datetime.now, description="Response timestamp")
|
|
request_id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Request identifier")
|
|
|
|
|
|
class ErrorResponse(BaseModel):
|
|
"""Error response format"""
|
|
success: bool = Field(default=False, description="Always false for errors")
|
|
error: Dict[str, Any] = Field(..., description="Error details")
|
|
timestamp: datetime = Field(default_factory=datetime.now, description="Error timestamp")
|
|
|
|
|
|
class ErrorDetail(BaseModel):
|
|
"""Error detail structure"""
|
|
code: str = Field(..., description="Error code")
|
|
message: str = Field(..., description="Human readable error message")
|
|
details: Optional[Dict[str, Any]] = Field(default={}, description="Additional error details")
|
|
request_id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Request identifier")
|
|
|
|
|
|
class HealthResponse(BaseModel):
|
|
"""Health check response"""
|
|
status: str = Field(..., description="Health status")
|
|
timestamp: datetime = Field(default_factory=datetime.now, description="Check timestamp")
|
|
uptime: Optional[float] = Field(None, description="Server uptime in seconds")
|
|
|
|
|
|
class SystemStatusResponse(BaseModel):
|
|
"""System status response"""
|
|
status: str = Field(..., description="Overall system status")
|
|
version: str = Field(..., description="API version")
|
|
mem0_version: str = Field(..., description="mem0 library version")
|
|
services: Dict[str, str] = Field(..., description="Service status")
|
|
database: Dict[str, Any] = Field(..., description="Database status")
|
|
timestamp: datetime = Field(default_factory=datetime.now, description="Status timestamp")
|
|
|
|
|
|
class UserStatsResponse(BaseModel):
|
|
"""User statistics response"""
|
|
user_id: str = Field(..., description="User identifier")
|
|
total_memories: int = Field(..., description="Total number of memories")
|
|
recent_memories: int = Field(..., description="Memories added in last 24h")
|
|
oldest_memory: Optional[datetime] = Field(None, description="Oldest memory timestamp")
|
|
newest_memory: Optional[datetime] = Field(None, description="Newest memory timestamp")
|
|
storage_usage: Dict[str, Any] = Field(..., description="Storage usage statistics")
|
|
|
|
|
|
class SearchResultsResponse(BaseModel):
|
|
"""Search results response"""
|
|
results: List[MemoryResponse] = Field(..., description="Search results")
|
|
query: str = Field(..., description="Original search query")
|
|
total_results: int = Field(..., description="Total number of results")
|
|
execution_time: float = Field(..., description="Search execution time in seconds")
|
|
|
|
|
|
class DeleteResponse(BaseModel):
|
|
"""Delete operation response"""
|
|
deleted: bool = Field(..., description="Deletion success status")
|
|
memory_id: str = Field(..., description="Deleted memory identifier")
|
|
message: str = Field(..., description="Deletion message")
|
|
|
|
|
|
class BulkDeleteResponse(BaseModel):
|
|
"""Bulk delete operation response"""
|
|
deleted_count: int = Field(..., description="Number of deleted memories")
|
|
user_id: str = Field(..., description="User identifier")
|
|
message: str = Field(..., description="Bulk deletion message")
|
|
|
|
|
|
class APIMetricsResponse(BaseModel):
|
|
"""API metrics response"""
|
|
total_requests: int = Field(..., description="Total API requests")
|
|
requests_per_minute: float = Field(..., description="Average requests per minute")
|
|
average_response_time: float = Field(..., description="Average response time in ms")
|
|
error_rate: float = Field(..., description="Error rate percentage")
|
|
active_users: int = Field(..., description="Active users in last hour")
|
|
top_endpoints: List[Dict[str, Any]] = Field(..., description="Most used endpoints")
|
|
timestamp: datetime = Field(default_factory=datetime.now, description="Metrics timestamp") |