← 返回首页
容器

Docker与Python

📂 python ⏱ 4 min 642 words

Docker与Python

Docker是现代应用部署的标准工具。本文将介绍如何使用Docker容器化Python应用,包括最佳实践和高级配置。

基础Dockerfile

# 基础Python镜像
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 设置环境变量
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 8000

# 运行命令
CMD ["python", "app.py"]

多阶段构建

# 构建阶段
FROM python:3.11 as builder

WORKDIR /app

COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

COPY . .
RUN python -m compileall .

# 运行阶段
FROM python:3.11-slim

WORKDIR /app

# 从构建阶段复制依赖
COPY --from=builder /root/.local /root/.local
COPY --from=builder /app .

ENV PATH=/root/.local/bin:$PATH

EXPOSE 8000
CMD ["python", "app.py"]

docker-compose配置

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    volumes:
      - .:/app
      - ./data:/app/data
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  db:
    image: postgres:15
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  celery_worker:
    build: .
    command: celery -A app.celery worker --loglevel=info
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    volumes:
      - .:/app

volumes:
  postgres_data:

Python应用示例

# app.py
from fastapi import FastAPI
import os
import redis
import psycopg2
from celery import Celery

app = FastAPI()

# Redis连接
redis_client = redis.Redis(
    host=os.getenv('REDIS_HOST', 'redis'),
    port=int(os.getenv('REDIS_PORT', 6379)),
    db=0
)

# Celery配置
celery = Celery(
    'tasks',
    broker=os.getenv('REDIS_URL', 'redis://redis:6379'),
    backend=os.getenv('REDIS_URL', 'redis://redis:6379')
)

@celery.task
def process_data(data):
    # 模拟长时间运行的任务
    import time
    time.sleep(5)
    return f"处理完成: {data}"

@app.get("/health")
def health_check():
    return {"status": "healthy"}

@app.get("/")
def root():
    return {"message": "Docker Python应用"}

@app.post("/process")
async def process_request(data: dict):
    # 提交异步任务
    task = process_data.delay(data)
    return {"task_id": task.id, "status": "submitted"}

@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
    task = process_data.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

优化Docker镜像

# 使用slim镜像
FROM python:3.11-slim

# 安装依赖时清理缓存
RUN apt-get update && \
    apt-get install -y --no-install-recommends gcc && \
    rm -rf /var/lib/apt/lists/* && \
    pip install --no-cache-dir --user -r requirements.txt && \
    apt-get purge -y --auto-remove gcc && \
    rm -rf /root/.cache

# 使用非root用户
RUN useradd --create-home --shell /bin/bash appuser
USER appuser

WORKDIR /home/appuser/app

COPY --chown=appuser:appuser . .

CMD ["python", "app.py"]

开发环境配置

# docker-compose.dev.yml
version: '3.8'

services:
  web:
    build:
      context: .
      dockerfile: Dockerfile.dev
    volumes:
      - .:/app
      - /app/node_modules  # 如果有前端依赖
    environment:
      - FLASK_ENV=development
      - DEBUG=1
    ports:
      - "8000:8000"
      - "5678:5678"  # 调试端口
    command: python -m debugpy --listen 0.0.0.0:5678 --wait-for-client -m flask run --host=0.0.0.0 --reload
# Dockerfile.dev
FROM python:3.11

WORKDIR /app

# 安装调试工具
RUN pip install debugpy pytest-watch

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000 5678

CMD ["python", "-m", "flask", "run", "--host=0.0.0.0", "--reload"]

环境变量管理

# config.py
import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///local.db')
    redis_url: str = os.getenv('REDIS_URL', 'redis://localhost:6379')
    debug: bool = os.getenv('DEBUG', 'false').lower() == 'true'
    secret_key: str = os.getenv('SECRET_KEY', 'dev-secret-key')
    
    @classmethod
    def from_env(cls):
        return cls(
            database_url=os.environ['DATABASE_URL'],
            redis_url=os.environ.get('REDIS_URL', 'redis://localhost:6379'),
            debug=os.environ.get('DEBUG', 'false').lower() == 'true',
            secret_key=os.environ.get('SECRET_KEY', os.urandom(24).hex())
        )

# 使用
config = Config.from_env()

健康检查和监控

# health.py
from fastapi import FastAPI, Response
import psutil
import redis
import psycopg2
from datetime import datetime

app = FastAPI()

@app.get("/health")
async def health_check():
    checks = {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "services": {}
    }
    
    # 检查内存使用
    memory = psutil.virtual_memory()
    checks["services"]["memory"] = {
        "status": "healthy" if memory.percent < 80 else "warning",
        "usage": f"{memory.percent}%"
    }
    
    # 检查Redis
    try:
        r = redis.Redis(host='redis', port=6379)
        r.ping()
        checks["services"]["redis"] = {"status": "healthy"}
    except Exception as e:
        checks["services"]["redis"] = {"status": "unhealthy", "error": str(e)}
    
    # 检查数据库
    try:
        conn = psycopg2.connect(os.getenv('DATABASE_URL'))
        conn.close()
        checks["services"]["database"] = {"status": "healthy"}
    except Exception as e:
        checks["services"]["database"] = {"status": "unhealthy", "error": str(e)}
    
    # 整体状态
    unhealthy = [s for s in checks["services"].values() if s["status"] == "unhealthy"]
    if unhealthy:
        checks["status"] = "unhealthy"
        return Response(content=json.dumps(checks), status_code=503, media_type="application/json")
    
    return checks

@app.get("/metrics")
async def metrics():
    """Prometheus格式的指标"""
    return {
        "process_cpu_seconds_total": psutil.cpu_times().user,
        "process_memory_bytes": psutil.Process().memory_info().rss,
        "process_threads": psutil.Process().num_threads()
    }

部署到生产环境

# 构建生产镜像
docker build -t myapp:latest -f Dockerfile .

# 运行容器
docker run -d \
  --name myapp \
  -p 8000:8000 \
  -e DATABASE_URL=postgresql://... \
  -e REDIS_URL=redis://... \
  -v /data/myapp:/app/data \
  --restart unless-stopped \
  myapp:latest

# 使用docker-compose生产配置
docker-compose -f docker-compose.prod.yml up -d

常见问题

  1. 镜像大小:使用多阶段构建和slim镜像
  2. 数据持久化:使用卷挂载
  3. 网络配置:使用自定义网络
  4. 日志管理:配置日志驱动
  5. 安全扫描:定期扫描镜像漏洞

总结

Docker为Python应用提供了标准化的部署环境。掌握Dockerfile编写、docker-compose配置和生产部署策略,可以确保应用在任何环境中一致运行。