2adbcd0535
- Implemented comprehensive multi-tenant data isolation using database-level security - Built JWT authentication system with role-based access control (Super Admin, Org Admin, User, Viewer) - Created RESTful API endpoints for user and organization operations - Added complete audit logging for all data modifications with IP tracking - Implemented API rate limiting and input validation with security middleware - Built webhook processing engine with async event handling and retry logic - Created external API call handlers with circuit breaker pattern and error handling - Implemented data synchronization between external services and internal data - Added integration health monitoring and status tracking - Created three mock external services (User Management, Payment, Communication) - Implemented idempotency for webhook processing to handle duplicates gracefully - Added comprehensive security headers and XSS/CSRF protection - Set up Alembic database migrations with proper SQLite configuration - Included extensive documentation and API examples Architecture features: - Multi-tenant isolation at database level - Circuit breaker pattern for external API resilience - Async background task processing - Complete audit trail with user context - Role-based permission system - Webhook signature verification - Request validation and sanitization - Health monitoring endpoints Co-Authored-By: Claude <noreply@anthropic.com>
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import List, Dict, Any, Optional
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
app = FastAPI(title="Mock User Management Service", version="1.0.0")
|
|
|
|
# Mock data storage
|
|
users_db: Dict[str, Dict[str, Any]] = {}
|
|
organizations_db: Dict[int, List[str]] = {}
|
|
|
|
class User(BaseModel):
|
|
id: Optional[str] = None
|
|
email: str
|
|
username: str
|
|
first_name: Optional[str] = None
|
|
last_name: Optional[str] = None
|
|
organization_id: int
|
|
is_active: bool = True
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
class WebhookEvent(BaseModel):
|
|
event_id: str
|
|
event_type: str
|
|
timestamp: datetime
|
|
data: Dict[str, Any]
|
|
|
|
@app.get("/health")
|
|
async def health_check():
|
|
return {"status": "healthy", "service": "user_management", "timestamp": datetime.utcnow()}
|
|
|
|
@app.post("/users", response_model=User)
|
|
async def create_user(user: User):
|
|
user_id = str(uuid.uuid4())
|
|
user.id = user_id
|
|
user.created_at = datetime.utcnow()
|
|
|
|
users_db[user_id] = user.dict()
|
|
|
|
# Add to organization
|
|
if user.organization_id not in organizations_db:
|
|
organizations_db[user.organization_id] = []
|
|
organizations_db[user.organization_id].append(user_id)
|
|
|
|
# Send webhook (simulated)
|
|
await send_webhook("user.created", user.dict())
|
|
|
|
return user
|
|
|
|
@app.get("/users/{user_id}", response_model=User)
|
|
async def get_user(user_id: str):
|
|
if user_id not in users_db:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return User(**users_db[user_id])
|
|
|
|
@app.put("/users/{user_id}", response_model=User)
|
|
async def update_user(user_id: str, user_update: Dict[str, Any]):
|
|
if user_id not in users_db:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user_data = users_db[user_id].copy()
|
|
user_data.update(user_update)
|
|
user_data["updated_at"] = datetime.utcnow()
|
|
users_db[user_id] = user_data
|
|
|
|
# Send webhook (simulated)
|
|
await send_webhook("user.updated", user_data)
|
|
|
|
return User(**user_data)
|
|
|
|
@app.delete("/users/{user_id}")
|
|
async def delete_user(user_id: str):
|
|
if user_id not in users_db:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user_data = users_db.pop(user_id)
|
|
|
|
# Remove from organization
|
|
org_id = user_data["organization_id"]
|
|
if org_id in organizations_db and user_id in organizations_db[org_id]:
|
|
organizations_db[org_id].remove(user_id)
|
|
|
|
# Send webhook (simulated)
|
|
await send_webhook("user.deleted", {"user_id": user_id, "organization_id": org_id})
|
|
|
|
return {"message": "User deleted successfully"}
|
|
|
|
@app.get("/organizations/{organization_id}/users")
|
|
async def get_organization_users(organization_id: int):
|
|
if organization_id not in organizations_db:
|
|
return []
|
|
|
|
org_users = []
|
|
for user_id in organizations_db[organization_id]:
|
|
if user_id in users_db:
|
|
org_users.append(users_db[user_id])
|
|
|
|
return org_users
|
|
|
|
async def send_webhook(event_type: str, data: Dict[str, Any]):
|
|
"""Simulate sending webhook to main service"""
|
|
webhook_data = {
|
|
"event_id": str(uuid.uuid4()),
|
|
"event_type": event_type,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"data": data
|
|
}
|
|
|
|
# In a real implementation, this would send HTTP request to main service
|
|
print(f"Webhook sent: {event_type} - {webhook_data['event_id']}")
|
|
|
|
return webhook_data
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8001) |