如何在Flask蓝图中集成Azure AD身份验证（适配Celery场景）

1. 重构项目结构，拆分认证逻辑到独立蓝图 #

将认证相关代码单独抽离到auth.py作为独立蓝图，与业务视图、Celery任务解耦，从根源避免循环导入问题。

auth.py 代码实现 #

from flask import Blueprint, render_template, session, url_for, current_app, request, redirect
import identity.web

auth_bp = Blueprint("auth", __name__)

# 懒加载Auth实例，规避模块初始化阶段无请求上下文的问题
def get_auth():
    return identity.web.Auth(
        session=session,
        authority=current_app.config["AUTHORITY"],
        client_id=current_app.config["CLIENT_ID"],
        client_credential=current_app.config["CLIENT_SECRET"],
    )

@auth_bp.route("/login")
def login():
    auth = get_auth()
    from .config import app_config
    return render_template(
        "login.html",
        version=identity.web.__version__,
        **auth.log_in(
            scopes=app_config.SCOPE,
            redirect_uri=url_for("auth.auth_response", _external=True),
            prompt="select_account",
        )
    )

@auth_bp.route("/auth-response")
def auth_response():
    auth = get_auth()
    result = auth.complete_log_in(request.args)
    if "error" in result:
        return render_template("error.html", result=result)
    return redirect(url_for("main.index"))

@auth_bp.route("/logout")
def logout():
    auth = get_auth()
    return render_template(
        "logout.html",
        sign_out_url=auth.log_out(url_for("main.index", _external=True)),
        post_logout_redirect_uri=url_for("main.index", _external=True),
    )

2. 修正views.py的核心问题 #

原代码存在模块级初始化Auth时无上下文、缺失必要导入、未做登录保护等问题，修正后代码如下：

from flask import Blueprint, render_template, request, redirect, url_for
from celery.result import AsyncResult
from functools import wraps
from .tasks import mytask

main = Blueprint("main", __name__)

# 登录保护装饰器，未登录用户自动跳转至登录页
def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        from .auth import get_auth
        auth = get_auth()
        if not auth.get_user():
            return redirect(url_for("auth.login"))
        return f(*args, **kwargs)
    return decorated_function

@main.route("/")
@login_required
def index():
    return render_template("index.html")

@main.route("/task/start", methods=["POST"])
@login_required
def task_start():
    task = mytask.delay(10)
    return {"task_id": task.id}

@main.route("/task/progress", methods=["POST"])
@login_required
def task_progress():
    data = request.get_json()
    task = AsyncResult(data["task_id"])
    return {"state": task.state, "progress": task.info.get("progress", 0)}

3. 解决循环导入的核心措施 #

• 延迟依赖初始化：所有依赖current_app或跨蓝图的操作，都放到视图函数内部执行（比如get_auth函数、登录保护装饰器内的导入），避免模块加载阶段的直接依赖。
• 工厂模式创建app：在项目根__init__.py中用工厂函数创建Flask实例，待app初始化完成后再注册所有蓝图，彻底规避蓝图与app的循环导入。

init.py 示例 #

from flask import Flask
from celery import Celery
from .config import Config

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    # 初始化Celery
    celery = Celery(
        app.import_name,
        broker=app.config["CELERY_BROKER_URL"],
        backend=app.config["CELERY_RESULT_BACKEND"]
    )
    celery.conf.update(app.config)

    # 注册蓝图
    from .auth import auth_bp
    from .views import main
    app.register_blueprint(auth_bp)
    app.register_blueprint(main)

    return app, celery

4. 集中管理配置（config.py） #

将Azure AD与Celery配置统一放在配置文件中，避免分散在各个模块：

class Config:
    # Azure AD配置
    AUTHORITY = "https://login.microsoftonline.com/your-tenant-id"
    CLIENT_ID = "your-client-id"
    CLIENT_SECRET = "your-client-secret"
    SCOPE = ["User.Read"]

    # Celery配置
    CELERY_BROKER_URL = "redis://localhost:6379/0"
    CELERY_RESULT_BACKEND = "redis://localhost:6379/0"

app_config = Config()

关键问题说明 #

1. 原代码的核心错误：模块级别初始化auth时，current_app尚未绑定请求上下文，会触发RuntimeError: Working outside of application context.，必须将Auth实例创建延迟到请求上下文可用的阶段。
2. 循环导入的本质解决：通过工厂模式延迟app与蓝图的绑定，跨蓝图依赖采用函数内部导入而非模块级导入，彻底切断循环依赖链。

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