Django中如何使用Celery实现每日定时自动调用2个GET接口

前置依赖安装 #

先安装HTTP请求库，用于发送GET请求：
pip install requests

1. 应用目录tasks.py代码 #

直接替换你应用下tasks.py的内容，把其中的API地址、鉴权信息改成你自己的即可：

import requests
from celery import shared_task
import logging

logger = logging.getLogger(__name__)

# -------------------------- 按需修改以下配置 --------------------------
FIRST_API_URL = "http://你的服务地址/第一个API路径/"
SECOND_API_URL = "http://你的服务地址/第二个API路径/"
# 接口需要鉴权就填请求头，不需要就保留空字典
REQUEST_HEADERS = {}
# 最大重试次数、重试间隔（秒）
MAX_RETRY_TIMES = 3
RETRY_DELAY = 60
REQUEST_TIMEOUT = 30
# --------------------------------------------------------------------

@shared_task(bind=True, max_retries=MAX_RETRY_TIMES, default_retry_delay=RETRY_DELAY)
def trigger_first_api(self):
    try:
        resp = requests.get(
            FIRST_API_URL,
            headers=REQUEST_HEADERS,
            timeout=REQUEST_TIMEOUT
        )
        resp.raise_for_status()
        logger.info(f"第一个API调用成功，状态码：{resp.status_code}")
        return True
    except Exception as e:
        logger.error(f"第一个API调用失败：{str(e)}，即将重试")
        raise self.retry(exc=e)

@shared_task(bind=True, max_retries=MAX_RETRY_TIMES, default_retry_delay=RETRY_DELAY)
def trigger_second_api(self):
    try:
        resp = requests.get(
            SECOND_API_URL,
            headers=REQUEST_HEADERS,
            timeout=REQUEST_TIMEOUT
        )
        resp.raise_for_status()
        logger.info(f"第二个API调用成功，状态码：{resp.status_code}")
        return True
    except Exception as e:
        logger.error(f"第二个API调用失败：{str(e)}，即将重试")
        raise self.retry(exc=e)

2. 主目录celery.py定时规则配置 #

在你已经创建的项目主目录celery.py中，补充定时调度配置，示例为每天凌晨2点依次调用两个接口：

import os
from celery import Celery
from celery.schedules import crontab

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目配置目录名.settings')

app = Celery('你的项目名')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# 新增以下定时任务配置
app.conf.beat_schedule = {
    'daily-trigger-first-api': {
        'task': '你的应用名.tasks.trigger_first_api',
        # 执行时间：每天2点0分，修改hour、minute即可调整固定触发时间
        'schedule': crontab(hour=2, minute=0),
    },
    'daily-trigger-second-api': {
        'task': '你的应用名.tasks.trigger_second_api',
        # 错开5分钟执行第二个接口，避免资源抢占
        'schedule': crontab(hour=2, minute=5),
    },
}
# 必须配置为国内时区，否则定时时间会和北京时间差8小时
app.conf.timezone = 'Asia/Shanghai'

3. 服务启动命令 #

Celery定时任务需要同时启动两个进程才能正常运行：

• 启动任务执行进程：celery -A 你的项目名 worker -l info
• 启动定时调度进程：celery -A 你的项目名 beat -l info

注意事项 #

• 如果两个API是本项目自身的接口，直接用127.0.0.1:端口号作为地址即可，不需要走公网域名，速度更快
• 如果接口需要登录态，可以直接把API对应的执行逻辑抽成普通Python函数在任务里直接调用，不需要走HTTP请求，减少网络开销
• crontab支持更灵活的时间配置，比如工作日执行、每小时执行等，可以根据自己的需求调整参数
• 生产环境建议用进程托管工具（比如supervisor）守护worker和beat进程，避免进程意外退出导致定时任务失效

内容的提问来源于stack exchange，提问作者Dieharder_3