Async SQL (Relational) Databases

    The current page still doesn’t have a translation for this language.

    But you can help translating it: Contributing.

    You can also use 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 .

    • Import SQLAlchemy.
    • Create a metadata object.
    • Create a table notes using the metadata object.

    Tip

    Notice that all this code is pure SQLAlchemy Core.

    databases is not doing anything here yet.

    Import and set up databases

    • Import databases.
    • Create a DATABASE_URL.
    • Create a database object.
    1. from typing import List
    3. import databases
    4. import sqlalchemy
    5. from fastapi import FastAPI
    6. from pydantic import BaseModel
    8. # SQLAlchemy specific code, as with any other app
    9. DATABASE_URL = "sqlite:///./test.db"
    10. # DATABASE_URL = "postgresql://user:password@postgresserver/db"
    12. database = databases.Database(DATABASE_URL)
    14. metadata = sqlalchemy.MetaData()
    16. notes = sqlalchemy.Table(
    17.     "notes",
    18.     metadata,
    19.     sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    20.     sqlalchemy.Column("text", sqlalchemy.String),
    21.     sqlalchemy.Column("completed", sqlalchemy.Boolean),
    22. )
    25. engine = sqlalchemy.create_engine(
    26.     DATABASE_URL, connect_args={"check_same_thread": False}
    27. )
    28. metadata.create_all(engine)
    31. class NoteIn(BaseModel):
    32.     text: str
    33.     completed: bool
    36. class Note(BaseModel):
    37.     id: int
    38.     text: str
    39.     completed: bool
    42. app = FastAPI()
    45. @app.on_event("startup")
    46. async def startup():
    47.     await database.connect()
    50. @app.on_event("shutdown")
    51. async def shutdown():
    52.     await database.disconnect()
    55. @app.get("/notes/", response_model=List[Note])
    56. async def read_notes():
    57.     query = notes.select()
    58.     return await database.fetch_all(query)
    61. @app.post("/notes/", response_model=Note)
    62. async def create_note(note: NoteIn):
    63.     query = notes.insert().values(text=note.text, completed=note.completed)
    64.     last_record_id = await database.execute(query)
    65.     return {**note.dict(), "id": last_record_id}

    If you were connecting to a different database (e.g. PostgreSQL), you would need to change the DATABASE_URL.

    Create the tables

    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.
    1. from typing import List
    3. import databases
    4. import sqlalchemy
    5. from fastapi import FastAPI
    6. from pydantic import BaseModel
    8. # SQLAlchemy specific code, as with any other app
    9. DATABASE_URL = "sqlite:///./test.db"
    10. # DATABASE_URL = "postgresql://user:password@postgresserver/db"
    13. metadata = sqlalchemy.MetaData()
    15. notes = sqlalchemy.Table(
    16.     "notes",
    17.     metadata,
    18.     sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    19.     sqlalchemy.Column("text", sqlalchemy.String),
    21. )
    24. engine = sqlalchemy.create_engine(
    25.     DATABASE_URL, connect_args={"check_same_thread": False}
    26. )
    27. metadata.create_all(engine)
    30. class NoteIn(BaseModel):
    31.     text: str
    32.     completed: bool
    35. class Note(BaseModel):
    36.     id: int
    37.     text: str
    38.     completed: bool
    41. app = FastAPI()
    44. @app.on_event("startup")
    45. async def startup():
    46.     await database.connect()
    49. @app.on_event("shutdown")
    50. async def shutdown():
    51.     await database.disconnect()
    54. @app.get("/notes/", response_model=List[Note])
    55. async def read_notes():
    56.     query = notes.select()
    57.     return await database.fetch_all(query)
    60. @app.post("/notes/", response_model=Note)
    61. async def create_note(note: NoteIn):
    62.     query = notes.insert().values(text=note.text, completed=note.completed)
    63.     last_record_id = await database.execute(query)
    64.     return {**note.dict(), "id": last_record_id}

    Create Pydantic models for:

    • Notes to be created (NoteIn).
    • Notes to be returned (Note).

    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.

    Connect and disconnect

    • Create your FastAPI application.
    • Create event handlers to connect and disconnect from the database.
    1. from typing import List
    3. import databases
    4. import sqlalchemy
    5. from fastapi import FastAPI
    6. from pydantic import BaseModel
    8. # SQLAlchemy specific code, as with any other app
    9. DATABASE_URL = "sqlite:///./test.db"
    10. # DATABASE_URL = "postgresql://user:password@postgresserver/db"
    12. database = databases.Database(DATABASE_URL)
    14. metadata = sqlalchemy.MetaData()
    16. notes = sqlalchemy.Table(
    17.     "notes",
    18.     metadata,
    19.     sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    20.     sqlalchemy.Column("text", sqlalchemy.String),
    21.     sqlalchemy.Column("completed", sqlalchemy.Boolean),
    22. )
    25. engine = sqlalchemy.create_engine(
    26.     DATABASE_URL, connect_args={"check_same_thread": False}
    27. )
    28. metadata.create_all(engine)
    31. class NoteIn(BaseModel):
    32.     text: str
    33.     completed: bool
    36. class Note(BaseModel):
    37.     id: int
    38.     text: str
    39.     completed: bool
    44. @app.on_event("startup")
    45. async def startup():
    46.     await database.connect()
    49. @app.on_event("shutdown")
    50. async def shutdown():
    51.     await database.disconnect()
    54. @app.get("/notes/", response_model=List[Note])
    55. async def read_notes():
    57.     return await database.fetch_all(query)
    60. @app.post("/notes/", response_model=Note)
    61. async def create_note(note: NoteIn):
    62.     query = notes.insert().values(text=note.text, completed=note.completed)
    63.     last_record_id = await database.execute(query)
    64.     return {**note.dict(), "id": last_record_id}

    Read notes

    Create the path operation function to read notes:

    1. from typing import List
    3. import databases
    4. import sqlalchemy
    5. from fastapi import FastAPI
    6. from pydantic import BaseModel
    8. # SQLAlchemy specific code, as with any other app
    9. DATABASE_URL = "sqlite:///./test.db"
    10. # DATABASE_URL = "postgresql://user:password@postgresserver/db"
    12. database = databases.Database(DATABASE_URL)
    14. metadata = sqlalchemy.MetaData()
    16. notes = sqlalchemy.Table(
    17.     "notes",
    18.     metadata,
    19.     sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    20.     sqlalchemy.Column("text", sqlalchemy.String),
    21.     sqlalchemy.Column("completed", sqlalchemy.Boolean),
    22. )
    25. engine = sqlalchemy.create_engine(
    26.     DATABASE_URL, connect_args={"check_same_thread": False}
    27. )
    28. metadata.create_all(engine)
    31. class NoteIn(BaseModel):
    32.     text: str
    33.     completed: bool
    36. class Note(BaseModel):
    37.     id: int
    38.     text: str
    39.     completed: bool
    42. app = FastAPI()
    45. @app.on_event("startup")
    46. async def startup():
    47.     await database.connect()
    50. @app.on_event("shutdown")
    51. async def shutdown():
    52.     await database.disconnect()
    55. @app.get("/notes/", response_model=List[Note])
    56. async def read_notes():
    57.     query = notes.select()
    58.     return await database.fetch_all(query)
    61. @app.post("/notes/", response_model=Note)
    62. async def create_note(note: NoteIn):
    63.     query = notes.insert().values(text=note.text, completed=note.completed)
    64.     last_record_id = await database.execute(query)
    65.     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.

    It uses typing.List.

    That documents (and validates, serializes, filters) the output data, as a list of Notes.

    Create the path operation function to create notes:

    Notice that as we communicate with the database using await, the path operation function is declared with async.

    About {**note.dict(), "id": last_record_id}

    note is a Pydantic Note object.

    note.dict() returns a dict with its data, something like:

    1. {
    2.     "text": "Some note",
    3.     "completed": False,
    4. }

    but it doesn’t have the id field.

    So we create a new dict, that contains the key-value pairs from note.dict() with:

    1. {**note.dict()}

    **note.dict() “unpacks” the key value pairs directly, so, {**note.dict()} would be, more or less, a copy of note.dict().

    And then, we extend that copy dict, adding another key-value pair: "id": last_record_id:

    So, the final result returned would be something like:

    1. {
    2.     "id": 1,
    3.     "text": "Some note",
    4.     "completed": False,

    Check it

    You can copy this code as is, and see the docs at http://127.0.0.1:8000/docs.

    There you can see all your API documented and interact with it:

    More info

    You can read more about encode/databases at its GitHub page.