Movatterモバイル変換

[0]ホーム
URL:

Saltar a contenido
Follow@fastapi onTwitter to stay updated
Subscribe to theFastAPI and friends newsletter 🎉
You can now sponsorFastAPI 🍰
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor

Bases de Datos SQL (Relacionales)

FastAPI no requiere que uses una base de datos SQL (relacional). Pero puedes utilizarcualquier base de datos que desees.

Aquí veremos un ejemplo usandoSQLModel.

SQLModel está construido sobreSQLAlchemy y Pydantic. Fue creado por el mismo autor deFastAPI para ser la combinación perfecta para aplicaciones de FastAPI que necesiten usarbases de datos SQL.

Consejo

Puedes usar cualquier otro paquete de bases de datos SQL o NoSQL que quieras (en algunos casos llamadas"ORMs"), FastAPI no te obliga a usar nada. 😎

Como SQLModel se basa en SQLAlchemy, puedes usar fácilmentecualquier base de datos soportada por SQLAlchemy (lo que las hace también soportadas por SQLModel), como:

En este ejemplo, usaremosSQLite, porque utiliza un solo archivo y Python tiene soporte integrado. Así que puedes copiar este ejemplo y ejecutarlo tal cual.

Más adelante, para tu aplicación en producción, es posible que desees usar un servidor de base de datos comoPostgreSQL.

Consejo

Hay un generador de proyectos oficial conFastAPI yPostgreSQL que incluye un frontend y más herramientas:https://github.com/fastapi/full-stack-fastapi-template

Este es un tutorial muy simple y corto, si deseas aprender sobre bases de datos en general, sobre SQL o más funcionalidades avanzadas, ve a ladocumentación de SQLModel.

InstalarSQLModel

Primero, asegúrate de crear tuentorno virtual, actívalo, y luego instalasqlmodel:

$pipinstallsqlmodel---> 100%

Crear la App con un Solo Modelo

Primero crearemos la versión más simple de la aplicación con un solo modelo deSQLModel.

Más adelante la mejoraremos aumentando la seguridad y versatilidad conmúltiples modelos a continuación. 🤓

Crear Modelos

ImportaSQLModel y crea un modelo de base de datos:

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}

La claseHero es muy similar a un modelo de Pydantic (de hecho, en el fondo, realmentees un modelo de Pydantic).

Hay algunas diferencias:

  • table=True le dice a SQLModel que este es unmodelo de tabla, que debe representar unatabla en la base de datos SQL, no es solo unmodelo de datos (como lo sería cualquier otra clase regular de Pydantic).

  • Field(primary_key=True) le dice a SQLModel queid es laclave primaria en la base de datos SQL (puedes aprender más sobre claves primarias de SQL en la documentación de SQLModel).

    Al tener el tipo comoint | None, SQLModel sabrá que esta columna debe ser unINTEGER en la base de datos SQL y que debe serNULLABLE.

  • Field(index=True) le dice a SQLModel que debe crear uníndice SQL para esta columna, lo que permitirá búsquedas más rápidas en la base de datos cuando se lean datos filtrados por esta columna.

    SQLModel sabrá que algo declarado comostr será una columna SQL de tipoTEXT (oVARCHAR, dependiendo de la base de datos).

Crear un Engine

Unengine de SQLModel (en el fondo, realmente es unengine de SQLAlchemy) es lo quemantiene las conexiones a la base de datos.

Tendríasun solo objetoengine para todo tu código para conectar a la misma base de datos.

# 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}

Usarcheck_same_thread=False permite a FastAPI usar la misma base de datos SQLite en diferentes hilos. Esto es necesario ya queuna sola request podría usarmás de un hilo (por ejemplo, en dependencias).

No te preocupes, con la forma en que está estructurado el código, nos aseguraremos de usaruna solasession de SQLModel por request más adelante, esto es realmente lo que intenta lograr elcheck_same_thread.

Crear las Tablas

Luego añadimos una función que usaSQLModel.metadata.create_all(engine) paracrear las tablas para todos losmodelos de tabla.

# 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}

Crear una Dependencia de Session

UnaSession es lo que almacena losobjetos en memoria y lleva un seguimiento de cualquier cambio necesario en los datos, luegousa elengine para comunicarse con la base de datos.

