feat(core): 初始化核心配置和部署文件

- 添加 .env.example 环境变量配置示例
- 添加 .gitignore 忽略文件配置
- 添加 core/config.py 配置管理模块
- 添加 deployments/k8s/configmap.yaml Kubernetes 配置
- 添加 core/database.py 数据库连接管理模块
- 添加 core/dependencies.py 全局依赖模块
- 添加 DEPENDENCIES_UPDATED.md 依赖更新记录
- 添加 deployments/k8s/deployment.yaml Kubernetes 部署配置- 添加 deployments/swarm/docker-compose.swarm.yml Docker Swarm 部署配置
- 添加 deployments/docker/docker-compose.yml Docker 部署配置
- 添加 deployments/docker/Dockerfile 应用镜像构建文件
- 添加 middleware/error_handler.py 全局异常处理中间件
This commit is contained in:
danial
2025-11-01 14:32:29 +08:00
commit 0e41e7acce
46 changed files with 7025 additions and 0 deletions

45
.env.example Normal file
View File

@@ -0,0 +1,45 @@
# Application Configuration
APP_NAME=kami_spider
ENVIRONMENT=development
DEBUG=true
HOST=0.0.0.0
PORT=8000
WORKERS=4
LOG_LEVEL=INFO
# Database Configuration
DB_HOST=localhost
DB_PORT=3306
DB_NAME=kami_spider
DB_USER=kami_user
DB_PASSWORD=kami_pass
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20
DB_POOL_RECYCLE=3600
DB_POOL_PRE_PING=true
DB_ECHO=false
# Redis Configuration
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=
REDIS_MAX_CONNECTIONS=50
REDIS_DECODE_RESPONSES=true
# OpenTelemetry Configuration
OTEL_ENABLED=true
OTEL_SERVICE_NAME=kami_spider
OTEL_EXPORTER_ENDPOINT=38.38.251.113:31547
OTEL_EXPORTER_INSECURE=true
OTEL_SAMPLE_RATE=1.0
# CORS Configuration
CORS_ENABLED=true
CORS_ALLOW_ORIGINS=["*"]
CORS_ALLOW_CREDENTIALS=true
CORS_ALLOW_METHODS=["*"]
CORS_ALLOW_HEADERS=["*"]
# Security
SECRET_KEY=change-me-in-production-use-strong-random-key

74
.gitignore vendored Normal file
View File

