async SQLAlchemy创建引擎报错 已装asyncpg仍调用psycopg2如何解决

根因分析 #

两个报错的根本原因是SQLAlchemy没有识别到你要使用asyncpg异步驱动：

• 未安装psycopg2时，默认的postgresql://协议匹配的是psycopg2同步驱动，所以会报找不到psycopg2的错误
• 安装psycopg2后，异步引擎检测到加载的是同步驱动，直接抛出不兼容错误

修复步骤 #

1. 修改数据库连接字符串前缀 #

把init_connection方法里的连接字符串前缀从postgresql://改为postgresql+asyncpg://，明确指定使用asyncpg异步驱动，修改后的代码如下：

def init_connection(self, user, password, host, port, db):
    """Connect to actual database."""
    # 修改前缀指定asyncpg驱动
    connection_string = "postgresql+asyncpg://{}:{}@{}:{}/{}".format(
        user, password, host, port, db
    )
    engine = create_async_engine(connection_string, future=True, echo=True)
    self._session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

2. 修正add_task方法的session使用逻辑 #

现有add_task方法里直接调用self._session.add()是错误的，self._session是session工厂，不是实际的会话实例，需要和get_active_tasks一样用上下文管理器打开会话：

async def add_task(self, author, channel_id, count, is_dm):
    """Add new task to db."""
    now = datetime.utcnow()
    task = TaskModel(
        author=author,
        channel_id=channel_id,
        is_dm=is_dm,
        start_time=now,
        end_time=now + timedelta(seconds=count),
        count=count,
        canceled=False
    )
    # 正确打开会话操作
    async with self._session() as session:
        session.add(task)
        await session.commit()
    logger.info(f"task added to db: {task}")
    return task

3. 可选清理 #

不需要的psycopg2可以直接卸载，避免后续干扰：
pip uninstall psycopg2 psycopg2-binary -y

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