Bases de Datos SQL (Relacionales)¶
🌐 Traducción por IA y humanos
Esta traducción fue hecha por IA guiada por humanos. 🤝
Podría tener errores al interpretar el significado original, o sonar poco natural, etc. 🤖
Puedes mejorar esta traducciónayudándonos a guiar mejor al LLM de IA.
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:
- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server, etc.
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
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}La claseHero es muy similar a un modelo de Pydantic (de hecho, en el fondo, realmentees un modelo de Pydantic).
Hay algunas diferencias:
table=Truele 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 queides laclave primaria en la base de datos SQL (puedes aprender más sobre claves primarias de SQL en la documentación de SQLModel).Nota: Usamos
int | Nonepara el campo de clave primaria para que en el código Python podamoscrear un objeto sin unid(id=None), asumiendo que la base de datos logenerará al guardar. SQLModel entiende que la base de datos proporcionará elidydefine la columna como unINTEGERno nulo en el esquema de la base de datos. Consulta ladocumentación de SQLModel sobre claves primarias para más detalles.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 como
strserá 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
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}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
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}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
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}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
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}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
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}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
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}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
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}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
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}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:
nameage
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}Hero - elmodelo de tabla¶
Luego, crearemosHero, elmodelo de tabla real, con loscampos extra que no siempre están en los otros modelos:
idsecret_name
Debido a queHero hereda deHeroBase,también tiene loscampos declarados enHeroBase, por lo que todos los campos paraHero son:
idnameagesecret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:str# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroPublic - 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):
idnameage
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:int# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroCreate - 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:
nameagesecret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:str# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroUpdate - 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:
nameagesecret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=None# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=Nonesqlite_file_name="database.db"sqlite_url=f"sqlite:///{sqlite_file_name}"connect_args={"check_same_thread":False}engine=create_engine(sqlite_url,connect_args=connect_args)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(status_code=404,detail="Hero not found")hero_data=hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}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
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}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
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}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
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}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
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}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
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}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 más largo sobre el uso de SQLModel conFastAPI. 🚀