如何在FastAPI路由中运行Pyrogram异步方法？

让我们一步步拆解并解决你的问题，先从核心问题入手，再给出完整的修正方案：

你原代码里的几个关键问题 #

1. Pyrogram Client初始化缺失必要参数：你的UserBot类里创建Client时没传入api_id和api_hash，这是Telegram API要求的必填项，缺少它们会直接导致连接失败，自然看不到任何输出。
2. 错误的日志调用：logging.INFO()是错误用法，INFO是日志级别常量，正确的调用是logging.info("日志内容")，原代码这里会抛出异常，只是你没捕获到。
3. FastAPI路由里的错误调用方式：你直接返回了task1（任务对象），而不是任务执行后的结果；另外原get_chat_history里捕获异常后返回了Exception对象，这在FastAPI里无法正确序列化返回。
4. 滥用Client.run()：app.run()是Pyrogram的同步启动方法，会创建自己的asyncio事件循环，和FastAPI的现有循环冲突，在FastAPI的异步环境里绝对不能用这个方法。

修正后的完整代码实现 #

第一步：修复UserBot类 #

import logging
from typing import Optional, List
from pyrogram import Client
from pyrogram.types import Message

# 配置日志输出
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

class UserBot:
    def __init__(self, username: str, api_id: int, api_hash: str, debug: Optional[bool] = False) -> None:
        self.username = username
        # 必须传入Telegram API的api_id和api_hash
        self.app = Client(f"sessions/{username}", api_id=api_id, api_hash=api_hash)
        self.debug = debug

    async def get_chat_history(self, chat_id: str) -> List[Message]:
        messages = []
        try:
            # 用async with自动管理Client的启动和关闭
            async with self.app:
                async for message in self.app.get_chat_history(chat_id):
                    if self.debug:
                        # 调试模式下打印消息内容
                        print(f"消息ID: {message.id} | 内容: {message.text or message.caption}")
                    logging.info(f"已获取消息 {message.id} 来自对话 {chat_id}")
                    messages.append(message)
            return messages
        except Exception as e:
            logging.error(f"获取聊天记录失败: {str(e)}")
            # 抛出异常让FastAPI处理HTTP错误响应
            raise e

第二步：FastAPI路由的正确写法 #

from fastapi import FastAPI, HTTPException
# 替换成你的UserBot所在的模块路径
from your_module import UserBot

# 从环境变量或配置文件读取，不要硬编码！
API_ID = 123456  # 替换成你的Telegram API ID
API_HASH = "your_api_hash_string_here"  # 替换成你的Telegram API Hash

app = FastAPI()

@app.get("/get-chat-history/")
async def fetch_chat_history(chat_id: str = "@CryptoVedma"):
    try:
        # 初始化UserBot
        ubot = UserBot(username="donqhomo", api_id=API_ID, api_hash=API_HASH, debug=True)
        # 直接await异步方法，不需要手动创建loop或task
        raw_messages = await ubot.get_chat_history(chat_id=chat_id)
        
        # 把Pyrogram的Message对象转换成可序列化的字典（FastAPI无法直接返回复杂对象）
        serialized_messages = [
            {
                "message_id": msg.id,
                "content": msg.text or msg.caption,
                "send_time": msg.date.isoformat(),
                "sender_username": msg.from_user.username if msg.from_user else "匿名用户"
            }
            for msg in raw_messages
        ]
        
        return {
            "status": "success",
            "total_messages": len(serialized_messages),
            "messages": serialized_messages
        }
    except Exception as e:
        # 返回标准的HTTP错误响应
        raise HTTPException(status_code=500, detail=f"操作失败: {str(e)}")

额外优化：复用UserBot实例（提升性能） #

如果你的API请求频繁，每次请求都创建新的Client会浪费连接资源，可以把UserBot实例全局化，在FastAPI启动时初始化：

# 全局UserBot实例
ubot = UserBot(username="donqhomo", api_id=API_ID, api_hash=API_HASH, debug=True)

# FastAPI启动事件：初始化Client连接
@app.on_event("startup")
async def startup():
    await ubot.app.start()
    logging.info("UserBot已启动")

# FastAPI关闭事件：关闭Client连接
@app.on_event("shutdown")
async def shutdown():
    await ubot.app.stop()
    logging.info("UserBot已关闭")

@app.get("/get-chat-history/")
async def fetch_chat_history(chat_id: str = "@CryptoVedma"):
    try:
        messages = []
        async for message in ubot.app.get_chat_history(chat_id):
            if ubot.debug:
                print(f"消息ID: {message.id} | 内容: {message.text or message.caption}")
            logging.info(f"已获取消息 {message.id} 来自对话 {chat_id}")
            messages.append(message)
        
        # 序列化消息...
        serialized_messages = [...]
        return {"status": "success", ...}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"操作失败: {str(e)}")

关键要点总结 #

• 永远不要在FastAPI的异步路由里用Client.run()，必须用async with或者await app.start()/await app.stop()管理Client生命周期。
• Pyrogram的Message对象不能直接返回给FastAPI，必须转换成字典等可序列化类型。
• 异常要正确抛出，让FastAPI生成标准的HTTP错误响应，而不是返回Exception对象。
• 一定要确保api_id和api_hash正确配置，这是连接Telegram API的基础。

内容的提问来源于stack exchange，提问作者Иван