Django生产环境BigQuery Google Auth无法跳转及回调异常问题

问题详情 #

本地运行Django调用BigQuery查询一切正常，但部署到生产服务器后遇到两个问题：

• 无法自动跳转到Google验证链接
• 手动打开验证链接完成授权后，无法正常回调到应用
• 修改appflow的redirect_uri参数后，生成的授权链接依然用localhost:8080作为回调地址
• 已在Google Cloud控制台添加正确的授权重定向URI，并将API设置为生产模式
• 生成的授权链接示例：

https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=123-nml31ekr2n0didomei5.apps.googleusercontent.com&redirect_uri=http%3A%2F%2Flocalhost%3A8080%2F&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fbigquery&state=XF1WdtCoR4HaICwzSKk9A1giBrSzBv&access_type=offline

• 相关代码片段：

def query_stackoverflow():
    launch_browser = True
    project = 'xx-prod'
    appflow = flow.InstalledAppFlow.from_client_secrets_file(
        "static/client_secret_518684-nmpoqtgo5flvcgnl31ekr2ni5.apps.googleusercontent.com.json", 
        scopes=["https://www.googleapis.com/auth/bigquery"], 
        redirect_uri=["https://xx.com/"]
    )

    if launch_browser:
        appflow.run_local_server()
    else:
        appflow.run_console()
    credentials = appflow.credentials
    client = bigquery.Client(project=project, credentials=credentials)
    client = bigquery.Client()
    query_job = client.query(
        """
        SELECT
          CONCAT(
            'https://stackoverflow.com/questions/',
            CAST(id as STRING)) as url,
          view_count
        FROM `bigquery-public-data.stackoverflow.posts_questions`
        WHERE tags like '%google-bigquery%'
        ORDER BY view_count DESC
        LIMIT 10"""
    )

    results = query_job.result()  # Waits for job to complete.

    for row in results:
        print("{} : {} views".format(row.url, row.view_count))

核心问题 #

你用错了授权流程类：InstalledAppFlow是专门给本地桌面应用设计的，它的run_local_server()方法会强制覆盖你设置的redirect_uri，默认用localhost:8080。生产服务器上的Web应用必须用WebAppFlow类来适配标准的OAuth 2.0 Web应用流程。

解决步骤 #

1. 替换为WebAppFlow并重写授权逻辑 #

修改代码，拆分授权发起、回调处理和查询执行三个部分：

from google_auth_oauthlib.flow import Flow
from google.oauth2.credentials import Credentials
from django.http import HttpResponseRedirect, HttpResponse
from django.shortcuts import render
from django.conf import settings
import os
import bigquery

# 初始化授权Flow
def get_auth_flow(request):
    return Flow.from_client_secrets_file(
        os.path.join(settings.STATIC_ROOT, "client_secret_518684-nmpoqtgo5flvcgnl31ekr2ni5.apps.googleusercontent.com.json"),
        scopes=["https://www.googleapis.com/auth/bigquery"],
        redirect_uri="https://xx.com/oauth2callback/"  # 必须和Google Cloud控制台配置完全一致
    )

# 发起授权请求
def oauth_authorize(request):
    flow = get_auth_flow(request)
    # 生成授权链接，请求离线访问权限（获取refresh_token）
    auth_url, state = flow.authorization_url(
        access_type='offline',
        include_granted_scopes='true'
    )
    # 保存state到session，回调时验证防止CSRF攻击
    request.session['oauth_state'] = state
    return HttpResponseRedirect(auth_url)

# 处理授权回调
def oauth2callback(request):
    saved_state = request.session.get('oauth_state')
    current_state = request.GET.get('state')
    
    if not saved_state or saved_state != current_state:
        return HttpResponse('无效的state参数', status=400)
    
    flow = get_auth_flow(request)
    # 用回调地址的完整URL获取token
    flow.fetch_token(authorization_response=request.build_absolute_uri())
    
    # 将凭证信息保存到session（生产环境建议加密后存数据库）
    credentials = flow.credentials
    request.session['bigquery_credentials'] = {
        'token': credentials.token,
        'refresh_token': credentials.refresh_token,
        'token_uri': credentials.token_uri,
        'client_id': credentials.client_id,
        'client_secret': credentials.client_secret,
        'scopes': credentials.scopes
    }
    
    # 回调完成后跳转到查询页面
    return HttpResponseRedirect('/query_stackoverflow/')

# 执行BigQuery查询
def query_stackoverflow(request):
    # 检查是否已有授权凭证
    creds_info = request.session.get('bigquery_credentials')
    if not creds_info:
        # 没有凭证就跳转到授权页面
        return HttpResponseRedirect('/oauth_authorize/')
    
    # 从session还原凭证
    credentials = Credentials(**creds_info)
    project = 'xx-prod'
    client = bigquery.Client(project=project, credentials=credentials)
    
    query_job = client.query(
        """
        SELECT
          CONCAT(
            'https://stackoverflow.com/questions/',
            CAST(id as STRING)) as url,
          view_count
        FROM `bigquery-public-data.stackoverflow.posts_questions`
        WHERE tags like '%google-bigquery%'
        ORDER BY view_count DESC
        LIMIT 10"""
    )
    
    results = query_job.result()
    # 将结果传递给模板渲染（替换成你的模板路径）
    return render(request, 'query_results.html', {'results': results})

2. 配置Django路由 #

在项目的urls.py中添加三个路由：

from django.urls import path
from . import views

urlpatterns = [
    path('oauth_authorize/', views.oauth_authorize, name='oauth_authorize'),
    path('oauth2callback/', views.oauth2callback, name='oauth2callback'),
    path('query_stackoverflow/', views.query_stackoverflow, name='query_stackoverflow'),
]

3. 验证Google Cloud控制台配置 #

• 确保OAuth客户端类型是Web应用（不要选桌面应用）
• 授权重定向URI列表中已添加https://xx.com/oauth2callback/（注意末尾的斜杠要和代码中一致）

4. 解决自动跳转问题 #

生产服务器上无法自动打开浏览器是正常的，Web应用的授权逻辑是引导用户访问/oauth_authorize/路由，直接重定向到Google的授权页面，而非依赖run_local_server()这种本地桌面应用的行为。

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