Crearemos unadependencia de FastAPI conyield que proporcionará una nuevaSession para cada request. Esto es lo que asegura que usemos una sola session por request. 🤓

Luego creamos una dependenciaAnnotatedSessionDep para simplificar el resto del código que usará esta dependencia.

# 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}

Crear Tablas de Base de Datos al Arrancar

Crearemos las tablas de la base de datos cuando arranque la aplicación.

# 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}

Aquí creamos las tablas en un evento de inicio de la aplicación.

Para producción probablemente usarías un script de migración que se ejecuta antes de iniciar tu aplicación. 🤓

Consejo

SQLModel tendrá utilidades de migración envolviendo Alembic, pero por ahora, puedes usarAlembic directamente.

Crear un Hero

Debido a que cada modelo de SQLModel también es un modelo de Pydantic, puedes usarlo en las mismasanotaciones de tipos que podrías usar en modelos de Pydantic.

Por ejemplo, si declaras un parámetro de tipoHero, será leído desde elJSON body.

De la misma manera, puedes declararlo como eltipo de retorno de la función, y luego la forma de los datos aparecerá en la interfaz automática de documentación de la API.

# 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}

Aquí usamos la dependenciaSessionDep (unaSession) para añadir el nuevoHero a la instanceSession, comiteamos los cambios a la base de datos, refrescamos los datos en elhero y luego lo devolvemos.

Leer Heroes

PodemosleerHeros de la base de datos usando unselect(). Podemos incluir unlimit yoffset para paginar los resultados.

# 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}

Leer Un Hero

Podemosleer un únicoHero.

# 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}

Eliminar un Hero

También podemoseliminar unHero.

# 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}

Ejecutar la App

Puedes ejecutar la aplicación:

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

Luego dirígete a la interfaz de/docs, verás queFastAPI está usando estosmodelos paradocumentar la API, y los usará paraserializar yvalidar los datos también.

Actualizar la App con Múltiples Modelos

Ahora vamos arefactorizar un poco esta aplicación para aumentar laseguridad y laversatilidad.

Si revisas la aplicación anterior, en la interfaz verás que, hasta ahora, permite al cliente decidir elid delHero a crear. 😱

No deberíamos permitir que eso suceda, podrían sobrescribir unid que ya tenemos asignado en la base de datos. Decidir elid debería ser tarea delbackend o labase de datos,no del cliente.

Además, creamos unsecret_name para el héroe, pero hasta ahora, lo estamos devolviendo en todas partes, eso no es muysecreto... 😅

Arreglaremos estas cosas añadiendo unosmodelos extra. Aquí es donde SQLModel brillará. ✨

Crear Múltiples Modelos

EnSQLModel, cualquier clase de modelo que tengatable=True es unmodelo de tabla.

Y cualquier clase de modelo que no tengatable=True es unmodelo de datos, estos son en realidad solo modelos de Pydantic (con un par de características extra pequeñas). 🤓

Con SQLModel, podemos usarherencia paraevitar duplicar todos los campos en todos los casos.

HeroBase - la clase base

Comencemos con un modeloHeroBase que tiene todos loscampos que son compartidos por todos los modelos:

  • 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 - elmodelo de tabla

Luego, crearemosHero, elmodelo de tabla real, con loscampos extra que no siempre están en los otros modelos:

  • id
  • secret_name

Debido a queHero hereda deHeroBase,también tiene loscampos declarados enHeroBase, por lo que todos los campos paraHero son:

  • 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 - elmodelo de datos público

A continuación, creamos un modeloHeroPublic, este es el que serádevuelto a los clientes de la API.

Tiene los mismos campos queHeroBase, por lo que no incluirásecret_name.

Por fin, la identidad de nuestros héroes está protegida! 🥷

También vuelve a declararid: int. Al hacer esto, estamos haciendo uncontrato con los clientes de la API, para que siempre puedan esperar que elid esté allí y sea unint (nunca seráNone).

Consejo

Tener el modelo de retorno asegurando que un valor siempre esté disponible y siempre seaint (noNone) es muy útil para los clientes de la API, pueden escribir código mucho más simple teniendo esta certeza.

Además, losclientes generados automáticamente tendrán interfaces más simples, para que los desarrolladores que se comuniquen con tu API puedan tener una experiencia mucho mejor trabajando con tu API. 😎

