Django中如何内存存储Pyrogram的app实例并跨用户共享？

针对你用Django + django-ninja开发API，需要复用Pyrogram Client实例避免重复创建会话的需求，下面是几种可行的实现方案：

方案一：带锁的全局模块变量（推荐单进程异步部署） #

创建一个单独的模块管理Pyrogram Client实例，用异步锁保证初始化的原子性，避免多请求同时创建实例。

1. 创建telegram_client.py模块：

from pyrogram import Client
import asyncio
from your_project.settings import config  # 替换成你的配置路径

# 全局存储Client实例
_pyrogram_app = None
# 异步锁，确保初始化操作唯一
_init_lock = asyncio.Lock()

async def get_pyrogram_client():
    global _pyrogram_app
    async with _init_lock:
        # 检查实例是否存在且已连接
        if _pyrogram_app is None or not _pyrogram_app.is_connected:
            _pyrogram_app = Client("your_session_name", config['api_id'], config['api_hash'])
            await _pyrogram_app.start()
    return _pyrogram_app

2. 在django-ninja视图中使用：

from ninja import NinjaAPI
from .telegram_client import get_pyrogram_client

api = NinjaAPI()

@api.get("/telegram/send-message")
async def send_telegram_message(request, chat_id: int, text: str):
    app = await get_pyrogram_client()
    message = await app.send_message(chat_id=chat_id, text=text)
    return {"message_id": message.id, "status": "success"}

3. 管理生命周期（启动/关闭）：
   借助django-ninja的生命周期钩子，在服务启动时初始化，关闭时清理连接：

# 在你的api模块中
@api.lifespan
async def pyrogram_lifespan(api):
    # 服务启动时预初始化客户端
    await get_pyrogram_client()
    yield
    # 服务关闭时停止客户端
    from .telegram_client import _pyrogram_app
    if _pyrogram_app and _pyrogram_app.is_connected:
        await _pyrogram_app.stop()

方案二：Django内存缓存（适合单进程同步/异步部署） #

利用Django自带的LocMemCache存储实例，适合简单场景，但同样只支持单进程。

1. 初始化并缓存实例：

from django.core.cache import cache
from pyrogram import Client
from your_project.settings import config

async def init_pyrogram_client():
    app = Client("your_session_name", config['api_id'], config['api_hash'])
    await app.start()
    cache.set("pyrogram_shared_client", app, timeout=None)  # 永久存储

# 在AppConfig的ready方法中触发初始化（注意异步转同步）
# apps.py
from django.apps import AppConfig
import asyncio

class YourAppConfig(AppConfig):
    name = "your_app"

    def ready(self):
        asyncio.run(init_pyrogram_client())

2. 视图中使用：

@api.get("/telegram/get-me")
async def get_telegram_user(request):
    app = cache.get("pyrogram_shared_client")
    # 容错：如果缓存丢失，重新初始化
    if not app or not app.is_connected:
        await init_pyrogram_client()
        app = cache.get("pyrogram_shared_client")
    user = await app.get_me()
    return {"user_id": user.id, "username": user.username}

方案三：多进程部署的解决方案 #

如果你的API是多进程部署（比如Gunicorn启动多个worker），上面的单进程方案会导致每个worker创建独立的Client实例，可能触发Pyrogram的会话数限制。这种情况下有两种选择：

• 控制worker数量：确保worker总数不超过Pyrogram允许的会话数上限。
• 独立托管Pyrogram服务：单独搭建一个异步服务（比如用FastAPI）专门管理Pyrogram Client，Django API通过HTTP请求调用该服务的接口，实现所有请求共享一个实例。

注意事项 #

• 连接状态检查：定期验证Client是否处于连接状态，断开时自动重新初始化。
• 资源泄漏：务必在服务关闭时调用await app.stop()释放资源，避免占用Telegram会话。

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