小编Mic*_*han的帖子

我制作了一个小应用程序，用于SQLAlchemy处理与 postgresql 数据库的连接。现在我想用asincio重写它。由于某种原因，当我运行它时，出现以下错误：

Traceback (most recent call last):
  File "D:\Space\discord_count_bot\bot\bot\main.py", line 12, in <module>
    dbConnection.init_connection(
  File "D:\Space\discord_count_bot\bot\bot\db_hanler.py", line 78, in init_connection
    engine = create_async_engine(connection_string, future=True, echo=True)
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\ext\asyncio\engine.py", line 40, in create_async_engine
    sync_engine = _create_engine(*arg, **kw)
  File "<string>", line 2, in create_engine
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\util\deprecations.py", line 298, in warned
    return fn(*args, **kwargs)
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\engine\create.py", line 560, in create_engine
    dbapi = dialect_cls.dbapi(**dbapi_args)
  File "D:\Space\discord_count_bot\bot_env\lib\site-packages\sqlalchemy\dialects\postgresql\psycopg2.py", line 782, in dbapi
    import psycopg2
ModuleNotFoundError: No module named 'psycopg2'

如果psycopg2安装了，我得到

Traceback (most …

我遇到过这个问题，虽然它一定是一个常见问题，但我看不到任何解决方案。所以，也许我在这里遗漏了一些东西。

我正在开发具有异步端点和与数据库异步连接的 FastAPI 应用程序。数据库连接作为依赖项传递。我想为所述应用程序编写一些异步测试。

engine = create_async_engine(connection_string, echo=True)

def get_session():
    return sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@router.post("/register")
async def register(
    user_data: UserRequest,
    authorize: AuthJWT = Depends(),
    async_session: sessionmaker = Depends(get_session),
):
    """Register new user."""
    if authorize.get_jwt_subject():
        raise LogicException("already authorized")

    session: AsyncSession
    async with async_session() as session:
        query = await session.execute(
            select(UserModel).where(UserModel.name == user_data.name)
        )
    ...

我正在使用 AsyncSession 来处理数据库。所以在我的测试中，数据库连接也必须是异步的。

engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
app.dependency_overrides[get_session] = lambda: sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

@pytest.mark.asyncio
async def test_create_user():
    async with engine.begin() as conn: …