FastAPI+SQLAlchemy操作PostgreSQL报错:UUID对象无replace属性
解决FastAPI+SQLAlchemy查询时的UUID类型错误
问题原因
你遇到的'asyncpg.pgproto.pgproto.UUID' object has no attribute 'replace'错误,是因为asyncpg驱动返回的UUID是它内部的pgproto.UUID类型,而你的SQLAlchemy模型UserDB期望接收标准库uuid.UUID类型的id值,直接传入会触发类型不兼容的错误。
解决方案
方案1:手动转换UUID类型
在将查询结果映射到UserDB之前,把id字段转换成标准uuid.UUID对象:
async def get_user_by_email(email: str) -> Optional[UserDB]: query = UserTable.select().where(UserTable.c.email == email) try: user_record = await database.fetch_one(query) if user_record: user_dict = dict(user_record) # 将asyncpg的UUID转换为标准库UUID user_dict["id"] = uuid.UUID(str(user_dict["id"])) return UserDB(**user_dict) except Exception as e: print(f"Error: {e}") return None
方案2:全局配置asyncpg返回标准UUID
如果想避免每次查询都手动转换,可以在初始化databases.Database时配置类型转换规则,让asyncpg直接返回标准uuid.UUID:
修改db/database.py中的数据库初始化代码:
from databases import Database import asyncpg import uuid # 定义UUID类型的编解码器 def encode_uuid(uuid_val): return str(uuid_val) def decode_uuid(uuid_str): return uuid.UUID(uuid_str) # 初始化数据库连接时指定类型转换 database = Database( "postgresql://your_user:your_password@localhost/your_db", connect_args={ "init": lambda conn: conn.set_type_codec( "uuid", encoder=encode_uuid, decoder=decode_uuid, format="text" ) } )
方案3:切换到SQLAlchemy异步会话(推荐)
改用SQLAlchemy官方的异步会话API,它会自动处理类型映射,彻底避免这类问题:
from sqlalchemy.ext.asyncio import AsyncSession from models.user import User as UserDB from sqlalchemy.future import select from typing import Optional async def get_user_by_email(db: AsyncSession, email: str) -> Optional[UserDB]: result = await db.execute(select(UserDB).where(UserDB.email == email)) return result.scalar_one_or_none()
使用时只需将异步会话对象传入函数即可,FastAPI中可以通过依赖注入提供会话。
内容的提问来源于stack exchange,提问作者Rashid
相关产品推荐
相关产品推荐

