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

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
POST request to create a user
response of the previous request
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