在 DevContainers 中高效管理 Celery、CeleryBeat、Flower 和 FastAPI

FastAPI、Celery 和相关服务的动态 Docker 设置

该脚本使用 Python 和 Docker Compose 在开发环境中配置动态服务管理。服务设置为仅在需要时运行，从而优化资源使用。

# Docker Compose file with profiles for selective service activation
version: '3.8'
services:
  trader:
    build:
      context: ..
      dockerfile: .devcontainer/Dockerfile
    volumes:
      - ../:/app:cached
      - ~/.ssh:/home/user/.ssh:ro
      - ~/.gitconfig:/home/user/.gitconfig:cached
    command: sleep infinity
    environment:
      - AGENT_CACHE_REDIS_HOST=redis
      - DB_URL=postgresql://myuser:mypassword@postgres:5432/db
    profiles:
      - default
  celery:
    build:
      context: ..
      dockerfile: .devcontainer/Dockerfile
    volumes:
      - ../:/app:cached
    command: celery -A src.celery worker --loglevel=debug
    environment:
      - AGENT_CACHE_REDIS_HOST=redis
      - DB_URL=postgresql://myuser:mypassword@postgres:5432/db
    profiles:
      - optional
  redis:
    image: redis:latest
    networks:
      - trader-network
    profiles:
      - default

使用Python脚本优化芹菜初创公司

此 Python 脚本根据用户操作自动启动 Celery 服务。它使用 Docker SDK for Python 来控制容器。

import docker
def start_optional_services():
    client = docker.from_env()
    services = ['celery', 'celerybeat', 'flower']
    for service in services:
        try:
            container = client.containers.get(service)
            if container.status != 'running':
                container.start()
                print(f"Started {service}")
            else:
                print(f"{service} is already running")
        except docker.errors.NotFound:
            print(f"Service {service} not found")
if __name__ == "__main__":
    start_optional_services()

Celery 工作流程的单元测试

该Python测试脚本使用Pytest来验证Celery任务执行环境，确保模块化和正确性。

import pytest
from celery import Celery
@pytest.fixture
def celery_app():
    return Celery('test', broker='redis://localhost:6379/0')
def test_task_execution(celery_app):
    @celery_app.task
    def add(x, y):
        return x + y
    result = add.delay(2, 3)
    assert result.get(timeout=5) == 5

有关 Docker 和 Celery 集成的常见问题

1. Docker构成个人资料的目的是什么？
2. 它们允许将服务分组为逻辑类别，从而实现容器的选择性启动。例如，“默认”配置文件可以包括 FastAPI 等基本服务，而“可选”配置文件包括 Celery 工人。
3. 如何从可选配置文件中启动特定服务？
4. 使用命令 docker compose --profile optional up celery 仅启动“可选”配置文件中的容器。
5. 使用Docker SDK来管理容器的优点是什么？
6. Docker SDK可以通过Python脚本对容器进行程序化控制，例如动态启动或停止特定服务。
7. 如何实时监视芹菜任务？
8. 您可以使用基于网络的监视工具Flower。开始使用它 celery -A app flower 通过 Web 界面查看任务进度和日志。
9. 仅按需运行 Celery 工作线程有什么好处？
10. 它节省了资源并避免了不必要的 API 调用。例如，您可以延迟启动工作程序，直到特定的集成测试需要后台任务处理。

开发人员有效的容器管理

有效管理开发资源对于保持生产力至关重要。通过有选择地启动 Celery 和 Flower 等服务，开发人员可以专注于构建功能，而不会受到不必要的干扰。 Docker Compose 配置文件简化了这个过程，确保资源得到明智的使用。

脚本和测试框架通过提供动态服务激活和验证，进一步增强了此设置。这些工具结合在一起，提供了一个灵活而健壮的环境，使开发人员可以轻松调试和部署其FastAPI应用程序。 🌟