SQLAlchemy 2.0异步PostgreSQL外键与关联关系问题求助

1. 修正ORM模型的关联关系 #

原Quotes模型的关联写法存在语法错误，需要明确外键字段并正确配置关联关系，确保通过unique_account_number实现两表关联：

from sqlalchemy import ForeignKey, UniqueConstraint, DateTime
from sqlalchemy.orm import relationship, Mapped, mapped_column
from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Customers(Base):
    __tablename__ = "customers"
    __table_args__ = (UniqueConstraint("unique_account_number"),)
    id = mapped_column(Integer, primary_key=True)
    customer_name: Mapped[str]
    unique_account_number: Mapped[str]
    updated_datetime = mapped_column(
        DateTime, default=datetime.now, onupdate=datetime.now
    )
    created_datetime = mapped_column(DateTime, default=datetime.now)

    # 可选反向关联，方便从客户维度查询关联报价
    quotes = relationship("Quotes", back_populates="customer")


class Quotes(Base):
    __tablename__ = "quotes"

    id = mapped_column(Integer, primary_key=True)
    origin: Mapped[str]
    destination: Mapped[str]
    # 新增外键字段，关联Customers表的unique_account_number
    unique_account_number = mapped_column(
        String, ForeignKey("customers.unique_account_number"), nullable=False
    )
    # 配置正向关联，与Customers的反向关联对应
    customer = relationship("Customers", back_populates="quotes")
    updated_datetime = mapped_column(
        DateTime, default=datetime.now, onupdate=datetime.now
    )
    created_datetime = mapped_column(DateTime, default=datetime.now)

2. 实现Quotes的插入逻辑 #

直接接收指定JSON格式的数据，通过外键字段unique_account_number关联已有客户记录：

异步插入代码示例 #

from sqlalchemy.ext.asyncio import AsyncSession

async def create_quote(db: AsyncSession, quote_data: dict):
    new_quote = Quotes(
        origin=quote_data["origin"],
        destination=quote_data["destination"],
        unique_account_number=quote_data["unique_account_number"]
    )
    db.add(new_quote)
    await db.commit()
    await db.refresh(new_quote)
    return new_quote

如果需要提前校验客户是否存在，可先查询客户记录：

async def create_quote(db: AsyncSession, quote_data: dict):
    # 先验证客户账号是否存在
    stmt = select(Customers).where(Customers.unique_account_number == quote_data["unique_account_number"])
    customer = await db.scalar(stmt)
    if not customer:
        raise ValueError("指定的客户账号不存在")
    
    new_quote = Quotes(
        origin=quote_data["origin"],
        destination=quote_data["destination"],
        customer=customer
    )
    db.add(new_quote)
    await db.commit()
    await db.refresh(new_quote)
    return new_quote

3. 实现嵌套格式的查询逻辑 #

使用预加载方式关联查询Customers表，避免N+1查询问题，并序列化为期望的嵌套JSON格式：

异步查询代码示例 #

from sqlalchemy import select
from sqlalchemy.orm import selectinload

async def get_all_quotes(db: AsyncSession):
    # 用selectinload预加载关联的客户数据，提升查询性能
    stmt = select(Quotes).options(selectinload(Quotes.customer))
    result = await db.execute(stmt)
    quotes = result.scalars().all()
    
    # 序列化为目标格式
    return [
        {
            "origin": q.origin,
            "destination": q.destination,
            "customer": {
                "unique_account_number": q.customer.unique_account_number,
                "customer_name": q.customer.customer_name
            }
        }
        for q in quotes
    ]

单条Quote查询示例 #

async def get_quote_by_id(db: AsyncSession, quote_id: int):
    stmt = select(Quotes).where(Quotes.id == quote_id).options(selectinload(Quotes.customer))
    quote = await db.scalar(stmt)
    if not quote:
        return None
    
    return {
        "origin": quote.origin,
        "destination": quote.destination,
        "customer": {
            "unique_account_number": quote.customer.unique_account_number,
            "customer_name": quote.customer.customer_name
        }
    }

注：预加载方式可根据场景选择：selectinload适合批量查询，会触发批量加载关联数据；joinedload会通过JOIN语句一次性加载所有数据，适合单条查询场景。

4. 异常处理（可选） #

插入时若传入无效的客户账号，会触发外键约束异常，可捕获并返回友好提示：

from sqlalchemy.exc import IntegrityError

async def create_quote(db: AsyncSession, quote_data: dict):
    new_quote = Quotes(
        origin=quote_data["origin"],
        destination=quote_data["destination"],
        unique_account_number=quote_data["unique_account_number"]
    )
    db.add(new_quote)
    try:
        await db.commit()
        await db.refresh(new_quote)
        return new_quote
    except IntegrityError:
        await db.rollback()
        raise ValueError("指定的客户账号不存在")

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