基于Django创建定时API拉取任务，求GitHub示例或操作手册

Hey there! Let's break down how to set up scheduled API data pulls in your Django project—this is super common once you start integrating external services, so you're not alone here. Below are the most practical, widely-used solutions with step-by-step guidance, plus tips on finding working GitHub examples:

1. Celery + Celery Beat（生产级主流方案） #

This is the go-to stack for larger, production-ready Django apps. Celery handles async tasks, while Celery Beat manages the scheduling logic.

• Install dependencies:

  pip install celery redis
  

  (We use Redis as the message broker here; RabbitMQ is another solid option.)

• Configure Celery in your project:
  Create a celery.py file in your project root:

  import os
  from celery import Celery
  
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
  app = Celery('your_project')
  app.config_from_object('django.conf:settings', namespace='CELERY')
  app.autodiscover_tasks()
  

• Add Celery settings to settings.py:

  CELERY_BROKER_URL = 'redis://localhost:6379/0'
  CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  CELERY_TIMEZONE = 'Asia/Shanghai'  # Replace with your timezone
  

• Write your API pull task:
  Create a tasks.py file in your app:

  import requests
  from celery import shared_task
  
  @shared_task
  def pull_external_api_data():
      try:
          response = requests.get('https://your-external-api-url.com/data')
          response.raise_for_status()
          data = response.json()
          # Add your data processing logic here (e.g., save to Django models)
          # Example: YourModel.objects.create(**data)
          return f"Successfully pulled {len(data)} records"
      except requests.exceptions.RequestException as e:
          return f"API pull failed: {str(e)}"
  

• Set up the schedule:
  Add this to settings.py to define when the task runs:

  from celery.schedules import crontab
  
  CELERY_BEAT_SCHEDULE = {
      # Run every hour
      'pull-api-hourly': {
          'task': 'your_app.tasks.pull_external_api_data',
          'schedule': 3600.0,
      },
      # Alternative: Run every day at 2 AM
      # 'pull-api-daily': {
      #     'task': 'your_app.tasks.pull_external_api_data',
      #     'schedule': crontab(hour=2, minute=0),
      # },
  }
  

• Start the services:
  First launch Redis, then run:

  # Start Celery worker
  celery -A your_project worker --loglevel=info
  # Start Celery Beat scheduler
  celery -A your_project beat --loglevel=info
  

To find GitHub examples, search for keywords like django celery beat example—you’ll find full open-source projects (like Django dashboards or data ingestion tools) using this stack.

2. Django Q（轻量简洁替代） #

If Celery feels overkill for your project, Django Q is a simpler option that works with your existing Django database (or Redis/MongoDB) without extra complex setup.

• Install & configure:

  pip install django-q
  

  Add 'django_q' to INSTALLED_APPS in settings.py, then add:

  Q_CLUSTER = {
      'name': 'DjangoQCluster',
      'workers': 4,
      'timeout': 90,
      'orm': 'default'  # Use Django's database for task storage
  }
  

• Write your task:
  Same as above, no decorator needed in tasks.py:

  import requests
  
  def pull_external_api_data():
      try:
          response = requests.get('https://your-external-api-url.com/data')
          response.raise_for_status()
          data = response.json()
          # Process and save data here
          return "Data pull successful"
      except Exception as e:
          return str(e)
  

• Set up scheduling:
  Run python manage.py migrate to create Django Q’s database tables, then either:

  1. Add schedules directly via the Django admin interface, or
  2. Define schedules in settings.py:

     Q_SCHEDULE = [
         {
             'name': 'Hourly API Pull',
             'task': 'your_app.tasks.pull_external_api_data',
             'schedule': 3600,  # Run every hour
         }
     ]
     

• Start the task cluster:

  python manage.py qcluster
  

Search GitHub for django q scheduled task example to find minimal, working projects using this tool.

3. APScheduler（通用定时任务库） #

APScheduler is a standalone scheduling library that integrates seamlessly with Django, great if you want flexibility beyond Django-specific tools.

• Install & configure:

  pip install apscheduler django-apscheduler
  

  Add 'django_apscheduler' to INSTALLED_APPS, then run python manage.py migrate.

• Set up the scheduler:
  Update your app’s apps.py to start the scheduler when Django launches:

  from django.apps import AppConfig
  from apscheduler.schedulers.background import BackgroundScheduler
  from .tasks import pull_external_api_data
  import os
  
  class YourAppConfig(AppConfig):
      default_auto_field = 'django.db.models.BigAutoField'
      name = 'your_app'
  
      def ready(self):
          # Avoid duplicate schedulers in dev environment
          if os.environ.get('RUN_MAIN') == 'true':
              scheduler = BackgroundScheduler(timezone='Asia/Shanghai')
              # Run every hour
              scheduler.add_job(pull_external_api_data, 'interval', hours=1)
              scheduler.start()
  

GitHub examples are easy to find with django apscheduler example—look for projects that use it for periodic data syncs.

Quick Tips for Newbies #

• Start with Django Q or APScheduler for your first project—they have lower learning curves than Celery.
• Test with short intervals (e.g., 10 seconds) to quickly verify your task works.
• Always handle API errors (timeouts, 4xx/5xx responses) to prevent tasks from crashing silently.
• In production, use tools like systemd or supervisor to keep your scheduler/worker processes running 24/7.

内容的提问来源于stack exchange，提问作者Ricardo Haninec Silva