我正在做一个关于Fastapi的小项目(我的第一个项目)，python 3.10，使用sqlalchemy, sqlmodel。我有两个测试模型:商店和产品。我希望端点shop返回以下结构:
{
"name": "Adidas",
"id": 1,
"is_main": true,
"products": [
{
"id": 1,
"name": "sneakers",
"shop_id": 1
},
{
"id": 2,
"name": "T-shirt",
"shop_id": 1
}
]
}
Github上的完整测试项目
我使用Relationship() from SQLModel:
models
product.py
from sqlmodel import SQLModel, Field, Relationship
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from app.models.shop import Shop
class ProductBase(SQLModel):
name: Optional[str] = None
shop_id: int = Field(default=None, foreign_key="shop.id")
class Product(ProductBase, table=True):
__tablename__ = "product"
id: Optional[int] = Field(default=None, primary_key=True, nullable=False)
shop: Optional["Shop"] = Relationship(back_populates="products")
class ProductRead(ProductBase):
id: int
class ProductCreate(ProductBase):
pass
shop.py
from sqlmodel import SQLModel, Field, Relationship
from typing import TYPE_CHECKING, List, Optional
if TYPE_CHECKING:
from app.models.product import Product, ProductRead
class ShopBase(SQLModel):
name: Optional[str] = None
class Shop(ShopBase, table=True):
__tablename__ = "shop"
id: Optional[int] = Field(default=None, primary_key=True, nullable=False)
products: List["Product"] = Relationship(back_populates="shop")
class ShopRead(ShopBase):
id: int
class ShopGet(ShopRead):
products: List["ProductRead"] = []
is_main: bool = True
class ShopCreate(ShopBase):
pass
repositories
product.py
from typing import Optional
from sqlalchemy.future import select
from app.models.product import Product, ProductCreate
from app.repositories.base import BaseRepository
class ProductRepository(BaseRepository):
async def create(self, product: ProductCreate) -> Product:
db_product = Product.from_orm(product)
self.session.add(db_product)
await self.session.commit()
await self.session.refresh(db_product)
return db_product
async def get_by_id(self, product_id: int) -> Optional[Product]:
result = await self.session.get(Product, int(product_id))
return result
shop.py
from typing import Optional
from sqlalchemy.future import select
from app.models.shop import Shop, ShopCreate, ShopGet
from app.repositories.base import BaseRepository
class ShopRepository(BaseRepository):
async def create(self, shop: ShopCreate) -> Shop:
db_shop = Shop.from_orm(shop)
self.session.add(db_shop)
await self.session.commit()
await self.session.refresh(db_shop)
return db_shop
async def get_by_id(self, shop_id: int) -> Optional[ShopGet]:
result = await self.session.get(Shop, int(shop_id))
return result
endpoints
shop.py
from fastapi import APIRouter, Depends, Query
from app.repositories.shop import ShopRepository
from app.models.shop import Shop, ShopCreate, ShopGet
from app.endpoints.depends import get_shop_repository
router = APIRouter()
@router.get("/get_by_id", response_model=ShopGet)
async def get_by_id(
shop_id: int = Query(description="Shop ID"),
shop: ShopRepository = Depends(get_shop_repository)):
return await shop.get_by_id(shop_id=shop_id)
@router.post("/create", response_model=Shop)
async def create(
name: str = Query(description="Shop name"),
shop: ShopRepository = Depends(get_shop_repository)):
return await shop.create(shop=ShopCreate(name=name))
db connect
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession
from app import settings
async_engine = create_async_engine(settings.db_async_connection_str, echo=True, future=True)
async def get_async_session() -> AsyncSession:
async_session = sessionmaker(
bind=async_engine, class_=AsyncSession, expire_on_commit=False
)
async with async_session() as session:
yield session
我有两个问题:
- 打开swagger时异常
TypeError: issubclass() arg 1 must be a class。原因:
if TYPE_CHECKING:
from app.models.product import Product, ProductRead
class ShopGet(ShopRead):
products: List["ProductRead"] = []
但是由于循环导入，这是必要的
- 主要问题:
Endpointshop(shop_id=1)返回空产品列表。细节。数据库中有2个产品的shop_id=1
这个问题用俄语说
我认为问题是sqlalchemy加载关系很懒。在同步环境中，当您试图访问关系时，它可以加载关系，但在异步环境中这是不可能的。因此，您需要要求sqlalchemy显式预加载关系。
class ShopRepository(BaseRepository):
...
async def get_by_id(self, shop_id: int) -> Optional[ShopGet]:
query = select(Shop).where(Shop.id == shop_id).options(selectinload(Shop.products))
return (await session.execute(query)).scalars().one()