Developing a Restful API with FastAPI using PostgreSQL, Async SQLAlchemy, and AIOHTTP

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.6+ based on standard Python type hints. In this tutorial, we will create a REST API using FastAPI with PostgreSQL as our database, Async SQLAlchemy for database operations, and AIOHTTP for asynchronous HTTP requests.

Step 1: Setting up the environment

First, create a new directory for your project and set up a virtual environment:

mkdir fastapi-tutorial
cd fastapi-tutorial
python3 -m venv env
source env/bin/activate

Next, install the necessary packages:

pip install fastapi[all]
pip install psycopg2-binary sqlalchemy databases aiohttp

Step 2: Set up the database

Create a PostgreSQL database and a table to store some data. You can do this using the psql command-line tool or a graphical client like pgAdmin.

psql postgres
CREATE DATABASE fastapi_tutorial;
connect fastapi_tutorial
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100)
);

Step 3: Create a FastAPI app

Create a new Python file main.py and add the following code to set up a FastAPI app with Async SQLAlchemy and AIOHTTP:

from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table
from databases import Database

app = FastAPI()

DATABASE_URL = "postgresql:///fastapi_tutorial"
database = Database(DATABASE_URL)

metadata = MetaData()

users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(100)),
    Column("email", String(100)),
)

engine = create_engine(DATABASE_URL)
metadata.create_all(engine)

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

Step 4: Create API endpoints

Now let’s create some API endpoints to perform CRUD operations on the users table. Add the following code to main.py:

from typing import List

@app.get("/users/", response_model=List[dict])
async def read_users():
    query = users.select()
    return await database.fetch_all(query)

@app.post("/users/", response_model=dict)
async def create_user(name: str, email: str):
    query = users.insert().values(name=name, email=email)
    return await database.execute(query)

@app.get("/users/{user_id}", response_model=dict)
async def read_user(user_id: int):
    query = users.select().where(users.c.id == user_id)
    result = await database.fetch_one(query)
    if result is None:
        raise HTTPException(status_code=404, detail="User not found")
    return result

@app.put("/users/{user_id}", response_model=dict)
async def update_user(user_id: int, name: str, email: str):
    query = users.update().where(users.c.id == user_id).values(name=name, email=email)
    return await database.execute(query)

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    query = users.delete().where(users.c.id == user_id)
    return await database.execute(query)

Step 5: Run the FastAPI app

Now you can run the FastAPI app by executing the following command in your terminal:

uvicorn main:app --reload

You can now access your API endpoints at http://localhost:8000/users/ and perform CRUD operations on the users table in your PostgreSQL database.

Congratulations! You have successfully created a REST API using FastAPI with PostgreSQL, Async SQLAlchemy, and AIOHTTP. Feel free to extend this tutorial by adding more endpoints, implementing authentication, or adding more complex business logic. FastAPI’s high performance and automatic data validation make it a powerful tool for building web APIs with Python.