FastAPI结合异步SQLAlchemy操作MySQL遇__aenter__错误求助

我用翻译软件表述可能有错误，见谅。我按照网络教程思路操作，但无法使用异步SQLAlchemy操作MySQL，测试时出现AttributeError: __aenter__错误，麻烦帮忙指出问题。

数据库连接代码 #

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from urllib import  parse
import  sys

engine = create_async_engine(Sql_URL)


SessionLocal = sessionmaker(bind=engine,autocommit=False,autoflush=False)
db_session = scoped_session(SessionLocal)

Base = declarative_base()
Base.query = db_session.query_property()

模型代码 #

import asyncio
import datetime

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, MetaData
from sqlalchemy.orm import registry
from  sqlalchemy_utils import EmailType,ChoiceType

from app.Fast_blog.database import Base


class user(Base):
    choices = [
        ('0', 'woman'),
        ('1', 'man'),
        ('2', 'NULL')
    ]
    __tablename__ = "usertable"
    __table_args__ = {'extend_existing': True}
    UserId = Column(Integer,primary_key = True,index = True)
    username = Column(String(255))
    userpassword = Column(String(255))
    gender = Column(ChoiceType(choices))
    creation_time = Column(DateTime,default = datetime.datetime.now)
    Last_Login_Time = Column(DateTime,default= datetime.datetime.now)
    UserUuid = Column(String(255))
    UserEmail = Column(EmailType(255))

class Blog(Base):
    __tablename__ = "blogtable"
    __table_args__ = {'extend_existing': True}
    title = Column(String(255))
    content = Column(String(255))
    author = Column(String(255))
    BlogId = Column(String(255),primary_key=True,index=True)
    BlogUuid = Column(String(32))

视图代码 #

import uuid
from typing import Generator

from sqlalchemy.ext.asyncio import async_session
from sqlalchemy import select

from app.Fast_blog import model
from app.Fast_blog.model import models
from app.Fast_blog.database.database import engine
from sqlalchemy.orm import sessionmaker
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import EmailStr
from  fastapi import APIRouter

from app.Fast_blog.model.models import user

SessionLocal = sessionmaker(autocommit=False,autoflush=False,bind=engine)
Session = SessionLocal()


UserApp = APIRouter()

templates = Jinja2Templates(directory="./Fast_blog/templates")

UserApp.mount("/static", StaticFiles(directory="./Fast_blog/static"), name="static")


def UUID_crt(UuidApi):
    x = uuid.uuid5(uuid.NAMESPACE_DNS,UuidApi)
    return x

@UserApp.get("/")
async def query():
    async with Session as session:
        sql = select(model).where(model.id == 1)
        print(sql)
        result = await session.execute(sql)
        data = result.scalars().first()
        # data = result.scalars().all()

报错信息 #

Traceback (most recent call last):
  File "d:\python3\lib\site-packages\uvicorn\protocols\http\h11_impl.py", line 404, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
  File "d:\python3\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 78, in __call__
    return await self.app(scope, receive, send)
  File "d:\python3\lib\site-packages\uvicorn\middleware\debug.py", line 106, in __call__
    raise exc from None
  File "d:\python3\lib\site-packages\uvicorn\middleware\debug.py", line 103, in __call__
    await self.app(scope, receive, inner_send)
  File "d:\python3\lib\site-packages\fastapi\applications.py", line 270, in __call__
    await super().__call__(scope, receive, send)
  File "d:\python3\lib\site-packages\starlette\applications.py", line 124, in __call__
    await self.middleware_stack(scope, receive, send)
  File "d:\python3\lib\site-packages\starlette\middleware\errors.py", line 184, in __call__
    raise exc
  File "d:\python3\lib\site-packages\starlette\middleware\errors.py", line 162, in __call__
    await self.app(scope, receive, _send)
  File "d:\python3\lib\site-packages\starlette\middleware\exceptions.py", line 79, in __call__
    raise exc
  File "d:\python3\lib\site-packages\starlette\middleware\exceptions.py", line 68, in __call__
    await self.app(scope, receive, sender)
  File "d:\python3\lib\site-packages\fastapi\middleware\asyncexitstack.py", line 21, in __call__
    raise e
  File "d:\python3\lib\site-packages\fastapi\middleware\asyncexitstack.py", line 18, in __call__
    await self.app(scope, receive, send)
  File "d:\python3\lib\site-packages\starlette\routing.py", line 706, in __call__
    await route.handle(scope, receive, send)
  File "d:\python3\lib\site-packages\starlette\routing.py", line 276, in handle
    await self.app(scope, receive, send)
  File "d:\python3\lib\site-packages\starlette\routing.py", line 66, in app
    response = await func(request)
  File "d:\python3\lib\site-packages\fastapi\routing.py", line 237, in app
    raw_response = await run_endpoint_function(
  File "d:\python3\lib\site-packages\fastapi\routing.py", line 163, in run_endpoint_function
    return await dependant.call(**values)
  File ".\Fast_blog\apps\User_app\user.py", line 37, in query
    async with Session as session:
AttributeError: __aenter__

错误分析与修复方案 #

核心错误点

1. 异步会话创建方式错误
   你用了同步的sessionmaker绑定异步引擎，这完全不符合异步SQLAlchemy的要求。异步场景必须使用AsyncSession对应的会话工厂，同步会话不支持async with上下文管理器，直接导致__aenter__属性不存在的报错。

2. 会话实例化时机错误
   你提前实例化了Session = SessionLocal()，并在接口中复用这个实例。每个请求都应该创建独立的异步会话，复用实例会引发协程安全问题。

3. 查询语句模型引用错误
   查询中的select(model)引用的是模块而非具体模型类，且你的user模型主键是UserId，不是id，字段匹配错误。

修正后的代码

数据库连接代码

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 异步MySQL必须用asyncmy驱动，URL格式：mysql+asyncmy://用户名:密码@主机:端口/数据库名
engine = create_async_engine(Sql_URL)

# 创建异步会话工厂
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False
)

Base = declarative_base()
# 异步场景下不要用scoped_session和query_property，不符合协程模型

视图代码

import uuid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.Fast_blog.model.models import user
from app.Fast_blog.database.database import AsyncSessionLocal
from fastapi import APIRouter, Depends

UserApp = APIRouter()

# 定义会话依赖，每个请求生成独立的异步会话
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

def UUID_crt(UuidApi):
    x = uuid.uuid5(uuid.NAMESPACE_DNS, UuidApi)
    return x

@UserApp.get("/")
async def query(db: AsyncSession = Depends(get_db)):
    # 使用正确的模型类和字段
    sql = select(user).where(user.UserId == 1)
    print(sql)
    result = await db.execute(sql)
    data = result.scalars().first()
    return {"data": data}

额外注意事项

• 必须安装异步MySQL驱动：pip install asyncmy，否则异步引擎无法连接MySQL
• 模型类建议用大驼峰命名（如User而非user），符合Python编码规范
• 异步场景下禁止使用scoped_session，它是基于线程本地存储实现的，不适用于协程模型

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