CODEWITHSUNDEEP
pythondatabasefastapipydanticsqlmodel
haku
haku

Reputation: 75

How to add obj to db with sqlmodel

I am new to FastAPI. How can i create record in db using sqlmodel and databases packages?

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)

When i finally try to do this, it shows me:

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

Referring to the sqlmodel docs, id will be set automatically, but using sqlmodel.Session. How to do the same thing with

import databases


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

Upvotes: 0

Views: 1742

Answers (1)

Daniil Fajnberg
Daniil Fajnberg

Reputation: 18643

As some comments suggested, you should probably not use databases together with SQLModel. The beauty of SQLModel is (among many other things) that the database inserts are so simple: You just add your model objects to a database session and commit.

Another suggestion is to make use of the FastAPI dependencies to automatically initialize, start, inject and finally close a session in every route.

Here is a working example:

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

Note that for testing purposes I simply drop and (re-)create all database tables on startup.

Hope this helps.

Upvotes: 1

Related Questions