Django / Flask / FastAPI 框架集成实战

第 16 章 · Django / Flask / FastAPI 框架集成实战

本章与 第 13 章 Spring Boot 同级，系统讲解三大 Python Web 框架在集群内对接 全部 PaaS 组件 的生产级写法：项目结构、依赖、配置、中间件代码、Worker 部署、健康检查与验收。

第 9 章 为快速上手与 RocketMQ 事务 / Kafka 消费者专题；本章 补齐三框架与 MySQL、Redis、MongoDB、MQ、ES、MinIO、Keycloak、监控的 完整对照集成。

前置：第 2～5 章组件已部署；应用通过 应用部署 → 自定义镜像 发布到 app Namespace。

16.1 三框架工程结构对照

django-shop/                 flask-shop/                  fastapi-shop/
├── config/                  ├── app/                     ├── app/
│   ├── settings.py          │   ├── __init__.py          │   ├── main.py
│   ├── urls.py              │   ├── routes.py            │   ├── deps.py
│   └── celery.py            │   ├── models.py            │   ├── routers/
├── apps/orders/             │   └── tasks.py             │   └── services/
├── manage.py                ├── wsgi.py                  ├── requirements.txt
├── requirements.txt         ├── requirements.txt         └── Dockerfile
└── Dockerfile               └── Dockerfile

集群内 DNS（跨 Namespace 必用 FQDN）：

mysql.data.svc.cluster.local:3306
redis.data.svc.cluster.local:6379
rabbitmq.middleware.svc.cluster.local:5672
kafka.middleware.svc.cluster.local:9092
rocketmq-namesrv.middleware.svc.cluster.local:9876
elasticsearch.middleware.svc.cluster.local:9200
minio.data.svc.cluster.local:9000
keycloak.security.svc.cluster.local:8080

16.2 依赖清单对照

16.3 Django 框架完整集成

16.3.1 settings.py（中间件全集）

import os
import environ

env = environ.Env()
environ.Env.read_env()  # 本地 .env；集群内主要靠 K8s env 注入

SECRET_KEY = env("DJANGO_SECRET_KEY")
DEBUG = env.bool("DJANGO_DEBUG", default=False)
ALLOWED_HOSTS = env.list("DJANGO_ALLOWED_HOSTS", default=["*"])

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "django_prometheus",
    "mozilla_django_oidc",
    "apps.orders",
]

MIDDLEWARE = [
    "django_prometheus.middleware.PrometheusBeforeMiddleware",
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "mozilla_django_oidc.middleware.SessionRefresh",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django_prometheus.middleware.PrometheusAfterMiddleware",
]

# ---------- MySQL ----------
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "HOST": env("MYSQL_HOST", default="mysql.data.svc.cluster.local"),
        "PORT": env("MYSQL_PORT", default="3306"),
        "NAME": env("MYSQL_DATABASE", default="appdb"),
        "USER": env("MYSQL_USER", default="app"),
        "PASSWORD": env("MYSQL_PASSWORD"),
        "OPTIONS": {"charset": "utf8mb4"},
    }
}

# ---------- Redis 缓存 / Session ----------
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": env("REDIS_URL", default="redis://:pass@redis.data.svc.cluster.local:6379/1"),
        "OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient"},
    }
}
SESSION_ENGINE = "django.contrib.sessions.backends.cache"
SESSION_CACHE_ALIAS = "default"

# ---------- Celery + RabbitMQ ----------
CELERY_BROKER_URL = env(
    "CELERY_BROKER_URL",
    default="amqp://app:pass@rabbitmq.middleware.svc.cluster.local:5672//",
)
CELERY_RESULT_BACKEND = env("REDIS_URL")
CELERY_TASK_SERIALIZER = "json"

# ---------- MinIO（S3）----------
DEFAULT_FILE_STORAGE = "storages.backends.s3boto3.S3Boto3Storage"
AWS_S3_ENDPOINT_URL = env("MINIO_ENDPOINT", default="http://minio.data.svc.cluster.local:9000")
AWS_ACCESS_KEY_ID = env("MINIO_ACCESS_KEY")
AWS_SECRET_ACCESS_KEY = env("MINIO_SECRET_KEY")
AWS_STORAGE_BUCKET_NAME = env("MINIO_BUCKET", default="uploads")
AWS_S3_ADDRESSING_STYLE = "path"

# ---------- Keycloak OIDC ----------
AUTHENTICATION_BACKENDS = (
    "mozilla_django_oidc.auth.OIDCAuthenticationBackend",
    "django.contrib.auth.backends.ModelBackend",
)
OIDC_RP_CLIENT_ID = env("OIDC_CLIENT_ID", default="django-web")
OIDC_RP_CLIENT_SECRET = env("OIDC_CLIENT_SECRET")
OIDC_OP_AUTHORIZATION_ENDPOINT = env(
    "OIDC_AUTH_URL",
    default="http://keycloak.security.svc.cluster.local:8080/realms/shop/protocol/openid-connect/auth",
)
OIDC_OP_TOKEN_ENDPOINT = env("OIDC_TOKEN_URL")
OIDC_OP_USER_ENDPOINT = env("OIDC_USER_URL")
LOGIN_URL = "oidc_authentication_init"

