Example combination of FastAPI and Pydantic with aiosql and aiosqlite

main.py

# -*- coding: utf-8 -*-
"""Example combination of FastAPI and Pydantic with aiosql and aiosqlite.

This module demonstrates the minimum viable integration approach for putting
together a few components:

- FastAPI. This provides a very high-performance and type-driving approach to
  building APIs in Python
- Pydantic. A powerful data validation library.
- aiosql. Inspired by annosql and originally Clojure's yeSql, a way to programatically,
  and safely, drive SQL databases with their own DSL: SQL.
- aiosqlite. An asyncio compatible version of the sqlite database adapter.

What's Shown:

1. How to load queries for aiosql to use.
2. Creating a dependency in FastAPI that can be used to inject the database connection 
   wherever you need it in an API endpoint.
3. Using differeny Pydantic models for the database and the API, and allowing FastAPI
   to automaticaly convert between them.

How to Use:

To use this, you will need to create a sqlite database called `testdb.sqlite`. You can
do this with the following script run in the same directory as this module:

  $ sqlite3 testdb.sqlite3 <<EOM 
  CREATE TABLE greetings (
    greeting_id INT PRIMARY KEY,
    greeting TEXT NOT NULL,
    note TEXT
  );
  INSERT INTO greetings (greeting, note) VALUES ("Test greeting", "Sample Note");
  EOM
  
Once that's created, you'll also need to install the correct dependencies. I use
Pipenv, but you can use anything that manages a virtualenv.  Just use a virtualenv.
You'll need to install the following packages:

  fastapi
  aiosql
  aiosqlite
  uvicorn

Once those are installed, you can run the script using something like this:

  $ uvicorn main:app

Enjoy. 
"""
from typing import List, Union

import aiosql
import aiosqlite
from fastapi import Depends, FastAPI
from pydantic import BaseModel

DB_FILE = "testdb.sqlite3"

SQL_QUERIES = """
-- name: get-all-greetings
-- Get all the greetings from the database
SELECT greeting_id, greeting, note FROM greetings;
"""

app = FastAPI()

queries = aiosql.from_str(SQL_QUERIES, "aiosqlite")


#############################################################################
# Pydantic Models
# This is used to provide strong typing on the data going into and out of the
# application. Note that normally, these would be stored in their own file,
# but to keep this simple, it's being kept in the same file.
class Greeting(BaseModel):
    """A greeting that can be used to the public.

    Attributes:
        greeting_id (int): The primary auto-generated key.
        greeting (str): The human-readable greeting.
    """

    greeting_id: int
    greeting: str


class GreetingDB(Greeting):
    """The database representation of the greeting.

    Attributes:
        note (str): An internal-only notation for the greeting.
    """

    note: Union[str, None]


#############################################################################
# Dependency
# Provide a re-usable way to inject a database connection into the individual
# endpoints.
async def get_db():
    """Return a database connection for use as a dependency.

    This connection has the Row row factory automatically attached."""

    db = await aiosqlite.connect(DB_FILE)
    # Provide a smarter version of the results. This keeps from having to unpack
    # tuples manually.
    db.row_factory = aiosqlite.Row

    try:
        yield db
    finally:
        await db.close()


#############################################################################
# Database Queries
# Leverage the `aiosql` library to query the database. This is a thinner
# approach than an ORM, and also allows you to be very specific about the
# query witho fewer surprises.
async def get_greetings(conn: aiosqlite.Connection) -> List[GreetingDB]:
    """Get all the greetings in the database."""

    greetings = await queries.get_all_greetings(conn)

    # parse_obj allows us to use the Row returned to populate the model by
    # name, rather than having to unpack the tuples.
    return [GreetingDB.parse_obj(greeting) for greeting in greetings]


#############################################################################
# FastAPI Endpoints
# Finally, we create a FastAPI endpoint to be called. Note the response model
# that is used is different from the one returned by the database call. FastAPI
# will automatically convert them.
@app.get("/", response_model=List[Greeting])
async def test_get(db_connection: aiosqlite.Connection = Depends(get_db)):
    greetings = await get_greetings(db_connection)

    return greetings

Nice implementation, exactly what I was looking for! Unfortunately, this does not work with pydantic v2.0 or greater (which also btw deprecated .parse_obj() for .model_validate()). Wonder if there is a fix for this.

The error message:

pydantic_core._pydantic_core.ValidationError: 1 validation error for GreetingDB
  Input should be a valid dictionary or instance of GreetingDB [type=model_type, input_value=<sqlite3.Row object at 0x7fc3a7a63310>, input_type=Row]
    For further information visit https://errors.pydantic.dev/2.3/v/model_type

Nice implementation, exactly what I was looking for! Unfortunately, this does not work with pydantic v2.0 or greater (which also btw deprecated .parse_obj() for .model_validate()). Wonder if there is a fix for this.

The error message:

pydantic_core._pydantic_core.ValidationError: 1 validation error for GreetingDB
  Input should be a valid dictionary or instance of GreetingDB [type=model_type, input_value=<sqlite3.Row object at 0x7fc3a7a63310>, input_type=Row]
    For further information visit https://errors.pydantic.dev/2.3/v/model_type

The call of aiosql ( your custom sql select statement) function returns the Record of the type that is provided by your database connector. In case asyncpg it is asyncpg.Record that can be converted to dict by dict(record.items()) method and after that can be converted to pydantic model by the call of the init method: GreetingDB(**dict(record.items())). In case of FastAPI, the result of aiosql."function" can be directly returned by endpoint and further converted and validated by the model specified in response_model param of path function of endpoint.

Right now you can just:

async def get_greetings(conn: aiosqlite.Connection) -> List[GreetingDB]:
    """Get all the greetings in the database."""

    greetings = await queries.get_all_greetings(conn)

    # parse_obj allows us to use the Row returned to populate the model by
    # name, rather than having to unpack the tuples.
    return [GreetingDB(**dict(greeting)) for greeting in greetings]