The new docs will include Pydantic v2 and will use SQLModel once it is updated to use Pydantic v2 as well.
You can also use encode/databases with FastAPI to connect to databases using async and await.
It is compatible with:
PostgreSQL
MySQL
SQLite
In this example, we'll use SQLite, because it uses a single file and Python has integrated support. So, you can copy this example and run it as is.
Later, for your production application, you might want to use a database server like PostgreSQL.
Tip
You could adopt ideas from the section about SQLAlchemy ORM (SQL (Relational) Databases), like using utility functions to perform operations in the database, independent of your FastAPI code.
This section doesn't apply those ideas, to be equivalent to the counterpart in Starlette.
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
Tip
Notice that all this code is pure SQLAlchemy Core.
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
Tip
If you were connecting to a different database (e.g. PostgreSQL), you would need to change the DATABASE_URL.
In this case, we are creating the tables in the same Python file, but in production, you would probably want to create them with Alembic, integrated with migrations, etc.
Here, this section would run directly, right before starting your FastAPI application.
Create an engine.
Create all the tables from the metadata object.
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
By creating these Pydantic models, the input data will be validated, serialized (converted), and annotated (documented).
So, you will be able to see it all in the interactive API docs.
Create event handlers to connect and disconnect from the database.
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
Note
Notice that as we communicate with the database using await, the path operation function is declared with async.
Create the path operation function to create notes:
fromtypingimportListimportdatabasesimportsqlalchemyfromfastapiimportFastAPIfrompydanticimportBaseModel# SQLAlchemy specific code, as with any other appDATABASE_URL="sqlite:///./test.db"# DATABASE_URL = "postgresql://user:password@postgresserver/db"database=databases.Database(DATABASE_URL)metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)engine=sqlalchemy.create_engine(DATABASE_URL,connect_args={"check_same_thread":False})metadata.create_all(engine)classNoteIn(BaseModel):text:strcompleted:boolclassNote(BaseModel):id:inttext:strcompleted:boolapp=FastAPI()@app.on_event("startup")asyncdefstartup():awaitdatabase.connect()@app.on_event("shutdown")asyncdefshutdown():awaitdatabase.disconnect()@app.get("/notes/",response_model=List[Note])asyncdefread_notes():query=notes.select()returnawaitdatabase.fetch_all(query)@app.post("/notes/",response_model=Note)asyncdefcreate_note(note:NoteIn):query=notes.insert().values(text=note.text,completed=note.completed)last_record_id=awaitdatabase.execute(query)return{**note.dict(),"id":last_record_id}
Note
Notice that as we communicate with the database using await, the path operation function is declared with async.