python
examples
examples.py🐍python
"""
Docker and Deployment - Examples
Learn to containerize Python applications and deploy to production.
"""
from typing import Optional
# =============================================================================
# DOCKERFILE EXAMPLES
# =============================================================================
BASIC_DOCKERFILE = '''
# Basic Python Dockerfile
FROM python:3.12-slim
# Set working directory
WORKDIR /app
# Copy requirements first (for better caching)
COPY requirements.txt .
# Install dependencies
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Expose port
EXPOSE 8000
# Run the application
CMD ["python", "app.py"]
'''
PRODUCTION_DOCKERFILE = '''
# Production-ready multi-stage Dockerfile
# Stage 1: Build
FROM python:3.12-slim as builder
WORKDIR /app
# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends build-essential && rm -rf /var/lib/apt/lists/*
# Create virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Stage 2: Production
FROM python:3.12-slim as production
# Create non-root user
RUN groupadd -r appuser && useradd -r -g appuser appuser
WORKDIR /app
# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Copy application code
COPY --chown=appuser:appuser . .
# Switch to non-root user
USER appuser
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 CMD curl -f http://localhost:8000/health || exit 1
EXPOSE 8000
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "app:app"]
'''
FASTAPI_DOCKERFILE = '''
# FastAPI with uvicorn
FROM python:3.12-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Run with uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
'''
# =============================================================================
# DOCKER COMPOSE EXAMPLES
# =============================================================================
DOCKER_COMPOSE_BASIC = '''
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/myapp
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
db:
image: postgres:15
environment:
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
POSTGRES_DB: myapp
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
volumes:
postgres_data:
'''
DOCKER_COMPOSE_PRODUCTION = '''
version: '3.8'
services:
web:
build:
context: .
dockerfile: Dockerfile.prod
ports:
- "8000:8000"
environment:
- DATABASE_URL=${DATABASE_URL}
- SECRET_KEY=${SECRET_KEY}
- DEBUG=false
depends_on:
db:
condition: service_healthy
deploy:
replicas: 3
restart_policy:
condition: on-failure
max_attempts: 3
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
db:
image: postgres:15
environment:
POSTGRES_USER: ${DB_USER}
POSTGRES_PASSWORD: ${DB_PASSWORD}
POSTGRES_DB: ${DB_NAME}
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U ${DB_USER}"]
interval: 10s
timeout: 5s
retries: 5
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- ./certs:/etc/nginx/certs:ro
depends_on:
- web
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
volumes:
postgres_data:
redis_data:
'''
# =============================================================================
# GITHUB ACTIONS CI/CD
# =============================================================================
GITHUB_ACTIONS_WORKFLOW = '''
# .github/workflows/ci.yml
name: CI/CD
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRES_PASSWORD: postgres
ports:
- 5432:5432
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
cache: 'pip'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-dev.txt
- name: Run linting
run: |
ruff check .
black --check .
- name: Run type checking
run: mypy .
- name: Run tests
run: pytest --cov=app --cov-report=xml
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
build:
needs: test
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Login to Container Registry
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Build and push
uses: docker/build-push-action@v5
with:
context: .
push: true
tags: ghcr.io/${{ github.repository }}:latest
cache-from: type=gha
cache-to: type=gha,mode=max
deploy:
needs: build
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- name: Deploy to production
run: |
echo "Deploying to production..."
# Add deployment commands here
'''
# =============================================================================
# CONFIGURATION MANAGEMENT
# =============================================================================
from dataclasses import dataclass, field
from typing import Any
import os
import json
@dataclass
class Config:
"""Application configuration with environment variable support."""
# Application
app_name: str = "myapp"
debug: bool = False
secret_key: str = ""
# Server
host: str = "0.0.0.0"
port: int = 8000
workers: int = 4
# Database
database_url: str = ""
database_pool_size: int = 5
# Redis
redis_url: str = ""
# Logging
log_level: str = "INFO"
log_format: str = "json"
@classmethod
def from_env(cls) -> 'Config':
"""Load configuration from environment variables."""
return cls(
app_name=os.getenv("APP_NAME", "myapp"),
debug=os.getenv("DEBUG", "false").lower() == "true",
secret_key=os.getenv("SECRET_KEY", ""),
host=os.getenv("HOST", "0.0.0.0"),
port=int(os.getenv("PORT", "8000")),
workers=int(os.getenv("WORKERS", "4")),
database_url=os.getenv("DATABASE_URL", ""),
database_pool_size=int(os.getenv("DATABASE_POOL_SIZE", "5")),
redis_url=os.getenv("REDIS_URL", ""),
log_level=os.getenv("LOG_LEVEL", "INFO"),
log_format=os.getenv("LOG_FORMAT", "json"),
)
def validate(self) -> list[str]:
"""Validate configuration and return list of errors."""
errors = []
if not self.secret_key:
errors.append("SECRET_KEY is required")
if not self.database_url:
errors.append("DATABASE_URL is required")
if self.port < 1 or self.port > 65535:
errors.append("PORT must be between 1 and 65535")
return errors
# =============================================================================
# HEALTH CHECKS
# =============================================================================
from datetime import datetime
import socket
@dataclass
class HealthCheck:
"""Health check result."""
name: str
status: str # "healthy", "unhealthy", "degraded"
message: str = ""
latency_ms: float = 0
details: dict = field(default_factory=dict)
class HealthChecker:
"""Application health checker."""
def __init__(self):
self.checks: list = []
def add_check(self, name: str, check_func):
"""Add a health check."""
self.checks.append((name, check_func))
def check_all(self) -> dict:
"""Run all health checks."""
results = []
overall_status = "healthy"
for name, check_func in self.checks:
start = datetime.now()
try:
result = check_func()
latency = (datetime.now() - start).total_seconds() * 1000
results.append(HealthCheck(
name=name,
status="healthy" if result else "unhealthy",
latency_ms=latency
))
if not result:
overall_status = "unhealthy"
except Exception as e:
latency = (datetime.now() - start).total_seconds() * 1000
results.append(HealthCheck(
name=name,
status="unhealthy",
message=str(e),
latency_ms=latency
))
overall_status = "unhealthy"
return {
"status": overall_status,
"timestamp": datetime.utcnow().isoformat(),
"checks": [
{
"name": r.name,
"status": r.status,
"latency_ms": round(r.latency_ms, 2),
"message": r.message
}
for r in results
]
}
def check_database(database_url: str) -> bool:
"""Check database connectivity."""
# Simplified - in real app would use actual DB connection
return bool(database_url)
def check_redis(redis_url: str) -> bool:
"""Check Redis connectivity."""
# Simplified - in real app would ping Redis
return bool(redis_url)
def check_disk_space(min_free_gb: float = 1.0) -> bool:
"""Check available disk space."""
import shutil
total, used, free = shutil.disk_usage("/")
free_gb = free / (1024 ** 3)
return free_gb >= min_free_gb
def check_memory(max_percent: float = 90.0) -> bool:
"""Check memory usage."""
try:
import psutil
return psutil.virtual_memory().percent < max_percent
except ImportError:
return True # Skip if psutil not available
# =============================================================================
# STRUCTURED LOGGING
# =============================================================================
import logging
import sys
class JSONFormatter(logging.Formatter):
"""JSON log formatter for production."""
def format(self, record: logging.LogRecord) -> str:
log_data = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"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
for key, value in record.__dict__.items():
if key not in logging.LogRecord.__dict__ and not key.startswith('_'):
log_data[key] = value
return json.dumps(log_data)
def setup_logging(level: str = "INFO", format: str = "json"):
"""Configure application logging."""
logger = logging.getLogger()
logger.setLevel(getattr(logging, level.upper()))
handler = logging.StreamHandler(sys.stdout)
if format == "json":
handler.setFormatter(JSONFormatter())
else:
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)
return logger
# =============================================================================
# GRACEFUL SHUTDOWN
# =============================================================================
import signal
import threading
from typing import Callable
class GracefulShutdown:
"""Handle graceful application shutdown."""
def __init__(self):
self.shutdown_event = threading.Event()
self.cleanup_handlers: list[Callable] = []
# Register signal handlers
signal.signal(signal.SIGTERM, self._handle_signal)
signal.signal(signal.SIGINT, self._handle_signal)
def _handle_signal(self, signum, frame):
"""Handle shutdown signal."""
print(f"\nReceived signal {signum}, initiating graceful shutdown...")
self.shutdown_event.set()
self._run_cleanup()
def add_cleanup(self, handler: Callable):
"""Add cleanup handler."""
self.cleanup_handlers.append(handler)
def _run_cleanup(self):
"""Run all cleanup handlers."""
for handler in self.cleanup_handlers:
try:
handler()
except Exception as e:
print(f"Cleanup error: {e}")
def should_shutdown(self) -> bool:
"""Check if shutdown was requested."""
return self.shutdown_event.is_set()
def wait(self, timeout: Optional[float] = None) -> bool:
"""Wait for shutdown signal."""
return self.shutdown_event.wait(timeout)
# =============================================================================
# DEPLOYMENT SCRIPTS
# =============================================================================
DEPLOY_SCRIPT = '''#!/bin/bash
# deploy.sh - Simple deployment script
set -e # Exit on error
# Configuration
APP_NAME="myapp"
DEPLOY_USER="deploy"
DEPLOY_HOST="production.example.com"
DEPLOY_PATH="/opt/apps/$APP_NAME"
DOCKER_IMAGE="ghcr.io/myorg/$APP_NAME:latest"
echo "🚀 Deploying $APP_NAME..."
# Pull latest image
echo "📦 Pulling latest image..."
ssh $DEPLOY_USER@$DEPLOY_HOST "docker pull $DOCKER_IMAGE"
# Stop old container
echo "🛑 Stopping old container..."
ssh $DEPLOY_USER@$DEPLOY_HOST "docker stop $APP_NAME || true"
ssh $DEPLOY_USER@$DEPLOY_HOST "docker rm $APP_NAME || true"
# Start new container
echo "▶️ Starting new container..."
ssh $DEPLOY_USER@$DEPLOY_HOST "docker run -d \\
--name $APP_NAME \\
--restart unless-stopped \\
-p 8000:8000 \\
--env-file $DEPLOY_PATH/.env \\
$DOCKER_IMAGE"
# Health check
echo "🏥 Running health check..."
sleep 5
HEALTH=$(ssh $DEPLOY_USER@$DEPLOY_HOST "curl -sf http://localhost:8000/health || echo 'unhealthy'")
if [[ "$HEALTH" == *"healthy"* ]]; then
echo "✅ Deployment successful!"
else
echo "❌ Health check failed, rolling back..."
ssh $DEPLOY_USER@$DEPLOY_HOST "docker stop $APP_NAME"
# Could add rollback logic here
exit 1
fi
'''
# =============================================================================
# DEMONSTRATIONS
# =============================================================================
if __name__ == "__main__":
print("=" * 60)
print("CONFIGURATION")
print("=" * 60)
# Set some environment variables for demo
os.environ["SECRET_KEY"] = "my-secret-key"
os.environ["DATABASE_URL"] = "postgresql://localhost/myapp"
os.environ["DEBUG"] = "true"
config = Config.from_env()
print(f"App: {config.app_name}")
print(f"Debug: {config.debug}")
print(f"Port: {config.port}")
errors = config.validate()
print(f"Validation errors: {errors}")
print("\n" + "=" * 60)
print("HEALTH CHECKS")
print("=" * 60)
health = HealthChecker()
health.add_check("database", lambda: check_database(config.database_url))
health.add_check("disk", lambda: check_disk_space(0.5))
result = health.check_all()
print(json.dumps(result, indent=2))
print("\n" + "=" * 60)
print("LOGGING")
print("=" * 60)
logger = setup_logging(level="INFO", format="json")
logger.info("Application started", extra={"version": "1.0.0"})
print("\n" + "=" * 60)
print("DOCKERFILE (Production)")
print("=" * 60)
print(PRODUCTION_DOCKERFILE[:500] + "...")