如何在API端点被触发时执行Python脚本？（Gmail API场景）

核心逻辑很简单：在Web框架的路由处理函数中，直接调用你封装好的Gmail API请求代码，无需单独触发外部脚本。以下是具体实现：

用Flask实现的完整示例 #

1. 先安装依赖：

pip install flask google-api-python-client google-auth-httplib2 google-auth-oauthlib

2. 编写API代码：

from flask import Flask, jsonify
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials

app = Flask(__name__)

# 封装Gmail API调用逻辑
def get_gmail_data():
    # 替换成你的Gmail认证逻辑（参考Google官方文档生成token.json）
    creds = Credentials.from_authorized_user_file('token.json', ['https://www.googleapis.com/auth/gmail.readonly'])
    service = build('gmail', 'v1', credentials=creds)
    
    # 获取最近5封邮件的基础信息
    message_list = service.users().messages().list(userId='me', maxResults=5).execute()
    messages = message_list.get('messages', [])
    
    # 提取邮件主题、摘要等信息
    email_list = []
    for msg in messages:
        msg_detail = service.users().messages().get(userId='me', id=msg['id']).execute()
        subject = next(h['value'] for h in msg_detail['payload']['headers'] if h['name'] == 'Subject')
        email_list.append({
            'email_id': msg['id'],
            'subject': subject,
            'snippet': msg_detail['snippet']
        })
    return email_list

# 定义/email端点
@app.route('/email', methods=['GET'])
def email_handler():
    # 触发Gmail数据获取逻辑
    emails = get_gmail_data()
    # 将结果封装成响应返回
    return jsonify({'status': 'success', 'emails': emails})

if __name__ == '__main__':
    app.run(debug=True)

用FastAPI实现的完整示例 #

如果偏好FastAPI，代码逻辑一致：

1. 安装依赖：

pip install fastapi uvicorn google-api-python-client google-auth-httplib2 google-auth-oauthlib

2. 编写API代码：

from fastapi import FastAPI
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials

app = FastAPI()

def get_gmail_data():
    creds = Credentials.from_authorized_user_file('token.json', ['https://www.googleapis.com/auth/gmail.readonly'])
    service = build('gmail', 'v1', credentials=creds)
    
    message_list = service.users().messages().list(userId='me', maxResults=5).execute()
    messages = message_list.get('messages', [])
    
    email_list = []
    for msg in messages:
        msg_detail = service.users().messages().get(userId='me', id=msg['id']).execute()
        subject = next(h['value'] for h in msg_detail['payload']['headers'] if h['name'] == 'Subject')
        email_list.append({
            'email_id': msg['id'],
            'subject': subject,
            'snippet': msg_detail['snippet']
        })
    return email_list

@app.get('/email')
async def email_handler():
    emails = get_gmail_data()
    return {'status': 'success', 'emails': emails}

关键要点 #

• 把Gmail API的请求逻辑封装成独立函数，避免路由处理函数过于臃肿。
• 当客户端访问/email端点时，框架会自动执行对应的处理函数，函数内部直接调用封装好的Gmail数据获取逻辑。
• 如果你的Gmail请求代码是单独的外部脚本（比如gmail_fetcher.py），只需在API代码中导入脚本里的函数即可调用：

  from gmail_fetcher import get_gmail_data
  

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