Module 2: Advanced Python & FastAPI
Master Python fundamentals and build production ML APIs
Overview
Module 2 provides the Python foundation and API development skills essential for MLOps engineering. You'll master advanced Python concepts, then apply them to build production-ready ML APIs with FastAPI.
Learning Objectives
By the end of this module, you will be able to:
- Write type-safe Python code with comprehensive type hints
- Manage Python projects efficiently with modern tooling (uv)
- Validate data robustly with Pydantic models
- Apply decorator patterns for cleaner, more maintainable code
- Build concurrent applications with async/await
- Create production-ready REST APIs with FastAPI
- Integrate databases with SQLAlchemy and Alembic
- Serve ML models with streaming capabilities
- Test APIs comprehensively
- Deploy APIs to production
Module Structure
This module consists of two major sections:
Part A: Advanced Python Foundations
Build the Python skills needed for modern MLOps development
Part B: FastAPI for ML APIs
Apply Python skills to build production ML serving infrastructure
Prerequisites
- Python Experience: Intermediate Python (functions, classes, modules)
- Command Line: Comfortable with terminal/command prompt
- Development Environment: Python 3.10+ installed
- Module 1: Git fundamentals (helpful for collaboration)
Study Path
Part A: Advanced Python
Core Concepts:
- Type annotations for better code quality
- Generic types and protocols
- Type checker integration (mypy/pyright)
- Why: Catch bugs early, improve IDE support
- Modern Python project setup
- Dependency management and lock files
- Development vs production dependencies
- Why: 10-100x faster than pip, reproducible builds
- Robust data models with automatic validation
- Custom validators and constraints
- JSON serialization/deserialization
- Why: Powers FastAPI, prevents runtime errors
- Function and class decorators
- Practical patterns (logging, caching, retry)
- Decorator composition
- Why: Reduce boilerplate, separation of concerns
- Asynchronous programming fundamentals
- Concurrent task execution
- Async HTTP and file I/O
- Why: Essential for high-performance APIs
Study Guide:
- Week 1: Types and Tools (Typing, uv)
- Week 2: Validation and Patterns (Pydantic, Decorators)
- Week 3: Async Programming (async/await)
Part B: FastAPI for ML APIs
Building Production APIs:
- Core FastAPI concepts and patterns
- Request/response models with Pydantic
- Dependency injection system
- Automatic API documentation
- File uploads for ML models and datasets
- Form data and multipart requests
- Background tasks for async processing
- Headers, cookies, and custom responses
API Architecture & Database Integration
- Scalable project structure
- SQLAlchemy 2.0 async ORM
- Alembic database migrations
- Middleware and configuration management
- Model loading and caching strategies
- Batch and streaming predictions
- Server-Sent Events (SSE) for progress updates
- WebSocket for real-time predictions
- Streaming LLM responses token-by-token
- Comprehensive API testing with pytest
- Testing async endpoints
- Database testing and fixtures
- Mocking dependencies
- Integration and load testing
- Uvicorn and Gunicorn configuration
- Docker containerization
- Health checks and monitoring
- Security best practices
- Performance optimization
Study Guide:
- Week 4: Core FastAPI (Fundamentals, Requests)
- Week 5: Architecture & Serving (Database, ML Models, Streaming)
- Week 6: Testing & Deployment (Tests, Production)
Common Patterns
Pattern 1: Type-Safe Configuration
from pydantic import BaseModel, Field
class AppConfig(BaseModel):
"""Application configuration with validation."""
database_url: str
api_key: str
max_workers: int = Field(ge=1, le=100)
debug: bool = False
config = AppConfig(**env_vars)Pattern 2: Async Model Serving
from fastapi import FastAPI, Depends
from typing import Annotated
app = FastAPI()
async def get_model():
"""Load model as dependency."""
return await load_model_async()
@app.post("/predict")
async def predict(
data: PredictionInput,
model: Annotated[Model, Depends(get_model)]
):
"""Async prediction endpoint."""
result = await model.predict_async(data.features)
return {"prediction": result}Pattern 3: Streaming Responses
from fastapi.responses import StreamingResponse
import asyncio
async def generate_predictions(data_stream):
"""Stream predictions as they're computed."""
for batch in data_stream:
result = await predict(batch)
yield f"data: {result}\n\n"
await asyncio.sleep(0.1)
@app.post("/predict/stream")
async def stream_predictions(data: list):
"""Server-sent events for real-time predictions."""
return StreamingResponse(
generate_predictions(data),
media_type="text/event-stream"
)Tools and Setup
Required Tools
# Python 3.10+
python --version
# Install uv (project management)
curl -LsSf https://astral.sh/uv/install.sh | sh
# Install PostgreSQL (for database practice)
# Docker recommended for easy setup
docker run -d \
--name postgres-mlops \
-e POSTGRES_PASSWORD=password \
-p 5432:5432 \
postgres:15-alpineDevelopment Environment
# Create new project
uv init ml-api
cd ml-api
# Add core dependencies
uv add fastapi uvicorn[standard] sqlalchemy[asyncio] alembic pydantic
# Add dev dependencies
uv add --dev pytest pytest-asyncio httpx mypy ruff
# Run development server
uv run uvicorn src.main:app --reloadAssessment Criteria
Demonstrate mastery by:
Code Quality
- [ ] All code is fully type-hinted
- [ ] Pydantic models for all data structures
- [ ] Proper async/await usage
- [ ] Clean decorator implementation
API Development
- [ ] RESTful API design principles
- [ ] Proper error handling
- [ ] Database integration working
- [ ] Streaming endpoints functional
Testing
- [ ] > 80% test coverage
- [ ] All endpoints tested
- [ ] Database operations tested
- [ ] Integration tests passing
Production Readiness
- [ ] Docker containerization
- [ ] Health checks implemented
- [ ] Logging and monitoring
- [ ] Security best practices
Learning Resources
Documentation
Video Tutorials
Practice Platforms
- Real Python - Python tutorials
- TestDriven.io - FastAPI projects
Troubleshooting
Type Checker Issues
# Run mypy on your code
mypy src/ --strict
# Common fixes
# - Add return type hints to all functions
# - Use Optional[T] for nullable values
# - Import types from typing moduleAsync Errors
# Wrong: Missing await
result = async_function() # Returns coroutine!
# Correct: Use await
result = await async_function()
# Wrong: Blocking call in async
def blocking_call():
time.sleep(5)
# Correct: Use asyncio.to_thread
await asyncio.to_thread(blocking_call)Database Connection Issues
# Check PostgreSQL is running
docker ps | grep postgres
# Test connection
psql postgresql://user:pass@localhost:5432/dbname
# Run Alembic migrations
alembic upgrade headNext Steps
After completing Module 2:
Module 3: CI/CD & Deployment
- Automated testing pipelines
- Container orchestration with Kubernetes
- Monitoring and observability
Build Your Portfolio
- Deploy your ML API to production
- Add to GitHub with comprehensive README
- Share with the community
Advanced Topics
- GraphQL with Strawberry
- gRPC for high-performance APIs
- Event-driven architectures with Kafka
Community and Support
- Questions: Open an issue in the repository
- Discussions: Join the MLOps training community
- Contributions: Submit improvements via pull requests
Quick Navigation
Start Learning
- Advanced Python → - Begin with Python foundations
- FastAPI → - Jump to API development
Reference
- Main Study Guide - Overall training navigation
- Module 1: Infrastructure - Previous module
- Module 3: Deployment - Next module
Ready to start? Begin with Advanced Python to build your Python foundations, then progress to FastAPI for production API development.