Movatterモバイル変換

[0]ホーム
URL:

Skip to content
Follow@fastapi onTwitter to stay updated
FollowFastAPI onLinkedIn to stay updated
Subscribe to theFastAPI and friends newsletter 🎉
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor

SQL (Relational) Databases

FastAPI doesn't require you to use a SQL (relational) database. But you can useany database that you want.

Here we'll see an example usingSQLModel.

SQLModel is built on top ofSQLAlchemy and Pydantic. It was made by the same author ofFastAPI to be the perfect match for FastAPI applications that need to useSQL databases.

Tip

You could use any other SQL or NoSQL database library you want (in some cases called"ORMs"), FastAPI doesn't force you to use anything. 😎

As SQLModel is based on SQLAlchemy, you can easily useany database supported by SQLAlchemy (which makes them also supported by SQLModel), like:

In this example, we'll useSQLite, because it uses a single file and Python has integrated support. So, you can copy this example and run it as is.

Later, for your production application, you might want to use a database server likePostgreSQL.

Tip

There is an official project generator withFastAPI andPostgreSQL including a frontend and more tools:https://github.com/fastapi/full-stack-fastapi-template

This is a very simple and short tutorial, if you want to learn about databases in general, about SQL, or more advanced features, go to theSQLModel docs.

InstallSQLModel

First, make sure you create yourvirtual environment, activate it, and then installsqlmodel:

$pipinstallsqlmodel---> 100%

Create the App with a Single Model

We'll create the simplest first version of the app with a singleSQLModel model first.

Later we'll improve it increasing security and versatility withmultiple models below. 🤓

Create Models

ImportSQLModel and create a database model:

fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:str# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

TheHero class is very similar to a Pydantic model (in fact, underneath, it actuallyis a Pydantic model).

There are a few differences:

  • table=True tells SQLModel that this is atable model, it should represent atable in the SQL database, it's not just adata model (as would be any other regular Pydantic class).

  • Field(primary_key=True) tells SQLModel that theid is theprimary key in the SQL database (you can learn more about SQL primary keys in the SQLModel docs).

    By having the type asint | None, SQLModel will know that this column should be anINTEGER in the SQL database and that it should beNULLABLE.

  • Field(index=True) tells SQLModel that it should create aSQL index for this column, that would allow faster lookups in the database when reading data filtered by this column.

    SQLModel will know that something declared asstr will be a SQL column of typeTEXT (orVARCHAR, depending on the database).

Create an Engine