# ---------- 业务常量（也可从 Nacos 拉取，见 16.6）----------
ROCKETMQ_NAMESRV = env("ROCKETMQ_NAMESRV", default="rocketmq-namesrv.middleware:9876")
KAFKA_BOOTSTRAP = env("KAFKA_BOOTSTRAP", default="kafka.middleware:9092")
MONGO_URI = env("MONGO_URI", default="mongodb://admin:pass@mongodb.data:27017/appdb?authSource=admin")
ES_URL = env("ES_URL", default="http://elasticsearch.middleware:9200")

16.3.2 业务层集成示例

MongoDB 商品文档

# apps/catalog/mongo.py
from django.conf import settings
from pymongo import MongoClient

_client = MongoClient(settings.MONGO_URI)

def products():
    return _client["appdb"]["products"]

def upsert_product(doc: dict):
    return products().update_one({"sku": doc["sku"]}, {"$set": doc}, upsert=True)

Elasticsearch 搜索

from django.conf import settings
from elasticsearch import Elasticsearch

es = Elasticsearch([settings.ES_URL])

def search_products(q: str):
    return es.search(index="products", query={"match": {"name": q}})

RocketMQ 发消息

from django.conf import settings
from rocketmq.client import Producer, Message

_producer = None

def get_producer():
    global _producer
    if _producer is None:
        p = Producer("django-producer")
        p.set_name_server_address(settings.ROCKETMQ_NAMESRV)
        p.start()
        _producer = p
    return _producer

def publish_order_created(order_id: int):
    msg = Message("TopicOrder")
    msg.set_body(str(order_id).encode())
    get_producer().send_sync(msg)

Kafka 埋点

from django.conf import settings
from kafka import KafkaProducer
import json

_producer = KafkaProducer(
    bootstrap_servers=settings.KAFKA_BOOTSTRAP,
    value_serializer=lambda v: json.dumps(v).encode(),
)

def track_event(user_id: int, event: str):
    _producer.send("user-events", {"user_id": user_id, "event": event})

限流（等价 Sentinel）

from django_ratelimit.decorators import ratelimit
from django.http import JsonResponse

@ratelimit(key="ip", rate="100/m", block=True)
def api_orders(request):
    if getattr(request, "limited", False):
        return JsonResponse({"detail": "限流"}, status=429)
    ...

16.3.3 健康检查与 Prometheus

config/urls.py

from django.urls import path, include
from django.http import JsonResponse

def health(_):
    return JsonResponse({"status": "ok"})

urlpatterns = [
    path("health/", health),
    path("", include("django_prometheus.urls")),
    path("oidc/", include("mozilla_django_oidc.urls")),
    path("api/", include("apps.orders.urls")),
]

Pod 探针（确认页 YAML）：

livenessProbe:
  httpGet:
    path: /health/
    port: 8000
  initialDelaySeconds: 30
readinessProbe:
  httpGet:
    path: /health/
    port: 8000
  periodSeconds: 10

Prometheus 注解：

metadata:
  annotations:
    prometheus.io/scrape: "true"
    prometheus.io/port: "8000"
    prometheus.io/path: "/metrics"

16.3.4 Deployment 环境变量

env:
  - name: DJANGO_SECRET_KEY
    value: "随机长字符串"
  - name: MYSQL_PASSWORD
    value: "与第2章一致"
  - name: REDIS_URL
    value: "redis://:RedisPass@redis.data.svc.cluster.local:6379/1"
  - name: CELERY_BROKER_URL
    value: "amqp://app:pass@rabbitmq.middleware.svc.cluster.local:5672//"
  - name: MINIO_ENDPOINT
    value: "http://minio.data.svc.cluster.local:9000"
  - name: MINIO_ACCESS_KEY
    value: "minioadmin"
  - name: MINIO_SECRET_KEY
    value: "Minio强密码"
  - name: ROCKETMQ_NAMESRV
    value: "rocketmq-namesrv.middleware.svc.cluster.local:9876"
  - name: KAFKA_BOOTSTRAP
    value: "kafka.middleware.svc.cluster.local:9092"
  - name: MONGO_URI
    value: "mongodb://admin:pass@mongodb.data.svc.cluster.local:27017/appdb?authSource=admin"
  - name: ES_URL
    value: "http://elasticsearch.middleware.svc.cluster.local:9200"
  - name: OIDC_CLIENT_SECRET
    value: "Keycloak客户端密钥"

Celery Worker（第二个 Deployment，镜像相同）：

command: ["celery", "-A", "config", "worker", "-l", "info"]
# env 与 django-api 完全一致

16.4 Flask 框架完整集成

16.4.1 应用工厂 app/__init__.py

import os
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from prometheus_flask_exporter import PrometheusMetrics

db = SQLAlchemy()
cache = Cache()
limiter = Limiter(key_func=get_remote_address)

def create_app():
    app = Flask(__name__)
    app.config.from_mapping(
        SECRET_KEY=os.environ["FLASK_SECRET_KEY"],