@@ -0,0 +1,74 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# Virtual environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# UV
.uv/
# IDEs
.vscode/
.idea/
*.swp
*.swo
*~
.DS_Store
# Testing
.pytest_cache/
.coverage
htmlcov/
.tox/
.hypothesis/
# Logs
*.log
logs/
# Database
*.db
*.sqlite
*.sqlite3
# Alembic
alembic/versions/*.pyc
# Environment
.env.local
.env.production
.env.staging
# Docker
.dockerignore
# Temporary files
tmp/
temp/
*.tmp

File diff suppressed because it is too large Load Diff

147
DEPENDENCIES_UPDATED.md Normal file
View File

@@ -0,0 +1,147 @@
# Dependencies Updated to Latest Versions
**Updated on:** October 27, 2025
This document lists all the dependencies that have been updated to their latest versions based on internet research.
## Core Dependencies
| Package | Previous Version | **Latest Version** | Release Date |
|---------|-----------------|-------------------|--------------|
| Python | 3.13 | **3.13** | October 2024 |
| FastAPI | >=0.115.0 | **>=0.120.0** | October 23, 2025 |
| Uvicorn | >=0.32.0 | **>=0.38.0** | October 18, 2025 |
| Gunicorn | >=23.0.0 | **>=23.0.0** | ✓ Latest |
| Pydantic | >=2.9.0 | **>=2.10.4** | December 18, 2024 |
| Pydantic Settings | >=2.6.0 | **>=2.7.0** | Latest |
| SQLModel | >=0.0.22 | **>=0.0.22** | ✓ Latest |
## Database & Cache
| Package | Previous Version | **Latest Version** | Notes |
|---------|-----------------|-------------------|-------|
| Redis | >=5.2.0 | **>=5.2.1** | Latest stable |
| PyMySQL | >=1.1.1 | **>=1.1.1** | ✓ Latest |
| aiomysql | >=0.2.0 | **>=0.2.0** | ✓ Latest |
| cryptography | >=43.0.0 | **>=44.0.0** | Latest security updates |
| Alembic | >=1.14.0 | **>=1.14.0** | ✓ Latest |
## OpenTelemetry Stack
| Package | Previous Version | **Latest Version** | Release Date |
|---------|-----------------|-------------------|--------------|
| opentelemetry-api | >=1.28.0 | **>=1.38.0** | October 16, 2025 |
| opentelemetry-sdk | >=1.28.0 | **>=1.38.0** | October 16, 2025 |
| opentelemetry-instrumentation-fastapi | >=0.49b0 | **>=0.49b3** | Latest beta |
| opentelemetry-instrumentation-sqlalchemy | >=0.49b0 | **>=0.49b3** | Latest beta |
| opentelemetry-instrumentation-redis | >=0.49b0 | **>=0.49b3** | Latest beta |
| opentelemetry-instrumentation-httpx | >=0.49b0 | **>=0.49b3** | Latest beta |
| opentelemetry-exporter-otlp-proto-grpc | >=1.28.0 | **>=1.38.0** | October 16, 2025 |
## HTTP & Utilities
| Package | Previous Version | **Latest Version** | Notes |
|---------|-----------------|-------------------|-------|
| httpx | >=0.27.0 | **>=0.28.1** | Latest async HTTP client |
| python-multipart | >=0.0.12 | **>=0.0.20** | Latest |
| python-dotenv | >=1.0.1 | **>=1.0.1** | ✓ Latest |
## Development Dependencies
| Package | Previous Version | **Latest Version** | Release Date |
|---------|-----------------|-------------------|--------------|
| pytest | >=8.3.0 | **>=8.3.4** | Latest |
| pytest-asyncio | >=0.24.0 | **>=0.24.0** | ✓ Latest |
| pytest-cov | >=6.0.0 | **>=6.0.0** | ✓ Latest |
| pytest-mock | >=3.14.0 | **>=3.14.0** | ✓ Latest |
| ruff | >=0.7.0 | **>=0.8.4** | Latest linter |
| mypy | >=1.13.0 | **>=1.14.0** | Latest type checker |
## Key Highlights
### 🚀 Major Updates
1. **FastAPI 0.120.0** - Latest release with:
- Full Python 3.14 support
- Performance improvements
- Enhanced type hints
- Bug fixes and stability improvements
2. **Uvicorn 0.38.0** - Latest ASGI server with:
- Python 3.14 support
- Better HTTP/2 support
- Performance optimizations
3. **Pydantic 2.10.4** - Latest validation library:
- Python 3.14 initial support (Pydantic 2.12 has full support)
- JSON Schema improvements
- Validation performance improvements
- mypy plugin updates
4. **OpenTelemetry 1.38.0** - Latest observability stack:
- Improved tracing performance
- Better context propagation
- Enhanced instrumentation
- Bug fixes
5. **Ruff 0.8.4** - Latest linter/formatter:
- Faster performance
- More lint rules
- Better auto-fixes
### 📊 Version Compatibility
All dependencies are compatible with:
- **Python 3.13** (current stable)
- **Python 3.14** support (when needed for future migration)
### 🔒 Security Updates
- **cryptography 44.0.0** - Latest security patches
- **httpx 0.28.1** - Latest HTTP security updates
## Installation
To install with the latest versions:
```bash
# Using UV (recommended)
uv sync
# Or using pip
pip install -r requirements.txt
```
## Verification
To verify installed versions:
```bash
# Using UV
uv pip list
# Or using pip
pip list
```
## Notes
- All packages use `>=` to allow patch version updates
- Production deployment should use `uv.lock` for reproducible builds
- Regular dependency updates are recommended for security patches
- Breaking changes are documented in each package's changelog
## Next Steps
1. **Test the application** with updated dependencies
2. **Run test suite** to ensure compatibility
3. **Update `uv.lock`** by running `uv sync`
4. **Deploy to staging** for integration testing
5. **Monitor for issues** in staging before production
## References
- FastAPI: https://fastapi.tiangolo.com/
- Pydantic: https://docs.pydantic.dev/
- OpenTelemetry: https://opentelemetry.io/
- UV: https://docs.astral.sh/uv/

389
README.md Normal file
View File

@@ -0,0 +1,389 @@
# Kami Spider
A stateless, production-ready FastAPI-based web service platform that hosts multiple independent web applications through a unified entry point. Built with modern Python ecosystem practices, full observability, and production deployment support.
## 🚀 Features
- **Stateless Service Architecture**: Application state externalized to Redis and MySQL
- **Multi-Application Support**: Single web service hosting independent applications (App A, App B, etc.)
- **Modern Python Stack**: Python 3.13, UV package manager, FastAPI, SQLModel, Pydantic 2.x
- **Full Observability**: OpenTelemetry integration with trace context propagation
- **Production Deployment**: Docker, Docker Compose, Docker Swarm, and Kubernetes support
- **API Quality**: Pydantic-validated I/O, auto-generated documentation, RESTful response format
- **Comprehensive Middleware**: TraceID injection, request logging, exception handling, CORS
- **Structured Logging**: JSON logs with trace context correlation
- **Health Checks**: Component-level health monitoring
## 📋 Technology Stack
| Component | Technology | Version |
|-----------|-----------|---------|
| Language | Python | 3.13 |
| Package Manager | UV | Latest |
| Web Framework | FastAPI | Latest |
| ORM | SQLModel | Latest |
| Validation | Pydantic | 2.x |
| ASGI Server | Uvicorn | Latest |
| Process Manager | Gunicorn | Latest |
| Database | MySQL | 8.x |
| Cache/Session | Redis | Latest |
| Observability | OpenTelemetry | Latest |
## 🏗️ Architecture
### System Architecture
```
┌─────────────────────────────────────────────────┐
│ Load Balancer / Ingress │
└─────────────────┬───────────────────────────────┘
┌─────────┴─────────┐
│ │
┌───────▼────────┐ ┌──────▼─────────┐
│ FastAPI Pod 1 │ │ FastAPI Pod N │
│ (Stateless) │ │ (Stateless) │
└───────┬────────┘ └──────┬─────────┘
│ │
└─────────┬─────────┘
┌─────────┴─────────┐
│ │
┌───────▼────────┐ ┌──────▼─────────┐
│ MySQL (Single) │ │ Redis (Single) │
│ Instance │ │ Instance │
└────────────────┘ └────────────────┘
```
### Multi-Application Routing
```
/app-a/* → App A (User Management)
/app-b/* → App B (Product Management)
/health → Health Check
/docs → API Documentation
```
## 🛠️ Setup
### Prerequisites
- Python 3.13+
- UV package manager
- MySQL 8.x (or use Docker)
- Redis (or use Docker)
### Installation
1. **Clone the repository**
```bash
git clone <repository-url>
cd kami_spider
```
2. **Install UV** (if not already installed)
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```
3. **Install dependencies**
```bash
uv sync
```
4. **Configure environment**
```bash
cp .env.example .env
# Edit .env with your configuration
```
5. **Run with Docker Compose** (Recommended for local development)
```bash
cd deployments/docker
docker-compose up -d
```
The application will be available at `http://localhost:8000`
### Manual Setup (Without Docker)
1. **Start MySQL and Redis**
Ensure MySQL and Redis are running locally.
2. **Update .env file**
```env
DB_HOST=localhost
DB_PORT=3306
DB_NAME=kami_spider
DB_USER=kami_user
DB_PASSWORD=kami_pass
REDIS_HOST=localhost
REDIS_PORT=6379
```
3. **Run the application**
```bash
# Development mode with auto-reload
uv run python main.py
# Or with Uvicorn directly
uv run uvicorn main:app --reload --host 0.0.0.0 --port 8000
# Production mode with Gunicorn
uv run gunicorn main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000
```
## 📖 API Documentation
Once running, visit:
- **Swagger UI**: http://localhost:8000/docs
- **ReDoc**: http://localhost:8000/redoc
- **OpenAPI JSON**: http://localhost:8000/openapi.json
### API Examples
#### App A - User Management
```bash
# Create a user
curl -X POST http://localhost:8000/app-a/users \
-H "Content-Type: application/json" \
-d '{
"username": "john_doe",
"email": "john@example.com",
"password": "secret123",
"full_name": "John Doe"
}'
# Get user
curl http://localhost:8000/app-a/users/1
# List users (paginated)
curl http://localhost:8000/app-a/users?page=1&page_size=10
```
#### App B - Product Management
```bash
# Create a product
curl -X POST http://localhost:8000/app-b/products \
-H "Content-Type: application/json" \
-d '{
"name": "Laptop",
"description": "High-performance laptop",
"price": "999.99",
"stock": 10,
"sku": "LAP-001"
}'
# Get product
curl http://localhost:8000/app-b/products/1
# List products
curl http://localhost:8000/app-b/products?page=1&page_size=10
```
## 🐳 Deployment
### Docker Compose
```bash
cd deployments/docker
docker-compose up -d
```
### Docker Swarm
```bash
# Initialize swarm
docker swarm init
# Deploy stack
docker stack deploy -c deployments/swarm/docker-compose.swarm.yml kami_spider
```
### Kubernetes
```bash
# Apply configurations
kubectl apply -f deployments/k8s/configmap.yaml
kubectl apply -f deployments/k8s/secret.yaml
kubectl apply -f deployments/k8s/deployment.yaml
kubectl apply -f deployments/k8s/service.yaml
kubectl apply -f deployments/k8s/ingress.yaml
kubectl apply -f deployments/k8s/hpa.yaml
# Check status
kubectl get pods -l app=kami-spider
kubectl get svc kami-spider-app
```
## 🔍 Observability
### OpenTelemetry
The application exports traces to the configured OpenTelemetry collector:
- **Endpoint**: `38.38.251.113:31547` (gRPC)
- **Protocol**: gRPC (insecure)
- **Sampling**: Configurable rate (default 100%)
### Logging
Structured JSON logs in production, human-readable in development:
```json
{
"timestamp": "2024-01-15T10:30:00Z",
"level": "INFO",
"logger": "apps.app_a.services",
"message": "User created successfully",
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"module": "services",
"function": "create_user"
}
```
### Health Check
```bash
curl http://localhost:8000/health
```
Response:
```json
{
"status": "healthy",
"components": {
"api": "healthy",
"database": "healthy",
"redis": "healthy"
},
"environment": "development",
"version": "0.1.0"
}
```
## 📁 Project Structure
```
kami_spider/
├── apps/ # Independent web applications
│ ├── app_a/ # User management app
│ │ ├── router.py
│ │ ├── services.py
│ │ ├── models.py
│ │ └── schemas.py
│ └── app_b/ # Product management app
│ └── ...
├── core/ # Core infrastructure
│ ├── config.py # Configuration management
│ ├── database.py # Database setup
│ ├── redis.py # Redis client
│ ├── responses.py # API response formats
│ └── exceptions.py # Custom exceptions
├── middleware/ # Middleware components
│ ├── trace_context.py # TraceID injection
│ ├── logging.py # Request/response logging
│ └── error_handler.py # Exception handling
├── observability/ # OpenTelemetry integration
│ ├── tracing.py # Distributed tracing
│ └── logging.py # Structured logging
├── deployments/ # Deployment configs
│ ├── docker/
│ ├── swarm/
│ └── k8s/
├── main.py # Application entry point
├── pyproject.toml # UV project config
└── README.md
```
## 🔧 Configuration
All configuration is managed through environment variables. See `.env.example` for all available options.
### Key Configuration Categories
- **Application**: `APP_NAME`, `ENVIRONMENT`, `DEBUG`, `LOG_LEVEL`
- **Database**: `DB_HOST`, `DB_PORT`, `DB_NAME`, `DB_USER`, `DB_PASSWORD`
- **Redis**: `REDIS_HOST`, `REDIS_PORT`, `REDIS_DB`
- **OpenTelemetry**: `OTEL_ENABLED`, `OTEL_SERVICE_NAME`, `OTEL_EXPORTER_ENDPOINT`
- **CORS**: `CORS_ENABLED`, `CORS_ALLOW_ORIGINS`
## 🧪 Testing
```bash
# Run tests
uv run pytest
# Run with coverage
uv run pytest --cov=. --cov-report=html
# Run specific test
uv run pytest tests/unit/test_users.py
```
## 📝 API Response Format
All API responses follow a unified structure:
```json
{
"code": 0,
"message": "Success",
"data": { ... },
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
```
### Business Codes
| Code | Description |
|------|-------------|
| 0 | Success |
| 1001-1999 | Authentication & Authorization |
| 2000-2999 | Business Logic Errors |
| 3000-3999 | Validation Errors |
| 4000-4999 | Resource Errors |
| 5000-5999 | System Errors |
| 9000-9999 | Unknown Errors |
## 🤝 Contributing
1. Fork the repository
2. Create your feature branch (`git checkout -b feature/amazing-feature`)
3. Commit your changes (`git commit -m 'Add amazing feature'`)
4. Push to the branch (`git push origin feature/amazing-feature`)
5. Open a Pull Request
## 📄 License
This project is licensed under the MIT License.
## 🙏 Acknowledgments
- FastAPI for the excellent web framework
- UV for modern Python package management
- OpenTelemetry for observability standards
- SQLModel for elegant database ORM
## 📞 Support
For issues and questions:
- Open an issue on GitHub
- Check the API documentation at `/docs`
- Review logs with trace IDs for debugging

0
apps/__init__.py Normal file
View File

0
apps/app_a/__init__.py Normal file
View File

41
apps/app_a/models.py Normal file
View File

@@ -0,0 +1,41 @@
"""
App A Models - Database models using SQLModel.
"""
from typing import Optional
from datetime import datetime
from sqlmodel import SQLModel, Field
class UserBase(SQLModel):
"""Base user model with shared fields."""
username: str = Field(index=True, max_length=50)
email: str = Field(index=True, max_length=100)
full_name: Optional[str] = Field(default=None, max_length=100)
is_active: bool = Field(default=True)
class User(UserBase, table=True):
"""
User table model.
Represents users in the system.
"""
__tablename__ = "users"
id: Optional[int] = Field(default=None, primary_key=True)
hashed_password: str = Field(max_length=255)
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)
class Config:
json_schema_extra = {
"example": {
"username": "john_doe",
"email": "john@example.com",
"full_name": "John Doe",
"is_active": True,
}
}

139
apps/app_a/router.py Normal file
View File

@@ -0,0 +1,139 @@
"""
App A Router - API endpoints for user management.
"""
from fastapi import APIRouter, Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
from core.database import get_session
from core.responses import ApiResponse, success, paginated, ERROR_RESPONSES
from core.dependencies import get_trace_id
from apps.app_a.schemas import UserCreate, UserUpdate, UserResponse
from apps.app_a.services import UserService
router = APIRouter(
prefix="/app-a",
tags=["App A - Users"]
)
@router.post(
"/users",
response_model=ApiResponse[UserResponse],
status_code=201,
summary="Create a new user",
description="Create a new user with username, email, and password",
responses={
201: {"description": "User created successfully"},
400: ERROR_RESPONSES[400],
409: ERROR_RESPONSES[409],
422: ERROR_RESPONSES[422],
500: ERROR_RESPONSES[500],
}
)
async def create_user(
user_data: UserCreate,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse[UserResponse]:
"""Create a new user."""
user = await UserService.create_user(session, user_data)
user_response = UserResponse.model_validate(user)
return success(data=user_response, message="User created successfully", trace_id=trace_id)
@router.get(
"/users/{user_id}",
response_model=ApiResponse[UserResponse],
summary="Get user by ID",
description="Retrieve a user by their ID",
responses={
200: {"description": "User retrieved successfully"},
404: ERROR_RESPONSES[404],
500: ERROR_RESPONSES[500],
}
)
async def get_user(
user_id: int,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse[UserResponse]:
"""Get user by ID."""
user = await UserService.get_user(session, user_id)
user_response = UserResponse.model_validate(user)
return success(data=user_response, trace_id=trace_id)
@router.get(
"/users",
response_model=ApiResponse,
summary="List all users",
description="List all users with pagination support",
responses={
200: {"description": "Users retrieved successfully"},
400: ERROR_RESPONSES[400],
500: ERROR_RESPONSES[500],
}
)
async def list_users(
page: int = 1,
page_size: int = 10,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse:
"""List all users with pagination."""
skip = (page - 1) * page_size
users, total = await UserService.list_users(session, skip, page_size)
user_responses = [UserResponse.model_validate(user) for user in users]
return paginated(
items=user_responses,
total=total,
page=page,
page_size=page_size,
trace_id=trace_id
)
@router.put(
"/users/{user_id}",
response_model=ApiResponse[UserResponse],
summary="Update user",
description="Update user information",
responses={
200: {"description": "User updated successfully"},
400: ERROR_RESPONSES[400],
404: ERROR_RESPONSES[404],
422: ERROR_RESPONSES[422],
500: ERROR_RESPONSES[500],
}
)
async def update_user(
user_id: int,
user_data: UserUpdate,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse[UserResponse]:
"""Update user."""
user = await UserService.update_user(session, user_id, user_data)
user_response = UserResponse.model_validate(user)
return success(data=user_response, message="User updated successfully", trace_id=trace_id)
@router.delete(
"/users/{user_id}",
response_model=ApiResponse[None],
summary="Delete user",
description="Delete a user by ID",
responses={
200: {"description": "User deleted successfully"},
404: ERROR_RESPONSES[404],
500: ERROR_RESPONSES[500],
}
)
async def delete_user(
user_id: int,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse[None]:
"""Delete user."""
await UserService.delete_user(session, user_id)
return success(data=None, message="User deleted successfully", trace_id=trace_id)

69
apps/app_a/schemas.py Normal file
View File

@@ -0,0 +1,69 @@
"""
App A Schemas - Pydantic schemas for request/response validation.
"""
from typing import Optional
from datetime import datetime
from pydantic import BaseModel, Field, EmailStr
class UserCreate(BaseModel):
"""Schema for creating a new user."""
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
password: str = Field(..., min_length=6)
full_name: Optional[str] = Field(None, max_length=100)
class Config:
json_schema_extra = {
"example": {
"username": "john_doe",
"email": "john@example.com",
"password": "secret123",
"full_name": "John Doe",
}
}
class UserUpdate(BaseModel):
"""Schema for updating a user."""
email: Optional[EmailStr] = None
full_name: Optional[str] = Field(None, max_length=100)
is_active: Optional[bool] = None
class Config:
json_schema_extra = {
"example": {
"email": "newemail@example.com",
"full_name": "John Smith",
"is_active": True,
}
}
class UserResponse(BaseModel):
"""Schema for user response."""
id: int
username: str
email: str
full_name: Optional[str]
is_active: bool
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
json_schema_extra = {
"example": {
"id": 1,
"username": "john_doe",
"email": "john@example.com",
"full_name": "John Doe",
"is_active": True,
"created_at": "2024-01-15T10:30:00Z",
"updated_at": "2024-01-15T10:30:00Z",
}
}

187
apps/app_a/services.py Normal file
View File

@@ -0,0 +1,187 @@
"""
App A Services - Business logic layer.
"""
from typing import Optional
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from apps.app_a.models import User
from apps.app_a.schemas import UserCreate, UserUpdate
from core.exceptions import NotFoundException, AlreadyExistsException
class UserService:
"""Service for user business logic."""
@staticmethod
async def create_user(
session: AsyncSession,
user_data: UserCreate
) -> User:
"""
Create a new user.
Args:
session: Database session
user_data: User creation data
Returns:
User: Created user
Raises:
AlreadyExistsException: If username or email already exists
"""
# Check if username exists
result = await session.execute(
select(User).where(User.username == user_data.username)
)
if result.scalar_one_or_none():
raise AlreadyExistsException(
message=f"Username '{user_data.username}' already exists",
resource="User"
)
# Check if email exists
result = await session.execute(
select(User).where(User.email == user_data.email)
)
if result.scalar_one_or_none():
raise AlreadyExistsException(
message=f"Email '{user_data.email}' already exists",
resource="User"
)
# Create user (in production, hash the password)
user = User(
username=user_data.username,
email=user_data.email,
full_name=user_data.full_name,
hashed_password=f"hashed_{user_data.password}", # TODO: Use proper hashing
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
session.add(user)
await session.commit()
await session.refresh(user)
return user
@staticmethod
async def get_user(
session: AsyncSession,
user_id: int
) -> User:
"""
Get user by ID.
Args:
session: Database session
user_id: User ID
Returns:
User: User instance
Raises:
NotFoundException: If user not found
"""
result = await session.execute(
select(User).where(User.id == user_id)
)
user = result.scalar_one_or_none()
if not user:
raise NotFoundException(
message=f"User with ID {user_id} not found",
resource="User"
)
return user
@staticmethod
async def list_users(
session: AsyncSession,
skip: int = 0,
limit: int = 100
) -> tuple[list[User], int]:
"""
List all users with pagination.
Args:
session: Database session
skip: Number of records to skip
limit: Maximum number of records to return
Returns:
tuple: (list of users, total count)
"""
# Get total count
count_result = await session.execute(
select(User)
)
total = len(count_result.all())
# Get paginated results
result = await session.execute(
select(User).offset(skip).limit(limit)
)
users = result.scalars().all()
return list(users), total
@staticmethod
async def update_user(
session: AsyncSession,
user_id: int,
user_data: UserUpdate
) -> User:
"""
Update user.
Args:
session: Database session
user_id: User ID
user_data: Update data
Returns:
User: Updated user
Raises:
NotFoundException: If user not found
"""
user = await UserService.get_user(session, user_id)
# Update fields
if user_data.email is not None:
user.email = user_data.email
if user_data.full_name is not None:
user.full_name = user_data.full_name
if user_data.is_active is not None:
user.is_active = user_data.is_active
user.updated_at = datetime.utcnow()
await session.commit()
await session.refresh(user)
return user
@staticmethod
async def delete_user(
session: AsyncSession,
user_id: int
) -> None:
"""
Delete user.
Args:
session: Database session
user_id: User ID
Raises:
NotFoundException: If user not found
"""
user = await UserService.get_user(session, user_id)
await session.delete(user)
await session.commit()

0
apps/app_b/__init__.py Normal file
View File

45
apps/app_b/models.py Normal file
View File

@@ -0,0 +1,45 @@
"""
App B Models - Database models for products.
"""
from typing import Optional
from datetime import datetime
from decimal import Decimal
from sqlmodel import SQLModel, Field
class ProductBase(SQLModel):
"""Base product model with shared fields."""
name: str = Field(index=True, max_length=100)
description: Optional[str] = Field(default=None, max_length=500)
price: Decimal = Field(decimal_places=2)
stock: int = Field(default=0, ge=0)
is_available: bool = Field(default=True)
class Product(ProductBase, table=True):
"""
Product table model.
Represents products in the system.
"""
__tablename__ = "products"
id: Optional[int] = Field(default=None, primary_key=True)
sku: str = Field(unique=True, index=True, max_length=50)
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)
class Config:
json_schema_extra = {
"example": {
"name": "Laptop",
"description": "High-performance laptop",
"price": "999.99",
"stock": 10,
"sku": "LAP-001",
"is_available": True,
}
}

93
apps/app_b/router.py Normal file
View File

@@ -0,0 +1,93 @@
"""
App B Router - API endpoints for product management.
"""
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from core.database import get_session
from core.responses import ApiResponse, success, paginated, ERROR_RESPONSES
from core.dependencies import get_trace_id
from apps.app_b.schemas import ProductCreate, ProductResponse
from apps.app_b.services import ProductService
router = APIRouter(
prefix="/app-b",
tags=["App B - Products"]
)
@router.post(
"/products",
response_model=ApiResponse[ProductResponse],
status_code=201,
summary="Create a new product",
description="Create a new product with SKU, name, price, and stock",
responses={
201: {"description": "Product created successfully"},
400: ERROR_RESPONSES[400],
409: ERROR_RESPONSES[409],
422: ERROR_RESPONSES[422],
500: ERROR_RESPONSES[500],
}
)
async def create_product(
product_data: ProductCreate,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse[ProductResponse]:
"""Create a new product."""
product = await ProductService.create_product(session, product_data)
product_response = ProductResponse.model_validate(product)
return success(data=product_response, message="Product created successfully", trace_id=trace_id)
@router.get(
"/products/{product_id}",
response_model=ApiResponse[ProductResponse],
summary="Get product by ID",
description="Retrieve a product by its ID",
responses={
200: {"description": "Product retrieved successfully"},
404: ERROR_RESPONSES[404],
500: ERROR_RESPONSES[500],
}
)
async def get_product(
product_id: int,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse[ProductResponse]:
"""Get product by ID."""
product = await ProductService.get_product(session, product_id)
product_response = ProductResponse.model_validate(product)
return success(data=product_response, trace_id=trace_id)
@router.get(
"/products",
response_model=ApiResponse,
summary="List all products",
description="List all products with pagination support",
responses={
200: {"description": "Products retrieved successfully"},
400: ERROR_RESPONSES[400],
500: ERROR_RESPONSES[500],
}
)
async def list_products(
page: int = 1,
page_size: int = 10,
session: AsyncSession = Depends(get_session),
trace_id: str = Depends(get_trace_id)
) -> ApiResponse:
"""List all products with pagination."""
skip = (page - 1) * page_size
products, total = await ProductService.list_products(session, skip, page_size)
product_responses = [ProductResponse.model_validate(product) for product in products]
return paginated(
items=product_responses,
total=total,
page=page,
page_size=page_size,
trace_id=trace_id
)

80
apps/app_b/schemas.py Normal file
View File

@@ -0,0 +1,80 @@
"""
App B Schemas - Pydantic schemas for products.
"""
from typing import Optional
from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field
class ProductCreate(BaseModel):
"""Schema for creating a new product."""
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=500)
price: Decimal = Field(..., gt=0, decimal_places=2)
stock: int = Field(default=0, ge=0)
sku: str = Field(..., min_length=1, max_length=50)
is_available: bool = Field(default=True)
class Config:
json_schema_extra = {
"example": {
"name": "Laptop",
"description": "High-performance laptop",
"price": "999.99",
"stock": 10,
"sku": "LAP-001",
"is_available": True,
}
}
class ProductUpdate(BaseModel):
"""Schema for updating a product."""
name: Optional[str] = Field(None, min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=500)
price: Optional[Decimal] = Field(None, gt=0, decimal_places=2)
stock: Optional[int] = Field(None, ge=0)
is_available: Optional[bool] = None
class Config:
json_schema_extra = {
"example": {
"name": "Updated Laptop",
"price": "899.99",
"stock": 15,
}
}
class ProductResponse(BaseModel):
"""Schema for product response."""
id: int
name: str
description: Optional[str]
price: Decimal
stock: int
sku: str
is_available: bool
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
json_schema_extra = {
"example": {
"id": 1,
"name": "Laptop",
"description": "High-performance laptop",
"price": "999.99",
"stock": 10,
"sku": "LAP-001",
"is_available": True,
"created_at": "2024-01-15T10:30:00Z",
"updated_at": "2024-01-15T10:30:00Z",
}
}

78
apps/app_b/services.py Normal file
View File

@@ -0,0 +1,78 @@
"""
App B Services - Business logic for products.
"""
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from apps.app_b.models import Product
from apps.app_b.schemas import ProductCreate, ProductUpdate
from core.exceptions import NotFoundException, AlreadyExistsException
class ProductService:
"""Service for product business logic."""
@staticmethod
async def create_product(
session: AsyncSession,
product_data: ProductCreate
) -> Product:
"""Create a new product."""
# Check if SKU exists
result = await session.execute(
select(Product).where(Product.sku == product_data.sku)
)
if result.scalar_one_or_none():
raise AlreadyExistsException(
message=f"Product with SKU '{product_data.sku}' already exists",
resource="Product"
)
product = Product(
**product_data.model_dump(),
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
session.add(product)
await session.commit()
await session.refresh(product)
return product
@staticmethod
async def get_product(
session: AsyncSession,
product_id: int
) -> Product:
"""Get product by ID."""
result = await session.execute(
select(Product).where(Product.id == product_id)
)
product = result.scalar_one_or_none()
if not product:
raise NotFoundException(
message=f"Product with ID {product_id} not found",
resource="Product"
)
return product
@staticmethod
async def list_products(
session: AsyncSession,
skip: int = 0,
limit: int = 100
) -> tuple[list[Product], int]:
"""List all products with pagination."""
count_result = await session.execute(select(Product))
total = len(count_result.all())
result = await session.execute(
select(Product).offset(skip).limit(limit)
)
products = result.scalars().all()
return list(products), total

0
apps/shared/__init__.py Normal file
View File

0
core/__init__.py Normal file
View File

145
core/config.py Normal file
View File

@@ -0,0 +1,145 @@
"""
Core configuration management using Pydantic Settings.
All configuration loaded from environment variables with validation.
"""
from typing import Literal
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""
Application settings with environment variable support.
Configuration is loaded from environment variables and .env files.
All settings are validated at startup using Pydantic validators.
"""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# Application Settings
app_name: str = Field(default="kami_spider", description="Application name")
environment: Literal["development", "staging", "production"] = Field(
default="development",
description="Runtime environment"
)
debug: bool = Field(default=False, description="Debug mode")
host: str = Field(default="0.0.0.0", description="Server host")
port: int = Field(default=8000, description="Server port")
workers: int = Field(default=1, description="Number of worker processes")
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
default="INFO",
description="Logging level"
)
# Database Settings
db_host: str = Field(default="localhost", description="MySQL host")
db_port: int = Field(default=3306, description="MySQL port")
db_name: str = Field(default="kami_spider", description="Database name")
db_user: str = Field(default="root", description="Database user")
db_password: str = Field(default="", description="Database password")
db_pool_size: int = Field(default=10, description="Database connection pool size")
db_max_overflow: int = Field(default=20, description="Database max overflow connections")
db_pool_recycle: int = Field(default=3600, description="Database pool recycle time in seconds")
db_pool_pre_ping: bool = Field(default=True, description="Test connections before using")
db_echo: bool = Field(default=False, description="Echo SQL statements")
# Redis Settings
redis_host: str = Field(default="localhost", description="Redis host")
redis_port: int = Field(default=6379, description="Redis port")
redis_db: int = Field(default=0, description="Redis database number")
redis_password: str = Field(default="", description="Redis password")
redis_max_connections: int = Field(default=50, description="Redis connection pool max connections")
redis_decode_responses: bool = Field(default=True, description="Decode Redis responses to strings")
# OpenTelemetry Settings
otel_enabled: bool = Field(default=True, description="Enable OpenTelemetry")
otel_service_name: str = Field(default="kami_spider", description="Service name for traces")
otel_exporter_endpoint: str = Field(
default="38.38.251.113:31547",
description="OpenTelemetry collector gRPC endpoint"
)
otel_exporter_insecure: bool = Field(default=True, description="Use insecure gRPC connection")
otel_sample_rate: float = Field(default=1.0, description="Trace sampling rate (0.0 to 1.0)")
# CORS Settings
cors_enabled: bool = Field(default=True, description="Enable CORS")
cors_allow_origins: list[str] = Field(
default=["*"],
description="Allowed CORS origins"
)
cors_allow_credentials: bool = Field(default=True, description="Allow credentials in CORS")
cors_allow_methods: list[str] = Field(
default=["*"],
description="Allowed HTTP methods"
)
cors_allow_headers: list[str] = Field(
default=["*"],
description="Allowed HTTP headers"
)
# Security Settings
secret_key: str = Field(
default="change-me-in-production",
description="Secret key for signing tokens"
)
@field_validator("workers")
@classmethod
def validate_workers(cls, v: int) -> int:
"""Ensure workers is at least 1."""
if v < 1:
raise ValueError("workers must be at least 1")
return v
@field_validator("otel_sample_rate")
@classmethod
def validate_sample_rate(cls, v: float) -> float:
"""Ensure sample rate is between 0.0 and 1.0."""
if not 0.0 <= v <= 1.0:
raise ValueError("otel_sample_rate must be between 0.0 and 1.0")
return v
@property
def database_url(self) -> str:
"""Generate async database URL for SQLModel/SQLAlchemy."""
password_part = f":{self.db_password}" if self.db_password else ""
return (
f"mysql+aiomysql://{self.db_user}{password_part}"
f"@{self.db_host}:{self.db_port}/{self.db_name}"
)
@property
def sync_database_url(self) -> str:
"""Generate sync database URL for Alembic migrations."""
password_part = f":{self.db_password}" if self.db_password else ""
return (
f"mysql+pymysql://{self.db_user}{password_part}"
f"@{self.db_host}:{self.db_port}/{self.db_name}"
)
@property
def redis_url(self) -> str:
"""Generate Redis URL."""
password_part = f":{self.redis_password}@" if self.redis_password else ""
return f"redis://{password_part}{self.redis_host}:{self.redis_port}/{self.redis_db}"
@property
def is_production(self) -> bool:
"""Check if running in production environment."""
return self.environment == "production"
@property
def is_development(self) -> bool:
"""Check if running in development environment."""
return self.environment == "development"
# Global settings instance
settings = Settings()

135
core/database.py Normal file
View File

@@ -0,0 +1,135 @@
"""
Database layer with SQLModel and async SQLAlchemy engine.
Provides connection pooling and session management.
"""
from typing import AsyncGenerator
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.pool import NullPool, AsyncAdaptedQueuePool
from sqlalchemy import text
from sqlmodel import SQLModel
from core.config import settings
# Create async engine with connection pooling
if settings.debug:
# In debug mode, use NullPool (no pooling parameters)
engine = create_async_engine(
settings.database_url,
echo=settings.db_echo,
pool_pre_ping=settings.db_pool_pre_ping,
poolclass=NullPool,
)
else:
# In production mode, use AsyncAdaptedQueuePool with pooling parameters
engine = create_async_engine(
settings.database_url,
echo=settings.db_echo,
pool_size=settings.db_pool_size,
max_overflow=settings.db_max_overflow,
pool_recycle=settings.db_pool_recycle,
pool_pre_ping=settings.db_pool_pre_ping,
poolclass=AsyncAdaptedQueuePool,
)
# Create async session factory
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False,
)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
"""
Dependency that provides async database session.
Usage in FastAPI:
@app.get("/users")
async def get_users(session: AsyncSession = Depends(get_session)):
...
Yields:
AsyncSession: Database session with automatic cleanup
"""
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
@asynccontextmanager
async def get_session_context() -> AsyncGenerator[AsyncSession, None]:
"""
Context manager for database session.
Usage:
async with get_session_context() as session:
result = await session.execute(select(User))
Yields:
AsyncSession: Database session with automatic cleanup
"""
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def create_db_and_tables() -> None:
"""
Create all database tables.
This should be called at application startup in development.
In production, use Alembic migrations instead.
"""
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
async def drop_db_and_tables() -> None:
"""
Drop all database tables.
WARNING: This will delete all data. Use with caution.
Only use in development/testing environments.
"""
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.drop_all)
async def check_database_connection() -> bool:
"""
Check if database connection is healthy.
Returns:
bool: True if connection is healthy, False otherwise
"""
try:
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
return True
except Exception:
return False
async def close_database_connection() -> None:
"""
Close database connection pool.
This should be called at application shutdown.
"""
await engine.dispose()

30
core/dependencies.py Normal file
View File

@@ -0,0 +1,30 @@
"""
Global dependencies for FastAPI dependency injection.
"""
from typing import AsyncGenerator
from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession
from redis.asyncio import Redis
from core.database import get_session
from core.redis import get_redis
async def get_trace_id(request: Request) -> str:
"""
Get trace ID from request state.
Args:
request: FastAPI request object
Returns:
str: Trace ID
"""
return getattr(request.state, "trace_id", "")
__all__ = [
"get_session",
"get_redis",
"get_trace_id",
]

249
core/exceptions.py Normal file
View File

@@ -0,0 +1,249 @@
"""
Custom exception hierarchy for application errors.
Maps exceptions to HTTP status codes and business codes.
"""
from typing import Optional, Any
from core.responses import BusinessCode
class BaseAppException(Exception):
"""
Base exception for all application exceptions.
Attributes:
message: Error message
code: Business error code
status_code: HTTP status code
details: Additional error details
"""
def __init__(
self,
message: str,
code: int = BusinessCode.UNKNOWN_ERROR,
status_code: int = 500,
details: Optional[dict[str, Any]] = None
):
self.message = message
self.code = code
self.status_code = status_code
self.details = details or {}
super().__init__(self.message)
class ValidationException(BaseAppException):
"""
Exception for validation errors.
HTTP Status: 400 Bad Request
Business Code Range: 3000-3999
"""
def __init__(
self,
message: str,
code: int = BusinessCode.INVALID_INPUT,
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=code,
status_code=400,
details=details
)
class NotFoundException(BaseAppException):
"""
Exception for resource not found errors.
HTTP Status: 404 Not Found
Business Code: 4001
"""
def __init__(
self,
message: str,
resource: str = "Resource",
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message or f"{resource} not found",
code=BusinessCode.RESOURCE_NOT_FOUND,
status_code=404,
details=details
)
class ConflictException(BaseAppException):
"""
Exception for resource conflict errors.
HTTP Status: 409 Conflict
Business Code: 4003
"""
def __init__(
self,
message: str,
code: int = BusinessCode.RESOURCE_CONFLICT,
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=code,
status_code=409,
details=details
)
class AlreadyExistsException(BaseAppException):
"""
Exception for resource already exists errors.
HTTP Status: 409 Conflict
Business Code: 4002
"""
def __init__(
self,
message: str,
resource: str = "Resource",
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message or f"{resource} already exists",
code=BusinessCode.RESOURCE_ALREADY_EXISTS,
status_code=409,
details=details
)
class AuthenticationException(BaseAppException):
"""
Exception for authentication errors.
HTTP Status: 401 Unauthorized
Business Code Range: 1001-1099
"""
def __init__(
self,
message: str,
code: int = BusinessCode.LOGIN_FAILED,
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=code,
status_code=401,
details=details
)
class PermissionException(BaseAppException):
"""
Exception for permission/authorization errors.
HTTP Status: 403 Forbidden
Business Code: 1003
"""
def __init__(
self,
message: str,
code: int = BusinessCode.INSUFFICIENT_PERMISSIONS,
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=code,
status_code=403,
details=details
)
class BusinessLogicException(BaseAppException):
"""
Exception for business logic errors.
HTTP Status: 400 Bad Request
Business Code Range: 2000-2999
"""
def __init__(
self,
message: str,
code: int = BusinessCode.OPERATION_NOT_ALLOWED,
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=code,
status_code=400,
details=details
)
class DatabaseException(BaseAppException):
"""
Exception for database errors.
HTTP Status: 503 Service Unavailable
Business Code: 5001
"""
def __init__(
self,
message: str = "Database error occurred",
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=BusinessCode.DATABASE_ERROR,
status_code=503,
details=details
)
class CacheException(BaseAppException):
"""
Exception for cache/Redis errors.
HTTP Status: 503 Service Unavailable
Business Code: 5003
"""
def __init__(
self,
message: str = "Cache service error",
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=BusinessCode.CACHE_ERROR,
status_code=503,
details=details
)
class ExternalServiceException(BaseAppException):
"""
Exception for external service errors.
HTTP Status: 502 Bad Gateway
Business Code: 5002
"""
def __init__(
self,
message: str = "External service unavailable",
details: Optional[dict[str, Any]] = None
):
super().__init__(
message=message,
code=BusinessCode.EXTERNAL_SERVICE_ERROR,
status_code=502,
details=details
)

249
core/redis.py Normal file
View File

@@ -0,0 +1,249 @@
"""
Redis integration with async client and connection pooling.
Provides shared Redis instance for all applications.
"""
from typing import Optional, Any
import json
from redis import asyncio as aioredis
from redis.asyncio import Redis, ConnectionPool
from core.config import settings
# Create Redis connection pool
redis_pool: Optional[ConnectionPool] = None
redis_client: Optional[Redis] = None
async def init_redis() -> Redis:
"""
Initialize Redis client with connection pool.
This should be called at application startup.
Returns:
Redis: Async Redis client
"""
global redis_pool, redis_client
redis_pool = ConnectionPool.from_url(
settings.redis_url,
max_connections=settings.redis_max_connections,
decode_responses=settings.redis_decode_responses,
)
redis_client = Redis(connection_pool=redis_pool)
return redis_client
async def get_redis() -> Redis:
"""
Dependency that provides Redis client.
Usage in FastAPI:
@app.get("/cache")
async def get_cache(redis: Redis = Depends(get_redis)):
...
Returns:
Redis: Async Redis client
Raises:
RuntimeError: If Redis client is not initialized
"""
if redis_client is None:
raise RuntimeError("Redis client not initialized. Call init_redis() at startup.")
return redis_client
async def check_redis_connection() -> bool:
"""
Check if Redis connection is healthy.
Returns:
bool: True if connection is healthy, False otherwise
"""
try:
if redis_client is None:
return False
await redis_client.ping()
return True
except Exception:
return False
async def close_redis_connection() -> None:
"""
Close Redis connection pool.
This should be called at application shutdown.
"""
global redis_pool, redis_client
if redis_client:
await redis_client.aclose()
redis_client = None
if redis_pool:
await redis_pool.disconnect()
redis_pool = None
class RedisCache:
"""
Redis cache utility with common operations.
Provides helper methods for caching with TTL and JSON serialization.
"""
def __init__(self, redis: Redis, prefix: str = ""):
"""
Initialize cache with Redis client and optional key prefix.
Args:
redis: Redis client instance
prefix: Optional key prefix for namespace isolation
"""
self.redis = redis
self.prefix = prefix
def _make_key(self, key: str) -> str:
"""Generate prefixed key."""
return f"{self.prefix}:{key}" if self.prefix else key
async def get(self, key: str) -> Optional[str]:
"""
Get value by key.
Args:
key: Cache key
Returns:
Optional[str]: Cached value or None if not found
"""
return await self.redis.get(self._make_key(key))
async def set(
self,
key: str,
value: str,
ttl: Optional[int] = None
) -> bool:
"""
Set value with optional TTL.
Args:
key: Cache key
value: Value to cache
ttl: Time to live in seconds (optional)
Returns:
bool: True if successful
"""
return await self.redis.set(
self._make_key(key),
value,
ex=ttl
)
async def get_json(self, key: str) -> Optional[Any]:
"""
Get JSON value by key.
Args:
key: Cache key
Returns:
Optional[Any]: Deserialized JSON value or None
"""
value = await self.get(key)
if value is None:
return None
try:
return json.loads(value)
except json.JSONDecodeError:
return None
async def set_json(
self,
key: str,
value: Any,
ttl: Optional[int] = None
) -> bool:
"""
Set JSON value with optional TTL.
Args:
key: Cache key
value: Value to serialize and cache
ttl: Time to live in seconds (optional)
Returns:
bool: True if successful
"""
json_value = json.dumps(value)
return await self.set(key, json_value, ttl)
async def delete(self, key: str) -> int:
"""
Delete key.
Args:
key: Cache key
Returns:
int: Number of keys deleted
"""
return await self.redis.delete(self._make_key(key))
async def exists(self, key: str) -> bool:
"""
Check if key exists.
Args:
key: Cache key
Returns:
bool: True if key exists
"""
return await self.redis.exists(self._make_key(key)) > 0
async def expire(self, key: str, ttl: int) -> bool:
"""
Set TTL for existing key.
Args:
key: Cache key
ttl: Time to live in seconds
Returns:
bool: True if successful
"""
return await self.redis.expire(self._make_key(key), ttl)
async def increment(self, key: str, amount: int = 1) -> int:
"""
Increment numeric value.
Args:
key: Cache key
amount: Increment amount (default 1)
Returns:
int: New value after increment
"""
return await self.redis.incrby(self._make_key(key), amount)
async def decrement(self, key: str, amount: int = 1) -> int:
"""
Decrement numeric value.
Args:
key: Cache key
amount: Decrement amount (default 1)
Returns:
int: New value after decrement
"""
return await self.redis.decrby(self._make_key(key), amount)

392
core/responses.py Normal file
View File

@@ -0,0 +1,392 @@
"""
RESTful response format with generic types.
Provides unified response structure for all API endpoints.
"""
from typing import TypeVar, Generic, Optional, Any
from datetime import datetime
from pydantic import BaseModel, Field
T = TypeVar("T")
class ApiResponse(BaseModel, Generic[T]):
"""
Unified API response structure.
Both success and error responses use this structure.
Attributes:
code: Business status code (0 = success, >0 = error)
message: Human-readable message
data: Response payload (typed) or None for errors
trace_id: Request trace ID for debugging
timestamp: Response timestamp in ISO 8601 format
"""
code: int = Field(description="Business status code (0=success, >0=error)")
message: str = Field(description="Human-readable message")
data: Optional[T] = Field(default=None, description="Response payload")
trace_id: str = Field(description="Request trace ID")
timestamp: datetime = Field(default_factory=datetime.utcnow, description="Response timestamp")
class Config:
json_schema_extra = {
"example": {
"code": 0,
"message": "Success",
"data": {"user_id": 12345, "username": "john_doe"},
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
class PaginationMeta(BaseModel):
"""Pagination metadata."""
total: int = Field(description="Total number of records")
page: int = Field(description="Current page number (1-based)")
page_size: int = Field(description="Number of records per page")
total_pages: int = Field(description="Total number of pages")
class PaginatedData(BaseModel, Generic[T]):
"""
Paginated data structure.
Attributes:
items: List of items for current page
pagination: Pagination metadata
"""
items: list[T] = Field(description="List of items")
pagination: PaginationMeta = Field(description="Pagination metadata")
def success(
data: Optional[T] = None,
message: str = "Success",
trace_id: str = ""
) -> ApiResponse[T]:
"""
Create success response.
Args:
data: Response payload
message: Success message
trace_id: Request trace ID
Returns:
ApiResponse: Success response with code=0
Example:
return success(data={"user_id": 123}, message="User created")
"""
return ApiResponse(
code=0,
message=message,
data=data,
trace_id=trace_id,
timestamp=datetime.utcnow()
)
def error(
code: int,
message: str,
trace_id: str = ""
) -> ApiResponse[None]:
"""
Create error response.
Args:
code: Business error code (must be > 0)
message: Error message
trace_id: Request trace ID
Returns:
ApiResponse: Error response with data=None
Example:
return error(code=1001, message="Login failed: Invalid credentials")
"""
if code <= 0:
raise ValueError("Error code must be greater than 0")
return ApiResponse(
code=code,
message=message,
data=None,
trace_id=trace_id,
timestamp=datetime.utcnow()
)
def paginated(
items: list[T],
total: int,
page: int,
page_size: int,
message: str = "Success",
trace_id: str = ""
) -> ApiResponse[PaginatedData[T]]:
"""
Create paginated response.
Args:
items: List of items for current page
total: Total number of records
page: Current page number (1-based)
page_size: Number of records per page
message: Success message
trace_id: Request trace ID
Returns:
ApiResponse: Paginated response
Example:
return paginated(
items=[user1, user2],
total=100,
page=1,
page_size=10
)
"""
total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
pagination_meta = PaginationMeta(
total=total,
page=page,
page_size=page_size,
total_pages=total_pages
)
paginated_data = PaginatedData(
items=items,
pagination=pagination_meta
)
return ApiResponse(
code=0,
message=message,
data=paginated_data,
trace_id=trace_id,
timestamp=datetime.utcnow()
)
# Business code constants
class BusinessCode:
"""
Business status code definitions.
Code ranges:
0: Success
1000-1999: Authentication & Authorization
2000-2999: Business Logic Errors
3000-3999: Validation Errors
4000-4999: Resource Errors
5000-5999: System Errors
9000-9999: Unknown Errors
"""
# Success
SUCCESS = 0
# Authentication & Authorization (1000-1999)
LOGIN_FAILED = 1001
TOKEN_EXPIRED = 1002
INSUFFICIENT_PERMISSIONS = 1003
INVALID_TOKEN = 1004
# Business Logic Errors (2000-2999)
ORDER_CREATION_FAILED = 2001
PAYMENT_FAILED = 2002
INSUFFICIENT_BALANCE = 2003
OPERATION_NOT_ALLOWED = 2004
# Validation Errors (3000-3999)
INVALID_INPUT = 3001
MISSING_REQUIRED_FIELD = 3002
INVALID_FORMAT = 3003
# Resource Errors (4000-4999)
RESOURCE_NOT_FOUND = 4001
RESOURCE_ALREADY_EXISTS = 4002
RESOURCE_CONFLICT = 4003
# System Errors (5000-5999)
DATABASE_ERROR = 5001
EXTERNAL_SERVICE_ERROR = 5002
CACHE_ERROR = 5003
# Unknown Errors (9000-9999)
UNKNOWN_ERROR = 9000
# Common error response examples for OpenAPI documentation
ERROR_RESPONSES = {
400: {
"description": "Bad Request - Validation or business logic error",
"content": {
"application/json": {
"example": {
"code": 3001,
"message": "Validation error: email - field required",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
401: {
"description": "Unauthorized - Authentication failed",
"content": {
"application/json": {
"example": {
"code": 1001,
"message": "Login failed: Invalid credentials",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
403: {
"description": "Forbidden - Insufficient permissions",
"content": {
"application/json": {
"example": {
"code": 1003,
"message": "Insufficient permissions to perform this action",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
404: {
"description": "Not Found - Resource does not exist",
"content": {
"application/json": {
"example": {
"code": 4001,
"message": "User not found",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
409: {
"description": "Conflict - Resource already exists or conflicts",
"content": {
"application/json": {
"example": {
"code": 4002,
"message": "User already exists",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
422: {
"description": "Unprocessable Entity - Validation error",
"content": {
"application/json": {
"example": {
"code": 3001,
"message": "Validation error: body -> email - value is not a valid email address",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
500: {
"description": "Internal Server Error - Unexpected error",
"content": {
"application/json": {
"example": {
"code": 9000,
"message": "An unexpected error occurred",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
502: {
"description": "Bad Gateway - External service error",
"content": {
"application/json": {
"example": {
"code": 5002,
"message": "External service unavailable",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
503: {
"description": "Service Unavailable - Database or cache error",
"content": {
"application/json": {
"example": {
"code": 5001,
"message": "Database error occurred",
"data": None,
"trace_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
"timestamp": "2024-01-15T10:30:00Z"
}
}
}
},
}
# Error message templates
class ErrorMessage:
"""Predefined error messages for common errors."""
# Authentication
LOGIN_FAILED = "Login failed: Invalid credentials"
TOKEN_EXPIRED = "Authentication token has expired"
INSUFFICIENT_PERMISSIONS = "Insufficient permissions to perform this action"
INVALID_TOKEN = "Invalid authentication token"
# Business Logic
ORDER_CREATION_FAILED = "Order creation failed: {reason}"
PAYMENT_FAILED = "Payment processing failed: {reason}"
INSUFFICIENT_BALANCE = "Insufficient account balance"
OPERATION_NOT_ALLOWED = "Operation not allowed: {reason}"
# Validation
INVALID_INPUT = "Invalid input: {field}"
MISSING_REQUIRED_FIELD = "Missing required field: {field}"
INVALID_FORMAT = "Invalid format: {field}"
# Resources
RESOURCE_NOT_FOUND = "{resource} not found"
RESOURCE_ALREADY_EXISTS = "{resource} already exists"
RESOURCE_CONFLICT = "{resource} conflict: {reason}"
# System
DATABASE_ERROR = "Database error occurred"
EXTERNAL_SERVICE_ERROR = "External service unavailable"
CACHE_ERROR = "Cache service error"
# Unknown
UNKNOWN_ERROR = "An unexpected error occurred"

View File

@@ -0,0 +1,67 @@
# Multi-stage Dockerfile for kami_spider
# Optimized for production deployment with UV package manager
# Stage 1: Builder - Install dependencies
FROM python:3.13-slim AS builder
# Set environment variables
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
UV_COMPILE_BYTECODE=1 \
UV_LINK_MODE=copy
# Install UV
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
# Set working directory
WORKDIR /app
# Copy dependency files
COPY pyproject.toml ./
# Install dependencies using UV
RUN uv sync --frozen --no-dev --no-install-project
# Stage 2: Runtime - Create final image
FROM python:3.13-slim
# Set environment variables
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
PATH="/app/.venv/bin:$PATH"
# Create non-root user
RUN groupadd -r appuser && useradd -r -g appuser appuser
# Set working directory
WORKDIR /app
# Copy virtual environment from builder
COPY --from=builder /app/.venv /app/.venv
# Copy application code
COPY . .
# Change ownership to non-root user
RUN chown -R appuser:appuser /app
# Switch to non-root user
USER appuser
# Expose port
EXPOSE 8000
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"
# Run application with Gunicorn
CMD ["gunicorn", "main:app", \
"--workers", "4", \
"--worker-class", "uvicorn.workers.UvicornWorker", \
"--bind", "0.0.0.0:8000", \
"--timeout", "30", \
"--graceful-timeout", "30", \
"--access-logfile", "-", \
"--error-logfile", "-", \
"--log-level", "info"]

View File

@@ -0,0 +1,109 @@
version: '3.8'
services:
# FastAPI Application
app:
build:
context: ../..
dockerfile: deployments/docker/Dockerfile
image: kami_spider:latest
container_name: kami_spider_app
ports:
- "8000:8000"
environment:
# Application
APP_NAME: kami_spider
ENVIRONMENT: development
DEBUG: "false"
LOG_LEVEL: INFO
# Database
DB_HOST: mysql
DB_PORT: 3306
DB_NAME: kami_spider
DB_USER: kami_user
DB_PASSWORD: kami_pass
DB_POOL_SIZE: 10
DB_MAX_OVERFLOW: 20
# Redis
REDIS_HOST: redis
REDIS_PORT: 6379
REDIS_DB: 0
REDIS_PASSWORD: ""
# OpenTelemetry
OTEL_ENABLED: "true"
OTEL_SERVICE_NAME: kami_spider
OTEL_EXPORTER_ENDPOINT: "38.38.251.113:31547"
OTEL_EXPORTER_INSECURE: "true"
OTEL_SAMPLE_RATE: "1.0"
# CORS
CORS_ENABLED: "true"
CORS_ALLOW_ORIGINS: '["*"]'
depends_on:
mysql:
condition: service_healthy
redis:
condition: service_healthy
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
interval: 30s
timeout: 10s
retries: 3
start_period: 10s
restart: unless-stopped
networks:
- kami_network
# MySQL Database
mysql:
image: mysql:8.0
container_name: kami_spider_mysql
ports:
- "3306:3306"
environment:
MYSQL_ROOT_PASSWORD: root_pass
MYSQL_DATABASE: kami_spider
MYSQL_USER: kami_user
MYSQL_PASSWORD: kami_pass
volumes:
- mysql_data:/var/lib/mysql
healthcheck:
test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
interval: 10s
timeout: 5s
retries: 5
start_period: 30s
restart: unless-stopped
networks:
- kami_network
# Redis Cache
redis:
image: redis:7-alpine
container_name: kami_spider_redis
ports:
- "6379:6379"
volumes:
- redis_data:/data
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 5
start_period: 5s
restart: unless-stopped
networks:
- kami_network
networks:
kami_network:
driver: bridge
volumes:
mysql_data:
driver: local
redis_data:
driver: local

View File

@@ -0,0 +1,25 @@
apiVersion: v1
kind: ConfigMap
metadata:
name: kami-spider-config
namespace: default
data:
APP_NAME: "kami_spider"
ENVIRONMENT: "production"
DEBUG: "false"
LOG_LEVEL: "INFO"
DB_HOST: "kami-spider-mysql"
DB_PORT: "3306"
DB_NAME: "kami_spider"
DB_USER: "kami_user"
DB_POOL_SIZE: "10"
DB_MAX_OVERFLOW: "20"
REDIS_HOST: "kami-spider-redis"
REDIS_PORT: "6379"
REDIS_DB: "0"
OTEL_ENABLED: "true"
OTEL_SERVICE_NAME: "kami_spider"
OTEL_EXPORTER_ENDPOINT: "38.38.251.113:31547"
OTEL_EXPORTER_INSECURE: "true"
OTEL_SAMPLE_RATE: "1.0"
CORS_ENABLED: "true"

View File

@@ -0,0 +1,224 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: kami-spider-app
namespace: default
labels:
app: kami-spider
component: app
spec:
replicas: 3
selector:
matchLabels:
app: kami-spider
component: app
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
template:
metadata:
labels:
app: kami-spider
component: app
spec:
containers:
- name: app
image: kami_spider:latest
imagePullPolicy: IfNotPresent
ports:
- containerPort: 8000
name: http
protocol: TCP
env:
- name: DB_PASSWORD
valueFrom:
secretKeyRef:
name: kami-spider-secret
key: DB_PASSWORD
- name: REDIS_PASSWORD
valueFrom:
secretKeyRef:
name: kami-spider-secret
key: REDIS_PASSWORD
- name: SECRET_KEY
valueFrom:
secretKeyRef:
name: kami-spider-secret
key: SECRET_KEY
envFrom:
- configMapRef:
name: kami-spider-config
resources:
requests:
cpu: 500m
memory: 512Mi
limits:
cpu: 2000m
memory: 2Gi
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 30
timeoutSeconds: 5
failureThreshold: 3
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kami-spider-mysql
namespace: default
labels:
app: kami-spider
component: mysql
spec:
serviceName: kami-spider-mysql
replicas: 1
selector:
matchLabels:
app: kami-spider
component: mysql
template:
metadata:
labels:
app: kami-spider
component: mysql
spec:
containers:
- name: mysql
image: mysql:8.0
ports:
- containerPort: 3306
name: mysql
protocol: TCP
env:
- name: MYSQL_ROOT_PASSWORD
valueFrom:
secretKeyRef:
name: kami-spider-secret
key: MYSQL_ROOT_PASSWORD
- name: MYSQL_DATABASE
value: "kami_spider"
- name: MYSQL_USER
value: "kami_user"
- name: MYSQL_PASSWORD
valueFrom:
secretKeyRef:
name: kami-spider-secret
key: DB_PASSWORD
volumeMounts:
- name: mysql-data
mountPath: /var/lib/mysql
resources:
requests:
cpu: 1000m
memory: 2Gi
limits:
cpu: 4000m
memory: 8Gi
livenessProbe:
exec:
command:
- mysqladmin
- ping
- -h
- localhost
initialDelaySeconds: 30
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 5
readinessProbe:
exec:
command:
- mysqladmin
- ping
- -h
- localhost
initialDelaySeconds: 10
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
volumeClaimTemplates:
- metadata:
name: mysql-data
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: 20Gi
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kami-spider-redis
namespace: default
labels:
app: kami-spider
component: redis
spec:
serviceName: kami-spider-redis
replicas: 1
selector:
matchLabels:
app: kami-spider
component: redis
template:
metadata:
labels:
app: kami-spider
component: redis
spec:
containers:
- name: redis
image: redis:7-alpine
ports:
- containerPort: 6379
name: redis
protocol: TCP
volumeMounts:
- name: redis-data
mountPath: /data
resources:
requests:
cpu: 500m
memory: 512Mi
limits:
cpu: 2000m
memory: 4Gi
livenessProbe:
exec:
command:
- redis-cli
- ping
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
readinessProbe:
exec:
command:
- redis-cli
- ping
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
volumeClaimTemplates:
- metadata:
name: redis-data
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: 5Gi

44
deployments/k8s/hpa.yaml Normal file
View File

@@ -0,0 +1,44 @@
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: kami-spider-hpa
namespace: default
labels:
app: kami-spider
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: kami-spider-app
minReplicas: 3
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
behavior:
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleUp:
stabilizationWindowSeconds: 0
policies:
- type: Percent
value: 100
periodSeconds: 30
- type: Pods
value: 2
periodSeconds: 30
selectPolicy: Max

View File

@@ -0,0 +1,31 @@
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: kami-spider-ingress
namespace: default
labels:
app: kami-spider
annotations:
# Nginx ingress controller annotations
nginx.ingress.kubernetes.io/rewrite-target: /
nginx.ingress.kubernetes.io/ssl-redirect: "false"
# Uncomment for HTTPS/TLS
# cert-manager.io/cluster-issuer: "letsencrypt-prod"
spec:
ingressClassName: nginx
rules:
- host: kami-spider.example.com # Change to your domain
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: kami-spider-app
port:
number: 8000
# Uncomment for TLS
# tls:
# - hosts:
# - kami-spider.example.com
# secretName: kami-spider-tls

View File

@@ -0,0 +1,11 @@
apiVersion: v1
kind: Secret
metadata:
name: kami-spider-secret
namespace: default
type: Opaque
stringData:
DB_PASSWORD: "kami_pass"
REDIS_PASSWORD: ""
SECRET_KEY: "change-me-in-production-use-kubectl-create-secret"
MYSQL_ROOT_PASSWORD: "root_pass"

View File

@@ -0,0 +1,58 @@
apiVersion: v1
kind: Service
metadata:
name: kami-spider-app
namespace: default
labels:
app: kami-spider
component: app
spec:
type: ClusterIP
selector:
app: kami-spider
component: app
ports:
- port: 8000
targetPort: 8000
protocol: TCP
name: http
---
apiVersion: v1
kind: Service
metadata:
name: kami-spider-mysql
namespace: default
labels:
app: kami-spider
component: mysql
spec:
type: ClusterIP
clusterIP: None # Headless service for StatefulSet
selector:
app: kami-spider
component: mysql
ports:
- port: 3306
targetPort: 3306
protocol: TCP
name: mysql
---
apiVersion: v1
kind: Service
metadata:
name: kami-spider-redis
namespace: default
labels:
app: kami-spider
component: redis
spec:
type: ClusterIP
clusterIP: None # Headless service for StatefulSet
selector:
app: kami-spider
component: redis
ports:
- port: 6379
targetPort: 6379
protocol: TCP
name: redis

View File

@@ -0,0 +1,128 @@
version: '3.8'
services:
# FastAPI Application (with replicas for scaling)
app:
image: kami_spider:latest
ports:
- "8000:8000"
environment:
APP_NAME: kami_spider
ENVIRONMENT: production
DEBUG: "false"
LOG_LEVEL: INFO
DB_HOST: mysql
DB_PORT: 3306
DB_NAME: kami_spider
DB_USER: kami_user
DB_PASSWORD: kami_pass
REDIS_HOST: redis
REDIS_PORT: 6379
OTEL_ENABLED: "true"
OTEL_SERVICE_NAME: kami_spider
OTEL_EXPORTER_ENDPOINT: "38.38.251.113:31547"
deploy:
replicas: 3
update_config:
parallelism: 1
delay: 10s
order: start-first
rollback_config:
parallelism: 1
delay: 5s
restart_policy:
condition: on-failure
delay: 5s
max_attempts: 3
resources:
limits:
cpus: '2'
memory: 2G
reservations:
cpus: '0.5'
memory: 512M
placement:
max_replicas_per_node: 1
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
interval: 30s
timeout: 10s
retries: 3
start_period: 10s
networks:
- kami_network
depends_on:
- mysql
- redis
# MySQL Database (single instance)
mysql:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: root_pass
MYSQL_DATABASE: kami_spider
MYSQL_USER: kami_user
MYSQL_PASSWORD: kami_pass
volumes:
- mysql_data:/var/lib/mysql
deploy:
replicas: 1
restart_policy:
condition: on-failure
resources:
limits:
cpus: '4'
memory: 8G
reservations:
cpus: '1'
memory: 2G
placement:
constraints:
- node.role == manager
healthcheck:
test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
interval: 10s
timeout: 5s
retries: 5
start_period: 30s
networks:
- kami_network
# Redis Cache (single instance)
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
deploy:
replicas: 1
restart_policy:
condition: on-failure
resources:
limits:
cpus: '2'
memory: 4G
reservations:
cpus: '0.5'
memory: 512M
placement:
constraints:
- node.role == manager
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 5
start_period: 5s
networks:
- kami_network
networks:
kami_network:
driver: overlay
attachable: true
volumes:
mysql_data:
driver: local
redis_data:
driver: local

184
main.py Normal file
View File

@@ -0,0 +1,184 @@
"""
Main FastAPI application entry point.
Bootstraps the application with middleware, routers, and lifecycle handlers.
"""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from fastapi.middleware.cors import CORSMiddleware
from core.config import settings
from core.database import create_db_and_tables, close_database_connection
from core.redis import init_redis, close_redis_connection
from core.responses import ERROR_RESPONSES
from observability.tracing import init_tracing, instrument_app, shutdown_tracing
from observability.logging import setup_logging, get_logger
from middleware.trace_context import TraceContextMiddleware
from middleware.logging import RequestLoggingMiddleware
from middleware.error_handler import register_exception_handlers
from apps.app_a.router import router as app_a_router
from apps.app_b.router import router as app_b_router
logger = get_logger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Application lifespan manager.
Handles startup and shutdown events.
"""
# Startup
logger.info("Starting application...")
# Setup logging
setup_logging()
logger.info(f"Logging configured: level={settings.log_level}")
# Initialize OpenTelemetry
if settings.otel_enabled:
init_tracing()
instrument_app(app)
logger.info(f"OpenTelemetry initialized: endpoint={settings.otel_exporter_endpoint}")
# Initialize Redis
await init_redis()
logger.info(f"Redis initialized: {settings.redis_host}:{settings.redis_port}")
# Create database tables (development only)
if settings.is_development:
await create_db_and_tables()
logger.info("Database tables created")
logger.info(f"Application started: environment={settings.environment}")
yield
# Shutdown
logger.info("Shutting down application...")
# Close Redis connection
await close_redis_connection()
logger.info("Redis connection closed")
# Close database connection
await close_database_connection()
logger.info("Database connection closed")
# Shutdown OpenTelemetry
if settings.otel_enabled:
await shutdown_tracing()
logger.info("OpenTelemetry shutdown")
logger.info("Application shutdown complete")
# Create FastAPI application
app = FastAPI(
title=settings.app_name,
description="A stateless, production-ready FastAPI web service platform",
version="0.1.0",
debug=settings.debug,
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json",
default_response_class=ORJSONResponse, # Use orjson for better performance
)
# Add CORS middleware
if settings.cors_enabled:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_allow_origins,
allow_credentials=settings.cors_allow_credentials,
allow_methods=settings.cors_allow_methods,
allow_headers=settings.cors_allow_headers,
)
# Add custom middleware (order matters: last added = first executed)
app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(TraceContextMiddleware)
# Register exception handlers
register_exception_handlers(app)
# Include application routers
app.include_router(app_a_router)
app.include_router(app_b_router)
# Health check endpoint
@app.get(
"/health",
tags=["Health"],
summary="Health check",
description="Check the health status of the application and its dependencies",
responses={
200: {"description": "Service is healthy"},
500: ERROR_RESPONSES[500],
503: ERROR_RESPONSES[503],
}
)
async def health_check():
"""
Health check endpoint.
Returns health status of the application and its components.
"""
from core.database import check_database_connection
from core.redis import check_redis_connection
# Check database
db_healthy = await check_database_connection()
# Check Redis
redis_healthy = await check_redis_connection()
# Overall health
healthy = db_healthy and redis_healthy
return {
"status": "healthy" if healthy else "unhealthy",
"components": {
"api": "healthy",
"database": "healthy" if db_healthy else "unhealthy",
"redis": "healthy" if redis_healthy else "unhealthy",
},
"environment": settings.environment,
"version": "0.1.0",
}
# Root endpoint
@app.get(
"/",
tags=["Root"],
summary="Root endpoint",
description="Get API information",
responses={
200: {"description": "API information retrieved successfully"},
}
)
async def root():
"""Root endpoint with API information."""
return {
"name": settings.app_name,
"version": "0.1.0",
"environment": settings.environment,
"docs": "/docs",
"health": "/health",
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host=settings.host,
port=settings.port,
reload=settings.debug,
log_level=settings.log_level.lower(),
)

0
middleware/__init__.py Normal file
View File

219
middleware/error_handler.py Normal file
View File

@@ -0,0 +1,219 @@
"""
Global exception handler middleware.
Catches all exceptions and returns standardized error responses.
"""
import traceback
from typing import Union
from fastapi import Request, status
from fastapi.responses import ORJSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from pydantic import ValidationError
from core.exceptions import BaseAppException
from core.responses import error, BusinessCode, ErrorMessage
from observability.logging import get_logger
logger = get_logger(__name__)
async def app_exception_handler(
request: Request,
exc: BaseAppException
) -> ORJSONResponse:
"""
Handle application-specific exceptions.
Args:
request: FastAPI request
exc: Application exception
Returns:
JSONResponse: Standardized error response
"""
# Get trace ID from request
trace_id = getattr(request.state, "trace_id", "")
# Log exception
logger.error(
f"Application error: {exc.message}",
extra={
"trace_id": trace_id,
"error_code": exc.code,
"details": exc.details,
},
exc_info=True
)
# Return error response
error_response = error(
code=exc.code,
message=exc.message,
trace_id=trace_id
)
return ORJSONResponse(
status_code=exc.status_code,
content=error_response.model_dump(mode='json')
)
async def validation_exception_handler(
request: Request,
exc: Union[RequestValidationError, ValidationError]
) -> ORJSONResponse:
"""
Handle Pydantic validation errors.
Args:
request: FastAPI request
exc: Validation error
Returns:
JSONResponse: Standardized error response
"""
# Get trace ID from request
trace_id = getattr(request.state, "trace_id", "")
# Extract validation errors
errors = exc.errors() if hasattr(exc, "errors") else []
# Format error message
if errors:
first_error = errors[0]
field = ".".join(str(loc) for loc in first_error.get("loc", []))
message = f"Validation error: {field} - {first_error.get('msg', 'Invalid value')}"
else:
message = "Validation error"
# Log validation error
logger.warning(
f"Validation error: {message}",
extra={
"trace_id": trace_id,
"validation_errors": errors,
}
)
# Return error response
error_response = error(
code=BusinessCode.INVALID_INPUT,
message=message,
trace_id=trace_id
)
return ORJSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=error_response.model_dump(mode='json')
)
async def http_exception_handler(
request: Request,
exc: StarletteHTTPException
) -> ORJSONResponse:
"""
Handle HTTP exceptions.
Args:
request: FastAPI request
exc: HTTP exception
Returns:
JSONResponse: Standardized error response
"""
# Get trace ID from request
trace_id = getattr(request.state, "trace_id", "")
# Map HTTP status to business code
code_mapping = {
404: BusinessCode.RESOURCE_NOT_FOUND,
401: BusinessCode.LOGIN_FAILED,
403: BusinessCode.INSUFFICIENT_PERMISSIONS,
409: BusinessCode.RESOURCE_CONFLICT,
}
business_code = code_mapping.get(exc.status_code, BusinessCode.UNKNOWN_ERROR)
# Log HTTP error
logger.warning(
f"HTTP error {exc.status_code}: {exc.detail}",
extra={
"trace_id": trace_id,
"status_code": exc.status_code,
}
)
# Return error response
error_response = error(
code=business_code,
message=str(exc.detail),
trace_id=trace_id
)
return ORJSONResponse(
status_code=exc.status_code,
content=error_response.model_dump(mode='json')
)
async def generic_exception_handler(
request: Request,
exc: Exception
) -> ORJSONResponse:
"""
Handle all other unhandled exceptions.
Args:
request: FastAPI request
exc: Any exception
Returns:
JSONResponse: Standardized error response
"""
# Get trace ID from request
trace_id = getattr(request.state, "trace_id", "")
# Log full exception with traceback
logger.error(
f"Unhandled exception: {str(exc)}",
extra={
"trace_id": trace_id,
"exception_type": type(exc).__name__,
"traceback": traceback.format_exc(),
},
exc_info=True
)
# Return generic error response
error_response = error(
code=BusinessCode.UNKNOWN_ERROR,
message=ErrorMessage.UNKNOWN_ERROR,
trace_id=trace_id
)
return ORJSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content=error_response.model_dump(mode='json')
)
def register_exception_handlers(app) -> None:
"""
Register all exception handlers with FastAPI application.
Args:
app: FastAPI application instance
"""
# Application exceptions
app.add_exception_handler(BaseAppException, app_exception_handler)
# Validation errors
app.add_exception_handler(RequestValidationError, validation_exception_handler)
app.add_exception_handler(ValidationError, validation_exception_handler)
# HTTP exceptions
app.add_exception_handler(StarletteHTTPException, http_exception_handler)
# Generic exception handler (catch-all)
app.add_exception_handler(Exception, generic_exception_handler)

107
middleware/logging.py Normal file
View File

@@ -0,0 +1,107 @@
"""
Request and response logging middleware.
Logs all incoming requests and outgoing responses with trace context.
"""
import time
from typing import Callable
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp
from observability.logging import get_logger
logger = get_logger(__name__)
class RequestLoggingMiddleware(BaseHTTPMiddleware):
"""
Middleware for logging HTTP requests and responses.
Logs:
- Request method, path, client IP
- Response status code, processing time
- Trace ID for correlation
"""
def __init__(self, app: ASGIApp):
"""
Initialize middleware.
Args:
app: ASGI application
"""
super().__init__(app)
async def dispatch(self, request: Request, call_next: Callable) -> Response:
"""
Process request and log details.
Args:
request: Incoming request
call_next: Next middleware or route handler
Returns:
Response: Response from handler
"""
# Record start time
start_time = time.time()
# Get trace ID from request state
trace_id = getattr(request.state, "trace_id", "")
# Get client IP
client_ip = request.client.host if request.client else "unknown"
# Log request
logger.info(
f"Request started: {request.method} {request.url.path}",
extra={
"method": request.method,
"path": request.url.path,
"query_params": str(request.query_params),
"client_ip": client_ip,
"trace_id": trace_id,
}
)
# Process request
try:
response = await call_next(request)
except Exception as e:
# Calculate processing time
process_time = time.time() - start_time
# Log error
logger.error(
f"Request failed: {request.method} {request.url.path} - {str(e)}",
extra={
"method": request.method,
"path": request.url.path,
"client_ip": client_ip,
"trace_id": trace_id,
"process_time_ms": round(process_time * 1000, 2),
},
exc_info=True
)
raise
# Calculate processing time
process_time = time.time() - start_time
# Log response
logger.info(
f"Request completed: {request.method} {request.url.path} - {response.status_code}",
extra={
"method": request.method,
"path": request.url.path,
"status_code": response.status_code,
"client_ip": client_ip,
"trace_id": trace_id,
"process_time_ms": round(process_time * 1000, 2),
}
)
# Add processing time header
response.headers["X-Process-Time"] = f"{process_time:.4f}"
return response

105
middleware/trace_context.py Normal file
View File

@@ -0,0 +1,105 @@
"""
TraceID injection middleware with W3C Trace Context support.
Extracts or generates trace IDs and propagates them through the request lifecycle.
"""
import uuid
from typing import Callable
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp
from observability.logging import set_trace_id
class TraceContextMiddleware(BaseHTTPMiddleware):
"""
Middleware for trace context management.
Extracts trace ID from W3C Trace Context headers or generates a new one.
Adds trace ID to request state and response headers.
"""
def __init__(self, app: ASGIApp):
"""
Initialize middleware.
Args:
app: ASGI application
"""
super().__init__(app)
async def dispatch(self, request: Request, call_next: Callable) -> Response:
"""
Process request and inject trace context.
Args:
request: Incoming request
call_next: Next middleware or route handler
Returns:
Response: Response with trace ID header
"""
# Extract trace ID from W3C Trace Context header (traceparent)
# Format: 00-{trace_id}-{parent_id}-{flags}
trace_id = self._extract_trace_id(request)
# Generate new trace ID if not present
if not trace_id:
trace_id = str(uuid.uuid4())
# Store trace ID in request state
request.state.trace_id = trace_id
# Set trace ID in logging context
set_trace_id(trace_id)
# Process request
response = await call_next(request)
# Add trace ID to response headers
response.headers["X-Trace-ID"] = trace_id
return response
def _extract_trace_id(self, request: Request) -> str:
"""
Extract trace ID from request headers.
Supports:
- W3C Trace Context (traceparent header)
- X-Trace-ID custom header
Args:
request: Incoming request
Returns:
str: Trace ID or empty string if not found
"""
# Try X-Trace-ID header first (custom header)
trace_id = request.headers.get("X-Trace-ID", "")
if trace_id:
return trace_id
# Try W3C Trace Context traceparent header
# Format: 00-{trace_id}-{parent_id}-{flags}
traceparent = request.headers.get("traceparent", "")
if traceparent:
parts = traceparent.split("-")
if len(parts) >= 2:
# Extract trace_id (second part)
return parts[1]
return ""
def get_trace_id_from_request(request: Request) -> str:
"""
Get trace ID from request state.
Args:
request: FastAPI request
Returns:
str: Trace ID or empty string
"""
return getattr(request.state, "trace_id", "")

View File

187
observability/logging.py Normal file
View File

@@ -0,0 +1,187 @@
"""
Structured logging with trace context propagation.
Integrates with OpenTelemetry for log correlation.
"""
import logging
import json
from typing import Any, MutableMapping
from datetime import datetime
from contextvars import ContextVar
from core.config import settings
# Context variable for trace ID
trace_id_var: ContextVar[str] = ContextVar("trace_id", default="")
class TraceContextFormatter(logging.Formatter):
"""
Custom formatter that adds trace context to log records.
Formats logs as JSON in production, human-readable in development.
"""
def __init__(self, use_json: bool = True):
"""
Initialize formatter.
Args:
use_json: Whether to format as JSON (True) or human-readable (False)
"""
super().__init__()
self.use_json = use_json
def format(self, record: logging.LogRecord) -> str:
"""
Format log record with trace context.
Args:
record: Log record
Returns:
str: Formatted log message
"""
# Get trace ID from context
trace_id = trace_id_var.get()
if self.use_json:
# JSON format for production
log_data = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"trace_id": trace_id,
"module": record.module,
"function": record.funcName,
"line": record.lineno,
}
# Add exception info if present
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
# Add extra fields
if hasattr(record, "extra"):
log_data["extra"] = getattr(record, "extra")
return json.dumps(log_data)
else:
# Human-readable format for development
timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
trace_part = f" [trace_id={trace_id}]" if trace_id else ""
message = f"{timestamp} - {record.levelname:8s} - {record.name:30s}{trace_part} - {record.getMessage()}"
if record.exc_info:
message += "\n" + self.formatException(record.exc_info)
return message
def setup_logging() -> None:
"""
Configure application logging.
This should be called at application startup.
"""
# Determine if we should use JSON format
use_json = settings.is_production
# Create formatter
formatter = TraceContextFormatter(use_json=use_json)
# Configure root logger
root_logger = logging.getLogger()
root_logger.setLevel(settings.log_level)
# Remove existing handlers
root_logger.handlers.clear()
# Create console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(settings.log_level)
console_handler.setFormatter(formatter)
# Add handler to root logger
root_logger.addHandler(console_handler)
# Set levels for third-party loggers
logging.getLogger("uvicorn").setLevel(logging.INFO)
logging.getLogger("uvicorn.access").setLevel(logging.INFO)
logging.getLogger("uvicorn.error").setLevel(logging.INFO)
logging.getLogger("fastapi").setLevel(logging.INFO)
# Reduce noise from OpenTelemetry
logging.getLogger("opentelemetry").setLevel(logging.WARNING)
def set_trace_id(trace_id: str) -> None:
"""
Set trace ID in context for current request.
Args:
trace_id: Trace ID to set
"""
trace_id_var.set(trace_id)
def get_trace_id() -> str:
"""
Get trace ID from context.
Returns:
str: Current trace ID or empty string
"""
return trace_id_var.get()
def get_logger(name: str) -> logging.Logger:
"""
Get logger instance.
Args:
name: Logger name (usually __name__)
Returns:
logging.Logger: Logger instance
"""
return logging.getLogger(name)
class LoggerAdapter(logging.LoggerAdapter):
"""
Logger adapter that automatically includes trace context.
"""
def process(self, msg: str, kwargs: MutableMapping[str, Any]) -> tuple[str, MutableMapping[str, Any]]:
"""
Process log message to add trace context.
Args:
msg: Log message
kwargs: Keyword arguments
Returns:
tuple: Processed message and kwargs
"""
trace_id = trace_id_var.get()
if trace_id:
extra = kwargs.get("extra", {})
extra["trace_id"] = trace_id
kwargs["extra"] = extra
return msg, kwargs
def get_logger_with_trace(name: str) -> LoggerAdapter:
"""
Get logger adapter with automatic trace context.
Args:
name: Logger name (usually __name__)
Returns:
LoggerAdapter: Logger adapter instance
"""
logger = logging.getLogger(name)
return LoggerAdapter(logger, {})

163
observability/tracing.py Normal file
View File

@@ -0,0 +1,163 @@
"""
OpenTelemetry tracing configuration with gRPC exporter.
Provides distributed tracing for the application.
"""
from typing import Optional
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, SpanProcessor
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.redis import RedisInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
from core.config import settings
tracer_provider: Optional[TracerProvider] = None
span_processor: Optional[SpanProcessor] = None
def init_tracing() -> trace.Tracer:
"""
Initialize OpenTelemetry tracing with gRPC exporter.
This should be called at application startup.
Returns:
trace.Tracer: Configured tracer instance
"""
global tracer_provider, span_processor
if not settings.otel_enabled:
# Return no-op tracer if OpenTelemetry is disabled
return trace.get_tracer(__name__)
# Create resource with service name
resource = Resource(attributes={
SERVICE_NAME: settings.otel_service_name
})
# Create sampler based on sample rate
sampler = TraceIdRatioBased(settings.otel_sample_rate)
# Create tracer provider
tracer_provider = TracerProvider(
resource=resource,
sampler=sampler
)
# Create OTLP gRPC exporter
otlp_exporter = OTLPSpanExporter(
endpoint=settings.otel_exporter_endpoint,
insecure=settings.otel_exporter_insecure
)
# Create batch span processor
span_processor = BatchSpanProcessor(otlp_exporter)
tracer_provider.add_span_processor(span_processor)
# Set global tracer provider
trace.set_tracer_provider(tracer_provider)
# Return tracer
return trace.get_tracer(__name__)
def instrument_app(app) -> None:
"""
Instrument FastAPI application with OpenTelemetry.
Args:
app: FastAPI application instance
"""
if not settings.otel_enabled:
return
# Instrument FastAPI
FastAPIInstrumentor.instrument_app(app)
# Instrument HTTP client
HTTPXClientInstrumentor().instrument()
# Redis instrumentation is done when Redis client is created
# SQLAlchemy instrumentation is done when engine is created
def instrument_sqlalchemy(engine) -> None:
"""
Instrument SQLAlchemy engine with OpenTelemetry.
Args:
engine: SQLAlchemy engine instance
"""
if not settings.otel_enabled:
return
SQLAlchemyInstrumentor().instrument(
engine=engine.sync_engine
)
def instrument_redis() -> None:
"""
Instrument Redis client with OpenTelemetry.
"""
if not settings.otel_enabled:
return
RedisInstrumentor().instrument()
async def shutdown_tracing() -> None:
"""
Shutdown tracing and flush remaining spans.
This should be called at application shutdown.
"""
global tracer_provider, span_processor
if span_processor:
span_processor.shutdown()
if tracer_provider:
tracer_provider.shutdown()
def get_tracer(name: str = __name__) -> trace.Tracer:
"""
Get tracer instance.
Args:
name: Tracer name (usually __name__)
Returns:
trace.Tracer: Tracer instance
"""
return trace.get_tracer(name)
def get_current_span() -> trace.Span:
"""
Get current active span.
Returns:
trace.Span: Current span
"""
return trace.get_current_span()
def get_trace_id() -> str:
"""
Get current trace ID as hex string.
Returns:
str: Trace ID or empty string if no active span
"""
span = get_current_span()
if span and span.get_span_context().is_valid:
return format(span.get_span_context().trace_id, '032x')
return ""

120
pyproject.toml Normal file
View File

@@ -0,0 +1,120 @@
[project]
name = "kami_spider"
version = "0.1.0"
description = "A stateless, production-ready FastAPI-based web service platform hosting multiple independent web applications"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"fastapi>=0.120.0",
"uvicorn[standard]>=0.38.0",
"gunicorn>=23.0.0",
"sqlmodel>=0.0.24",
"pydantic>=2.12.3",
"pydantic-settings>=2.11.0",
"redis>=5.2.1",
"pymysql>=1.1.1",
"cryptography>=46.0.1",
"aiomysql>=0.2.0",
"alembic>=1.14.0",
"python-dotenv>=1.0.1",
"opentelemetry-api>=1.38.0",
"opentelemetry-sdk>=1.38.0",
"opentelemetry-instrumentation-fastapi>=0.59b0",
"opentelemetry-instrumentation-sqlalchemy>=0.59b0",
"opentelemetry-instrumentation-redis>=0.59b0",
"opentelemetry-instrumentation-httpx>=0.59b0",
"opentelemetry-exporter-otlp-proto-grpc>=1.38.0",
"httpx>=0.28.1",
"python-multipart>=0.0.20",
"email-validator>=2.3.0",
"greenlet>=3.2.4",
"orjson>=3.11.4",
]
[project.optional-dependencies]
dev = [
"pytest>=8.3.4",
"pytest-asyncio>=0.24.0",
"pytest-cov>=6.0.0",
"pytest-mock>=3.14.0",
"httpx>=0.28.1",
"ruff>=0.8.4",
"mypy>=1.14.1",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["apps", "core", "middleware", "observability"]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
asyncio_mode = "auto"
addopts = [
"--strict-markers",
"--strict-config",
"--cov=.",
"--cov-report=term-missing",
"--cov-report=html",
]
[tool.coverage.run]
source = ["."]
omit = [
"tests/*",
"*/site-packages/*",
"*/__pycache__/*",
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise AssertionError",
"raise NotImplementedError",
"if __name__ == .__main__.:",
"if TYPE_CHECKING:",
"class .*\\bProtocol\\):",
"@(abc\\.)?abstractmethod",
]
[tool.ruff]
line-length = 120
target-version = "py313"
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"UP", # pyupgrade
]
ignore = [
"E501", # line too long (handled by formatter)
"B008", # do not perform function calls in argument defaults
"W191", # indentation contains tabs
]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]
[tool.mypy]
python_version = "3.13"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
strict_equality = true

0
tests/__init__.py Normal file
View File

View File

0
tests/unit/__init__.py Normal file
View File

1407
uv.lock generated Normal file

File diff suppressed because it is too large Load Diff