A SQLModelengine (underneath it's actually a SQLAlchemyengine) is whatholds the connections to the database.

You would haveone singleengine object for all your code to connect to the same database.

# Code above omitted 👆sqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Usingcheck_same_thread=False allows FastAPI to use the same SQLite database in different threads. This is necessary asone single request could usemore than one thread (for example in dependencies).

Don't worry, with the way the code is structured, we'll make sure we usea single SQLModelsession per request later, this is actually what thecheck_same_thread is trying to achieve.

Create the Tables

We then add a function that usesSQLModel.metadata.create_all(engine) tocreate the tables for all thetable models.

# Code above omitted 👆defcreate_db_and_tables():SQLModel.metadata.create_all(engine)# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Create a Session Dependency

ASession is what stores theobjects in memory and keeps track of any changes needed in the data, then ituses theengine to communicate with the database.

We will create a FastAPIdependency withyield that will provide a newSession for each request. This is what ensures that we use a single session per request. 🤓

Then we create anAnnotated dependencySessionDep to simplify the rest of the code that will use this dependency.

# Code above omitted 👆defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Create Database Tables on Startup

We will create the database tables when the application starts.

# Code above omitted 👆app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Here we create the tables on an application startup event.

For production you would probably use a migration script that runs before you start your app. 🤓

Tip

SQLModel will have migration utilities wrapping Alembic, but for now, you can useAlembic directly.

Create a Hero

Because each SQLModel model is also a Pydantic model, you can use it in the sametype annotations that you could use Pydantic models.

For example, if you declare a parameter of typeHero, it will be read from theJSON body.

The same way, you can declare it as the function'sreturn type, and then the shape of the data will show up in the automatic API docs UI.

# Code above omitted 👆@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Here we use theSessionDep dependency (aSession) to add the newHero to theSession instance, commit the changes to the database, refresh the data in thehero, and then return it.

Read Heroes

We canreadHeros from the database using aselect(). We can include alimit andoffset to paginate the results.

# Code above omitted 👆@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Read One Hero

We canread a singleHero.

# Code above omitted 👆@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Delete a Hero

We can alsodelete aHero.

# Code above omitted 👆@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(SQLModel,table=True):id:Union[int,None]=Field(default=None,primary_key=True)name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)secret_name:strsqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->List[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Run the App

You can run the app:

$fastapidevmain.py<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Then go to the/docs UI, you will see thatFastAPI is using thesemodels todocument the API, and it will use them toserialize andvalidate the data too.

Update the App with Multiple Models

Now let'srefactor this app a bit to increasesecurity andversatility.

If you check the previous app, in the UI you can see that, up to now, it lets the client decide theid of theHero to create. 😱

We shouldn't let that happen, they could overwrite anid we already have assigned in the DB. Deciding theid should be done by thebackend or thedatabase,not by the client.

Additionally, we create asecret_name for the hero, but so far, we are returning it everywhere, that's not verysecret... 😅

We'll fix these things by adding a fewextra models. Here's where SQLModel will shine. ✨

Create Multiple Models

InSQLModel, any model class that hastable=True is atable model.

And any model class that doesn't havetable=True is adata model, these ones are actually just Pydantic models (with a couple of small extra features). 🤓

With SQLModel, we can useinheritance toavoid duplicating all the fields in all the cases.

HeroBase - the base class

Let's start with aHeroBase model that has all thefields that are shared by all the models:

  • name
  • age
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Hero - thetable model

Then let's createHero, the actualtable model, with theextra fields that are not always in the other models:

  • id
  • secret_name

BecauseHero inherits formHeroBase, italso has thefields declared inHeroBase, so all the fields forHero are:

  • id
  • name
  • age
  • secret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:str# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

HeroPublic - the publicdata model

Next, we create aHeroPublic model, this is the one that will bereturned to the clients of the API.

It has the same fields asHeroBase, so it won't includesecret_name.

Finally, the identity of our heroes is protected! 🥷

It also re-declaresid: int. By doing this, we are making acontract with the API clients, so that they can always expect theid to be there and to be anint (it will never beNone).

Tip

Having the return model ensure that a value is always available and alwaysint (notNone) is very useful for the API clients, they can write much simpler code having this certainty.

Also,automatically generated clients will have simpler interfaces, so that the developers communicating with your API can have a much better time working with your API. 😎

All the fields inHeroPublic are the same as inHeroBase, withid declared asint (notNone):

  • id
  • name
  • age
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:int# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

HeroCreate - thedata model to create a hero

Now we create aHeroCreate model, this is the one that willvalidate the data from the clients.

It has the same fields asHeroBase, and it also hassecret_name.

Now, when the clientscreate a new hero, they will send thesecret_name, it will be stored in the database, but those secret names won't be returned in the API to the clients.

Tip

This is how you would handlepasswords. Receive them, but don't return them in the API.

You would alsohash the values of the passwords before storing them,never store them in plain text.

The fields ofHeroCreate are:

  • name
  • age
  • secret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:str# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

HeroUpdate - thedata model to update a hero

We didn't have a way toupdate a hero in the previous version of the app, but now withmultiple models, we can do it. 🎉

TheHeroUpdatedata model is somewhat special, it hasall the same fields that would be needed to create a new hero, but all the fields areoptional (they all have a default value). This way, when you update a hero, you can send just the fields that you want to update.

Because all thefields actually change (the type now includesNone and they now have a default value ofNone), we need tore-declare them.

We don't really need to inherit fromHeroBase because we are re-declaring all the fields. I'll leave it inheriting just for consistency, but this is not necessary. It's more a matter of personal taste. 🤷

The fields ofHeroUpdate are:

  • name
  • age
  • secret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=None# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Create withHeroCreate and return aHeroPublic

Now that we havemultiple models, we can update the parts of the app that use them.

We receive in the request aHeroCreatedata model, and from it, we create aHerotable model.

This newtable modelHero will have the fields sent by the client, and will also have anid generated by the database.

Then we return the sametable modelHero as is from the function. But as we declare theresponse_model with theHeroPublicdata model,FastAPI will useHeroPublic to validate and serialize the data.

# Code above omitted 👆@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Now we useresponse_model=HeroPublic instead of thereturn type annotation-> HeroPublic because the value that we are returning is actuallynot aHeroPublic.

If we had declared-> HeroPublic, your editor and linter would complain (rightfully so) that you are returning aHero instead of aHeroPublic.

By declaring it inresponse_model we are tellingFastAPI to do its thing, without interfering with the type annotations and the help from your editor and other tools.

Read Heroes withHeroPublic

We can do the same as before toreadHeros, again, we useresponse_model=list[HeroPublic] to ensure that the data is validated and serialized correctly.

# Code above omitted 👆@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Read One Hero withHeroPublic

We canread a single hero:

# Code above omitted 👆@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Update a Hero withHeroUpdate

We canupdate a hero. For this we use an HTTPPATCH operation.

And in the code, we get adict with all the data sent by the client,only the data sent by the client, excluding any values that would be there just for being the default values. To do it we useexclude_unset=True. This is the main trick. 🪄

Then we usehero_db.sqlmodel_update(hero_data) to update thehero_db with the data fromhero_data.

# Code above omitted 👆@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db# Code below omitted 👇
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Delete a Hero Again

Deleting a hero stays pretty much the same.

We won't satisfy the desire to refactor everything in this one. 😅

# Code above omitted 👆@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
🤓 Other versions and variants
fromtypingimportAnnotated,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}
fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectfromtyping_extensionsimportAnnotatedclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Tip

Prefer to use theAnnotated version if possible.

fromtypingimportList,UnionfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:Union[int,None]=Field(default=None,index=True)classHero(HeroBase,table=True):id:Union[int,None]=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:Union[str,None]=Noneage:Union[int,None]=Nonesecret_name:Union[str,None]=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=List[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}

Run the App Again

You can run the app again:

$fastapidevmain.py<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

If you go to the/docs API UI, you will see that it is now updated, and it won't expect to receive theid from the client when creating a hero, etc.

Recap

You can useSQLModel to interact with a SQL database and simplify the code withdata models andtable models.

You can learn a lot more at theSQLModel docs, there's a longer minitutorial on using SQLModel withFastAPI. 🚀

[8]ページ先頭
©2009-2025 Movatter.jp