FastAPI with SQLAlchemy, PostgreSQL, Alembic and Docker [Part-2] (asynchronous version)

Image for post
Image for post
An imitation of a database using boxes

Intro

Lets Start

pipenv install databases
pipenv install databases[postgresql]
pipenv install asyncpg
# Pull base image
FROM python:3.7

# Set environment varibles
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /code/

# Install dependencies
RUN pip install pipenv
COPY Pipfile Pipfile.lock /code/
RUN pipenv install --system --dev

COPY . /code/

EXPOSE 8000
version: "3"

services:
db:
image: postgres:11
ports:
- "5432:5432"
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- POSTGRES_DB=test_db
web:
build: .
command: bash -c "uvicorn main:app --host 0.0.0.0 --port 8000 --reload"
volumes:
- .:/code
ports:
- "8000:8000"
depends_on:
- db

pgadmin:
container_name: pgadmin
image: dpage/pgadmin4
environment:
- PGADMIN_DEFAULT_EMAIL=pgadmin4@pgadmin.org
- PGADMIN_DEFAULT_PASSWORD=admin
ports:
- "5050:80"
depends_on:
- db
from pydantic import BaseModel


class User(BaseModel):
first_name: str
last_name: str
age: int

class Config:
orm_mode = True
DATABASE_URL=postgresql://postgres:postgres@db:5432/postgres
import os
from databases import Database
from dotenv import load_dotenv
import sqlalchemy
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(BASE_DIR, ".env"))

db = Database(os.environ["DATABASE_URL"])
metadata = sqlalchemy.MetaData()
from db import db
from fastapi import FastAPI


app = FastAPI(title="Async FastAPI")


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


@app.on_event("shutdown")
async def shutdown():
await db.disconnect()
from db import db

users = sqlalchemy.Table(
"users",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("first_name", sqlalchemy.String),
sqlalchemy.Column("last_name", sqlalchemy.String),
sqlalchemy.Column("age", sqlalchemy.Integer),
)
import sqlalchemy
from db import db, metadata, sqlalchemy


users = sqlalchemy.Table(
"users",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("first_name", sqlalchemy.String),
sqlalchemy.Column("last_name", sqlalchemy.String),
sqlalchemy.Column("age", sqlalchemy.Integer),
)


class User:
@classmethod
async def get(cls, id):
query = users.select().where(users.c.id == id)
user = await db.fetch_one(query)
return user

@classmethod
async def create(cls, **user):
query = users.insert().values(**user)
user_id = await db.execute(query)
return user_id
import uvicorn
from models import User as ModelUser
from schema import User as SchemaUser
from app import app
from db import db


@app.post("/user/")
async def create_user(user: SchemaUser):
user_id = await ModelUser.create(**user.dict())
return {"user_id": user_id}


@app.get("/user/{id}", response_model=SchemaUser)
async def get_user(id: int):
user = await ModelUser.get(id)
return SchemaUser(**user).dict()


if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
import models

target_metadata = models.Base.metadata
import models
from db import metadata

target_metadata = metadata
docker-compose build
docker-compose run web alembic revision --autogenerate
docker-compose run web alembic upgrade head
docker-compose up
Image for post
Image for post
POST request to create a user
Image for post
Image for post
response of the previous request
Image for post
Image for post
GET request of the same user with response

Software Developer

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store