我獨自 Django：快速打造天氣 API 服務（下篇）

2025/08/23 更新2025/08/23 發佈閱讀 14 分鐘

Generating by ChatGPT-4o

在上一篇《我獨自 Django：快速打造天氣 API 服務（中篇）》中，我們探討了使用 API 的不同方式，一是即時取得資料並顯示；另一則是將資料儲存進資料庫。在理解了如何串接 API 與儲存之後，接下來的重點，就是要逐步的累積資料，以備未來分析時使用。

說到「排程」，相信不少讀者對 Windows 內建的排程系統並不陌生，簡單來說，就是在指定時間點自動執行預先設定的任務。而在 Django 中，我們可以透過適合的工具去設定與執行，今天要介紹 Celery，逐步的帶大家了解設定過程與實際的執行。

Celery 是一個處理非同步應用的 Python 套件，支援分散式系統架構，能有效處理大量背景任務，像是將耗時的操作，如發送郵件、數據處理與多次 API 請求等操作，都可以設定在背景執行，避免阻塞主程式。更進階的應用也可以使用 RabbitMQ 或是 Redis 來做中介系統整合，進行排程管理任務。

本篇也會使用到 Redis 作為 Celery 的中介(Message Broker)；而對這個工具有興趣的讀者，也可以在熟悉之後做更深入的研究。
*部分的程式因為 blog 的編譯器會有點跑版，可以到我的 Medium 觀看更好複製的版本

那我們就從程式碼開始吧！回到 VSCode 介面，我們先安裝需要的工具

cd backend # 切換到 backend

./django-backend/Scripts/activate # 記得切換虛擬環境

pip install celery
pip install redis

另外 Windows 中 Redis 可能不支援，要另外安裝 Memurai，其應用與 Redis 幾乎一樣，直接使用開發版本就好 :

Memurai 安裝

安裝完之後開啟任意的 cmd 測試是否有執行，預設的 Port 為 6379

memurai-cli # 檢查是否有進入 127.0.0.1:6379

net start memurai # 若沒有啟動則使用這個指令

一切就緒之後回到程式，建立新的 .py 檔 weather/tasks.py ，輸入以下程式碼 :

from celery import shared_task
import requests
from .models import Weather
from datetime import datetime

@shared_task
def fetch_and_store_weather():
    cities = {
        "Taipei": {"lat": 25.05, "lon": 121.53},
        "Taichung": {"lat": 24.15, "lon": 120.67},
        "Kaohsiung": {"lat": 22.63, "lon": 120.30},
    }
    
    for city, coord in cities.items():
        url = f"https://api.open-meteo.com/v1/forecast?latitude={coord['lat']}&longitude={coord['lon']}&current_weather=true"
        response = requests.get(url)
        data = response.json()
        
        current = data.get("current_weather", {})
        if current:

            Weather.objects.create(
                city=city,
                temperature=current.get("temperature"),
                windspeed=current.get("windspeed"),
                time=datetime.fromisoformat(current.get("time"))
            )

然後再建立新的 .py 檔 config/celery.py ，輸入以下程式碼 :

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

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')

app = Celery('config')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.conf.worker_pool = 'solo' # 可先這行註解後執行看看
app.conf.broker_url = 'redis://localhost:6379/0'
app.autodiscover_tasks(['weather'])
app.conf.timezone = 'Asia/Taipei'
app.conf.beat_schedule = {
    'fetch-weather-every-day': {
        'task': 'weather.tasks.fetch_and_store_weather',
        'schedule': crontab(hour=10, minute=10), # 該時區的時間，這裡是 10:10
    },}

* worker_pool = 'solo' 是為了解決 Windows 環境中多處理程序可能導致的序列化問題。在正式開發階段，建議將此設定移除，讓 Celery 自行使用最適合的處理方式，以獲得更高效的執行效能。

再來到設定 config/settings.py

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'weather',
    'rest_framework',
    'django_celery_beat' # 添加這行
]

開啟兩個 Terminal 執行程式

cd backend # 記得切換到後端專案內

首先第一個 Terminal 執行 Celery 的 worker

celery -A config worker --loglevel=info

此時可以看到連接資訊，以及 Redis 連線成功

Celery worker

再到第二個 Terminal 上執行 beat

celery -A config beat --loglevel=info

連接資訊如下

Celery beat

等到設定的時間(這裡是設定為10:10)，就會把昨日的氣象資訊匯入資料庫了！

執行排程後，成功匯入資料庫

以上就是使用 Celery 設定排程的基本流程。有實務經驗的讀者可能會發現，還有許多可以優化的地方，例如資料庫想再多加一些欄位，或是 Call API 的頻率與時機。這些問題往往要經過實際操作與測試後，才能更有感的體會與改善。希望各位也能找到自己實作的樂趣和成就感！

在理解了 Celery 的基礎應用之後，下一步我們要來探討：Django 中的 API 架構應該如何設計，才能兼顧可讀性、擴展性與維護性？

當專案逐漸成長，功能變得複雜時，良好的 API 結構將會是每位後端工程師不可或缺的技能。這也是區分 Junior 與 Senior 的關鍵之一。

在傳統的 Django 架構中，後端邏輯通常集中在 views.py 中，再將完成的邏輯使用諸如 JsonResponse 等用法回傳資料，並由 urls.py 負責對應 URL 路徑，構成基本的 API。

而隨著專案功能的擴展，將所有的業務邏輯都堆疊在views.py 會變得難以維護，也很難測試。所以適當的做法是將主要邏輯抽離到 Service Layer，讓 views.py 僅負責接收請求與回應結果，使整體架構更清晰易懂。這樣做的好處是：當你看到某隻 API 時，可以一眼了解其用途與結構，並能在 services/ 中集中維護商業邏輯。

以下簡單實作一部分讓讀者更深刻一些，以我們已建立的 App weather 來做說明，這是目前的架構

weather 架構

我們先新增未來要擴展邏輯的 services，並且新增一個 .py 檔

weather 架構

回到我們的 weather/views.py ，把程式改成 :

from rest_framework.decorators import api_view
from rest_framework.response import Response
from .services.weather_service import get_weather_summary

@api_view(['GET']) # 添加該 API 的簡易註解
def get_weather(request):
    data = get_weather_summary()    
    return Response(data)

再到 weather/services/weather_service.py ，新增以下程式 :

import requests

def get_weather_summary():
    cities = {
        "Taipei": {"lat": 25.05, "lon": 121.53},
        "Taichung": {"lat": 24.15, "lon": 120.67},
        "Kaohsiung": {"lat": 22.63, "lon": 120.30},
    }
    
    weather_data = []
    
    for city, coord in cities.items():
        url = f"https://api.open-meteo.com/v1/forecast?latitude={coord['lat']}&longitude={coord['lon']}&current_weather=true"
        response = requests.get(url)
        data = response.json()
        current = data.get("current_weather", {})

        if current:
            weather_data.append({
                "city": city,
                "temperature": current.get("temperature"),
                "windspeed": current.get("windspeed"),
                "time": current.get("time")
            })
            
     return weather_data