Todos los campos enHeroPublic son los mismos que enHeroBase, conid declarado comoint (noNone):

  • 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: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 - elmodelo de datos para crear un héroe

Ahora creamos un modeloHeroCreate, este es el quevalidará los datos de los clientes.

Tiene los mismos campos queHeroBase, y también tienesecret_name.

Ahora, cuando los clientescrean un nuevo héroe, enviarán elsecret_name, se almacenará en la base de datos, pero esos nombres secretos no se devolverán en la API a los clientes.

Consejo

Esta es la forma en la que manejaríascontraseñas. Recíbelas, pero no las devuelvas en la API.

Tambiénhashea los valores de las contraseñas antes de almacenarlos,nunca los almacenes en texto plano.

Los campos deHeroCreate son:

  • 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 - elmodelo de datos para actualizar un héroe

No teníamos una forma deactualizar un héroe en la versión anterior de la aplicación, pero ahora conmúltiples modelos, podemos hacerlo. 🎉

Elmodelo de datosHeroUpdate es algo especial, tienetodos los mismos campos que serían necesarios para crear un nuevo héroe, pero todos los campos sonopcionales (todos tienen un valor por defecto). De esta forma, cuando actualices un héroe, puedes enviar solo los campos que deseas actualizar.

Debido a que todos loscampos realmente cambian (el tipo ahora incluyeNone y ahora tienen un valor por defecto deNone), necesitamosvolver a declararlos.

Realmente no necesitamos heredar deHeroBase porque estamos volviendo a declarar todos los campos. Lo dejaré heredando solo por consistencia, pero esto no es necesario. Es más una cuestión de gusto personal. 🤷

Los campos deHeroUpdate son:

  • 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}

Crear conHeroCreate y devolver unHeroPublic

Ahora que tenemosmúltiples modelos, podemos actualizar las partes de la aplicación que los usan.

Recibimos en la request unmodelo de datosHeroCreate, y a partir de él, creamos unmodelo de tablaHero.

Este nuevomodelo de tablaHero tendrá los campos enviados por el cliente, y también tendrá unid generado por la base de datos.

Luego devolvemos el mismomodelo de tablaHero tal cual desde la función. Pero como declaramos elresponse_model con elmodelo de datosHeroPublic,FastAPI usaráHeroPublic para validar y serializar los datos.

# 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}

Consejo

Ahora usamosresponse_model=HeroPublic en lugar de laanotación de tipo de retorno-> HeroPublic porque el valor que estamos devolviendo en realidadno es unHeroPublic.

Si hubiéramos declarado-> HeroPublic, tu editor y linter se quejarían (con razón) de que estás devolviendo unHero en lugar de unHeroPublic.

Al declararlo enresponse_model le estamos diciendo aFastAPI que haga lo suyo, sin interferir con las anotaciones de tipo y la ayuda de tu editor y otras herramientas.

Leer Heroes conHeroPublic

Podemos hacer lo mismo que antes paraleerHeros, nuevamente, usamosresponse_model=list[HeroPublic] para asegurar que los datos se validen y serialicen correctamente.

# 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}

Leer Un Hero conHeroPublic

Podemosleer un único héroe:

# 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}

Actualizar un Hero conHeroUpdate

Podemosactualizar un héroe. Para esto usamos una operación HTTPPATCH.

Y en el código, obtenemos undict con todos los datos enviados por el cliente,solo los datos enviados por el cliente, excluyendo cualquier valor que estaría allí solo por ser valores por defecto. Para hacerlo usamosexclude_unset=True. Este es el truco principal. 🪄

Luego usamoshero_db.sqlmodel_update(hero_data) para actualizar elhero_db con los datos dehero_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}

Eliminar un Hero de Nuevo

Eliminar un héroe se mantiene prácticamente igual.

No satisfaremos el deseo de refactorizar todo en este punto. 😅

# 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}

Ejecutar la App de Nuevo

Puedes ejecutar la aplicación de nuevo:

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

Si vas a la interfaz de/docs de la API, verás que ahora está actualizada, y no esperará recibir elid del cliente al crear un héroe, etc.

Resumen

Puedes usarSQLModel para interactuar con una base de datos SQL y simplificar el código conmodelos de datos ymodelos de tablas.

Puedes aprender mucho más en la documentación deSQLModel, hay un minitutorial sobre el uso de SQLModel conFastAPI. 🚀

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