如何使用sqlmodel向数据库添加obj

Question

我是FastAPI的新手。如何使用sqlmodel和数据库包在db中创建记录？

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

    team_id: Optional[int] = Field(default=None, foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="heroes")

@app.post("/hero", description="Create new hero")
async def create_hero(data: Hero):
    hero = Hero(**data.dict())
    q = insert(Hero).values(**hero.dict())
    h_id = await db.execute(q)

当我最终尝试这样做的时候，它向我展示了：

asyncpg.exceptions.NotNullViolationError: null value in column "id" of relation "hero" violates not-null constraint
DETAIL:  Failing row contains (null, spider, black, 18, null).

参考sqlmodel文档，id将自动设置，但使用sqlmodel.Session.如何做同样的事情

import databases


db = databases.Database("postgresql+asyncpg://postgres:postgres@localhost:5432/testdb")

Answer 1

正如一些评论所建议的那样，您可能不应该将databases与SQLModel一起使用。SQLModel的优点在于数据库插入非常简单:您只需将模型对象添加到数据库会话并提交即可。

另一个建议是使用FastAPI 依赖关系自动初始化、启动、注入并最终关闭每个路由中的会话。

下面是一个有用的例子：

from typing import Optional

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio.engine import create_async_engine
from sqlalchemy.ext.asyncio.session import AsyncSession
from sqlalchemy.orm.session import sessionmaker
from sqlmodel import Field, SQLModel


api = FastAPI()

db_uri = "postgresql+asyncpg://postgres:postgres@localhost:5432/testdb"
engine = create_async_engine(db_uri, future=True)
session_maker = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


@api.on_event('startup')
async def initialize_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)
        await conn.run_sync(SQLModel.metadata.create_all)


async def get_session() -> AsyncSession:
    session = session_maker()
    try:
        yield session
    finally:
        await session.close()


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

    ...


@api.post("/hero", description="Create new hero")
async def create_hero(hero: Hero, session: AsyncSession = Depends(get_session)):
    session.add(hero)
    await session.commit()
    await session.refresh(hero)
    return hero

请注意，出于测试目的，我只需在启动时删除并(重新)创建所有数据库表。

希望这能有所帮助。