Skip to content

Latest commit

 

History

History
189 lines (152 loc) · 6.04 KB

File metadata and controls

189 lines (152 loc) · 6.04 KB

FastAPI Enterprise Microservice Demo

A production-ready FastAPI microservice demonstrating modern Python backend architecture patterns, async operations, and enterprise-grade features.

🏗️ Architecture Highlights

  • Async/Await Patterns: Full async implementation for high performance
  • Dependency Injection: Clean separation of concerns
  • Health Checks: Liveness and readiness probes for Kubernetes/ECS
  • Error Handling: Comprehensive exception handling with structured responses
  • Database Integration: MongoDB with connection pooling
  • CORS Configuration: Production-ready CORS setup
  • Logging: Structured logging with request tracing
  • Testing: Comprehensive test coverage with async test patterns

🚀 Features

  • Employee Management API: CRUD operations with validation
  • Draft System: Temporary data persistence with auto-cleanup
  • Admin Functions: Role-based access patterns
  • Health Monitoring: Built-in health check endpoints
  • Request Validation: Pydantic models for type safety
  • Database Migrations: Async database initialization

📁 Project Structure

fastapi-microservice-demo/
├── app/
│   ├── main.py              # FastAPI app initialization
│   ├── dependencies.py      # Dependency injection
│   └── middleware.py        # Custom middleware
├── config/
│   ├── settings.py          # Configuration management
│   └── database.py          # Database configuration
├── models/
│   ├── employee.py          # Pydantic models
│   └── responses.py         # API response models
├── routers/
│   ├── employees.py         # Employee endpoints
│   ├── admin.py            # Admin endpoints
│   └── health.py           # Health check endpoints
├── utils/
│   ├── database.py         # Database utilities
│   ├── encryption.py       # Data security utilities
│   └── validators.py       # Custom validators
├── tests/
│   ├── test_endpoints.py   # API endpoint tests
│   └── test_models.py      # Model validation tests
├── docker-compose.yml      # Local development setup
├── Dockerfile             # Container configuration
└── requirements.txt       # Python dependencies

🛠️ Technology Stack

  • Framework: FastAPI 0.104+
  • Database: MongoDB with Motor (async driver)
  • Validation: Pydantic v2
  • Testing: Pytest with async support
  • Containerization: Docker & Docker Compose
  • Security: JWT authentication, data encryption
  • Monitoring: Structured logging, health checks

🔧 Key Patterns Demonstrated

1. Async Database Operations

async def get_employee(employee_id: str) -> Optional[Employee]:
    db = await get_database()
    employee_data = await db.employees.find_one({"_id": ObjectId(employee_id)})
    return Employee(**employee_data) if employee_data else None

2. Dependency Injection

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    # Token validation and user retrieval
    pass

@router.get("/employees/{employee_id}")
async def get_employee_endpoint(
    employee_id: str,
    current_user: User = Depends(get_current_user),
    db = Depends(get_database)
):
    # Endpoint logic here
    pass

3. Health Check Implementation

@app.get("/health/ready")
async def health_ready():
    try:
        await db.command("ping")
        return {"status": "ready", "timestamp": datetime.utcnow()}
    except Exception as e:
        raise HTTPException(status_code=503, detail="Database unavailable")

4. Structured Error Handling

@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_error",
            "details": exc.errors(),
            "request_id": request.headers.get("X-Request-ID")
        }
    )

🚀 Quick Start

# Clone and setup
git clone <repository-url>
cd fastapi-microservice-demo

# Install dependencies
pip install -r requirements.txt

# Start with Docker Compose
docker-compose up -d

# Run locally
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

🧪 Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=app --cov-report=html

# Run async tests
pytest -v tests/test_async_endpoints.py

📊 Performance Features

  • Connection Pooling: Efficient database connection management
  • Async Operations: Non-blocking I/O for high concurrency
  • Response Caching: Strategic caching for frequently accessed data
  • Request Validation: Early validation to reduce processing overhead
  • Health Monitoring: Proactive health checks for container orchestration

🔒 Security Features

  • Input Validation: Comprehensive request validation with Pydantic
  • SQL Injection Prevention: Parameterized queries and ORM usage
  • Data Encryption: Sensitive data encryption at rest
  • CORS Configuration: Secure cross-origin resource sharing
  • Authentication: JWT-based authentication with refresh tokens

📈 Monitoring & Observability

  • Structured Logging: JSON-formatted logs with correlation IDs
  • Health Endpoints: Kubernetes/ECS compatible health checks
  • Metrics Collection: Request timing and error rate tracking
  • Request Tracing: End-to-end request tracking capabilities

🔧 Configuration Management

Environment-based configuration with validation:

class Settings(BaseSettings):
    app_name: str = "Employee Management API"
    database_url: str
    secret_key: str
    allowed_origins: List[str] = ["http://localhost:3000"]
    log_level: str = "INFO"
    
    class Config:
        env_file = ".env"

📝 Note

This is a sanitized demonstration repository. All sensitive data, credentials, and proprietary business logic have been removed. The focus is on showcasing architecture patterns, code quality, and technical implementation approaches.