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:
- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server, etc.
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
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}TheHero class is very similar to a Pydantic model (in fact, underneath, it actuallyis a Pydantic model).
There are a few differences:
table=Truetells 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 theidis theprimary key in the SQL database (you can learn more about SQL primary keys in the SQLModel docs).Note: We use
int | Nonefor the primary key field so that in Python code we cancreate an object without anid(id=None), assuming the database willgenerate it when saving. SQLModel understands that the database will provide theidanddefines the column as a non-nullINTEGERin the database schema. SeeSQLModel docs on primary keys for details.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 as
strwill 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
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}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
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}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
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}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
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}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
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}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
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}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
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}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
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}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:
nameage
# 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
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}Hero - thetable model¶
Then let's createHero, the actualtable model, with theextra fields that are not always in the other models:
idsecret_name
BecauseHero inherits formHeroBase, italso has thefields declared inHeroBase, so all the fields forHero are:
idnameagesecret_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
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}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):
idnameage
# 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
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}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:
nameagesecret_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
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}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:
nameagesecret_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
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}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
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
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
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}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
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}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
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}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
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}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. 🚀