FastAPI CRUD API using Postgresql and Docker-compose

Part 1. Setup rest API calls.

For making requests to a random user generator API we will use requests library. To install the library run pip install requests in your terminal. To specify our request we will use parameters:

1. Number of results
2. Sex of users

Proper request URL is https://randomuser.me/api?results=100&gender=male

First, we need to create file helpers.py where we will write all our helpers.

This URL will return JSON result in this format

Now let’s write a function to request this result. As parameters we will use integer value number of results and boolean value of is sex is male.

def get_users_from_api(results: int, gender: str):
 response = requests.get(f’https://randomuser.me/api?results={results}&gender={gender}')
 if response.status_code == 200:
 return response.json()

Part 2. Setup database

As a database we will use PostgreSQL with SQAlchemy ORM, to simplify our requests to the database. First of all, we need to install SQAlchemy via pip. Use this library guide: https://pypi.org/project/SQLAlchemy/. Or to install the library run pip install sqlachemy psycopg2-binary in your terminal

After this, we need to import and declare all the stuff that we need to do to make our database work.

from sqlalchemy import Column, Integer, MetaData, String, ForeignKey, Boolean, DateTime, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship
from settings import PostgresConfiguration
from sqlalchemy.dialects.postgresql import UUID
from uuid import uuid4
pg = PostgresConfiguration()
engine = create_engine(pg.postgres_db_path)
meta = MetaData(engine)
Base = declarative_base()

Let’s convert our user data from JSON response to a database model, for this we are splitting user data into 3 tables and setting up relationships between parents and children.

class UserTable(Base):
 __tablename__ = ‘users’
 uuid = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
 cell = Column(String)
 dob_date = Column(DateTime)
 email = Column(String)
 gender_m = Column(Boolean)
 id_name = Column(String)
 id_value = Column(String)
 locations = relationship(“LocationTable”, uselist=False, back_populates=”users”)
 logins = relationship(“LoginTable”, uselist=False, back_populates=”users”)
 name_first = Column(String)
 name_last = Column(String)
 name_title = Column(String)
 nat = Column(String)
 phone = Column(String)
 picture_id = Column(Integer)
 registered_date = Column(DateTime)

class LocationTable(Base):
 __tablename__ = ‘locations’
 location_id = Column(Integer, primary_key=True, autoincrement=True)
 user_uuid = Column(UUID(as_uuid=True), ForeignKey(‘users.uuid’, ondelete=’CASCADE’))
 users = relationship(“UserTable”, back_populates=”locations”)
 city = Column(String)
 latitude = Column(Float)
 longitude = Column(Float)
 country = Column(String)
 postcode = Column(Integer)
 state = Column(String)
 street_name = Column(String)
 street_number = Column(Integer)
 timezone_description = Column(String)
 timezone_offset = Column(String)

class LoginTable(Base):
 __tablename__ = ‘logins’
 login_id = Column(Integer, primary_key=True, autoincrement=True)
 user_uuid = Column(UUID(as_uuid=True), ForeignKey(‘users.uuid’, ondelete=’CASCADE’))
 users = relationship(“UserTable”, back_populates=”logins”)
 md5 = Column(String)
 password = Column(String)
 salt = Column(String)
 sha1 = Column(String)
 sha256 = Column(String)
 username = Column(String)

Also, we need to create a function that creates all tables on the app deploy.

def create_tables():
 Base.metadata.create_all(engine, checkfirst=True)

Now let’s create a class where we will set up all connection data and write crud helper methods. First of all, let’s create the get_user_by_uuid() method with a UUID parameter because we will more often check if a user with this UUID exists in the database than other operations. It will return an SQLAlchemy user object.

def get_user_by_uuid(self, uuid: str):
 user = self.session.query(UserTable, LoginTable, LocationTable).join(LoginTable).join(LocationTable).filter(
 UserTable.uuid == uuid).all()
 if user:
 return user

After this let’s create all other DB methods.

Add the user with a dictionary parameter.

def add_user(self, data: dict):
 user = self.get_user_by_uuid(data[‘uuid’])
 if user:
 return user
 user = UserTable(**data)
 self.session.add(user)
 try:
 self.session.commit()
 except InvalidRequestError:
 self.session.rollback()
 raise InvalidRequestError
 return user

Get_users method without any parameters - to retrieve all users

def get_users(self):
 users = self.session.query(UserTable, LoginTable, LocationTable).join(LoginTable).join(LocationTable).all()
 if users:
 return users

And delete_user with user_uuid parameter - to delete user by id

def delete_user(self, uuid: int):
 self.session.query(UserTable).filter(UserTable.uuid == uuid).delete()
 try:
 self.session.commit()
 return True
 except InvalidRequestError:
 self.session.rollback()
 raise InvalidRequestError

Also, we need to add some methods for location table:

def add_location(self, data: dict):
 location_table = LocationTable(**data)
 self.session.add(location_table)
 try:
 self.session.commit()
 except InvalidRequestError:
 self.session.rollback()
 raise InvalidRequestError
 return location_table
def get_location_by_id(self, location_id: int):
 location = self.session.query(LocationTable).filter_by(location_id=location_id).scalar()
 if location:
 return location

And for the login table:

def add_login(self, data: dict):
 login_table = LoginTable(**data)
 self.session.add(login_table)
 try:
 self.session.commit()
 except InvalidRequestError:
 self.session.rollback()
 raise InvalidRequestError
 return login_table
def get_login_by_id(self, login_id: int):
 login = self.session.query(LoginTable).filter_by(login_id=login_id).scalar()
 if login:
 return login

Finally, we need to initialize an instance of our class with data from our settings

pg_handler = PosgresHandler(PostgresConfiguration().postgres_db_path)

To use database functions we also need to write some database helpers. Let’s start with converting API response from JSON to DB model and conversion database response to JSON.

def add_user_to_db(user_data: dict):
 user = pg_handler.add_user(dict(uuid=user_data[‘login’][‘uuid’],
 cell=user_data[‘cell’],
 dob_date=user_data[‘dob’][‘date’],
 email=user_data[‘email’],
 gender_m=True if user_data[‘gender’] == ‘male’ else False,
 id_name=user_data[‘id’][‘name’],
 id_value=user_data[‘id’][‘value’],
 name_first=user_data[‘name’][‘first’],
 name_last=user_data[‘name’][‘last’],
 name_title=user_data[‘name’][‘title’],
 nat=user_data[‘nat’],
 phone=user_data[‘phone’],
 picture_id=user_data[‘picture’][‘large’].split(‘/’)[-1].replace(‘.jpg’, ‘’),
 registered_date=user_data[‘registered’][‘date’]
 ))
 pg_handler.add_login(dict(md5=user_data[‘login’][‘md5’],
 user_uuid=user_data[‘login’][‘uuid’],
 password=user_data[‘login’][‘password’],
 salt=user_data[‘login’][‘salt’],
 sha1=user_data[‘login’][‘sha1’],
 sha256=user_data[‘login’][‘sha256’],
 username=user_data[‘login’][‘username’])
 )
 pg_handler.add_location(dict(city=user_data[‘location’][‘city’],
 user_uuid=user_data[‘login’][‘uuid’],
 latitude=user_data[‘location’][‘coordinates’][‘latitude’],
 longitude=user_data[‘location’][‘coordinates’][‘longitude’],
 country=user_data[‘location’][‘country’],
 state=user_data[‘location’][‘state’],
 street_name=user_data[‘location’][‘street’][‘name’],
 street_number=user_data[‘location’][‘street’][‘number’],
 timezone_description=user_data[‘location’][‘timezone’][‘description’],
 timezone_offset=user_data[‘location’][‘timezone’][‘offset’]
 ))
 return user.uuid

The next big function will be the normalize_result(user_data) function that converts an object from a database into a JSON object. Also, we don’t store in the database some data that can be calculated and depends on the current date. That is the date of birth age and registered age. For these fields, we need to use these calculations

round((datetime.now() — user_data.UserTable.registered_date).days / 365)

Here is all code of the function:

def normalize_result(user_data):
 return {‘cell’: user_data.UserTable.cell,
 ‘dob’: {‘age’: round((datetime.now() — user_data.UserTable.dob_date).days / 365),
 ‘date’: user_data.UserTable.dob_date},
 ‘email’: user_data.UserTable.email,
 ‘gender’: ‘male’ if user_data.UserTable.gender_m else ‘female’,
 ‘id’: {‘name’: user_data.UserTable.id_name, ‘value’: user_data.UserTable.id_value},
 ‘location’: {‘city’: user_data.LocationTable.city,
 ‘coordinates’: {‘latitude’: user_data.LocationTable.latitude,
 ‘longitude’: user_data.LocationTable.longitude},
 ‘country’: user_data.LocationTable.country,
 ‘postcode’: user_data.LocationTable.postcode,
 ‘state’: user_data.LocationTable.state,
 ‘street’: {‘name’: user_data.LocationTable.street_name,
 ‘number’: user_data.LocationTable.street_number},
 ‘timezone’: {‘description’: user_data.LocationTable.timezone_description,
 ‘offset’: user_data.LocationTable.timezone_offset}},
 ‘login’: {‘md5’: user_data.LoginTable.md5,
 ‘password’: user_data.LoginTable.password,
 ‘salt’: user_data.LoginTable.salt,
 ‘sha1’: user_data.LoginTable.sha1,
 ‘sha256’: user_data.LoginTable.sha256,
 ‘username’: user_data.LoginTable.username,
 ‘uuid’: user_data.UserTable.uuid},
 ‘name’: {‘first’: user_data.UserTable.name_first,
 ‘last’: user_data.UserTable.name_last,
 ‘title’: user_data.UserTable.name_title},
 ‘nat’: user_data.UserTable.nat,
 ‘phone’: user_data.UserTable.phone,
 ‘picture’: {‘large’: f’https://randomuser.me/api/portraits/women/{user_data.UserTable.picture_id}.jpg’,
 ‘medium’: f’https://randomuser.me/api/portraits/med/women/{user_data.UserTable.picture_id}.jpg’,
 ‘thumbnail’: f’https://randomuser.me/api/portraits/thumb/women/{user_data.UserTable.picture_id}.jpg’},
 ‘registered’: {‘age’: round((datetime.now() — user_data.UserTable.registered_date).days / 365),
 ‘date’: {user_data.UserTable.registered_date}}}

Now let’s finish to write helpers with functions

def get_users_from_db():
 normalized_users = list()
 users = pg_handler.get_users()
 for user in users:
 normalized_users.append(normalize_result(user))
 return users
def get_user_by_uuid(uuid: str):
 user_data = pg_handler.get_user_by_uuid(uuid)
 if user_data:
 return normalize_result(user_data[0])

def delete_user_from_db(user_uuid):
 return pg_handler.delete_user(user_uuid)

Part 3. Setup Fastapi

To use our app as a REST API we will use Flask like FastAPI library that provides everything that we will need.

First of all, we need to install and set up our API. To install can use a guide from a library home page https://fastapi.tiangolo.com/tutorial/first-steps/ or just by running a command

pip install fastapi unicorn in your terminal

After this we need to import all functions and libraries that we need:

import uvicorn
from fastapi import FastAPI
from handlers import add_user_to_db, get_users_from_api, get_user_by_uuid, get_users_from_db, delete_user_from_db
from models import create_tables
from settings import APP_PORT
app = FastAPI()

Now we will write our first endpoint what will populate our users

@app.post(‘/users/’, summary=’populate users’, tags=[‘Users’])
async def populate_users(quantity: int, male: bool):
 users = get_users_from_api(quantity, ‘male’ if male else ‘female’)
 user_ids = []
 for user in users[‘results’]:
 user_ids.append(add_user_to_db(user))
 return dict(added_user_ids=user_ids)

This function will get our quantity and sex parameters and return UUIDs of users added to the database, on URL /users.

Also, FastAPI provides Swagger UI to test our API. Our endpoint will look like:

Now let's create endpoints for other CRUD functions

Get all users:

@app.get(‘/users/’, summary=’get all users’, tags=[‘Users’])
async def get_users():
 users_data = get_users_from_db()
 if not users_data:
 users_data = ‘No users in database’
 return dict(users_data=users_data)

Get user by UUID:

@app.get(‘/users/{user_uuid}’, summary=’get user by user_uuid’, tags=[‘Users’])
async def get_user(user_uuid: str):
 user_data = get_user_by_uuid(user_uuid)
 if not user_data:
 user_data = ‘Could not find user in database’
 return dict(user_data=user_data)

Delete a user by UUID:

@app.delete(‘/users/{user_uuid}’, summary=’delete user by user_uuid’, tags=[‘Users’])
async def delete_user(user_uuid: str):
 result = delete_user_from_db(user_uuid)
 if result:
 return dict(message=f’User {user_uuid} deleted successfully’)
 return dict(message=f’User not presented in database’)

Part 4. Setup docker

Now let’s create a new docker file with this data:

FROM tiangolo/uvicorn-gunicorn-fastapi:python3.7
COPY . .
RUN pip install -r requirements.txt
EXPOSE 8021
CMD python app.py

After this we are creating the docker-compose file that will run our docker images:

version: “3.5”
services:
 user-database:
 restart: always
 container_name: ‘user_stat’
 image: postgres
 ports:
 — ‘54320:5432’
 env_file:
 — .env
 networks:
 — user_stat
 user-stat:
 build: .
 depends_on:
 — user-database
 restart: always
 ports:
 — ‘8023:8023’
 tty: true
 volumes:
 — .:/app
 stdin_open: true
 networks:
 — user_stat
networks:
 user_stat:
 name: user_stat

Part 5. Setup, requirements.txt and .env files

To make a nice setup for our application we need to install a python-dotenv library. To install to your local environment run pip install python-dotenv

import logging
import os
from dotenv import load_dotenv
log = logging.getLogger()
log.setLevel(logging.DEBUG)
env = load_dotenv()
APP_PORT = os.getenv(‘APP_PORT’)

class PostgresConfiguration:
 POSTGRES_DB_PORT = os.getenv(‘POSTGRES_PORT’)
 POSTGRES_DB_NAME = os.getenv(‘POSTGRES_DB’)
 POSTGRES_DB_LOGIN = os.getenv(‘POSTGRES_USER’)
 POSTGRES_DB_PASSWORD = os.getenv(‘POSTGRES_PASSWORD’)
 POSTGRES_DB_ADDRESS = os.getenv(‘POSTGRES_ADDRESS’)
 @property
 def postgres_db_path(self):
 return f’postgres://{self.POSTGRES_DB_LOGIN}:{self.POSTGRES_DB_PASSWORD}@’ \
 f’{self.POSTGRES_DB_ADDRESS}:’ \
 f’{self.POSTGRES_DB_PORT}/{self.POSTGRES_DB_NAME}’

Add .env file

Here we need to create .env file with our app data

certifi==2020.4.5.1
chardet==3.0.4
idna==2.9
requests==2.23.0
urllib3==1.25.8
click==7.1.1
fastapi==0.54.0
h11==0.9.0
httptools==0.1.1
psycopg2-binary==2.8.4
pydantic==1.4
python-dotenv==0.12.0
SQLAlchemy==1.3.13
starlette==0.13.2
uvicorn==0.11.3
uvloop==0.14.0
websockets==8.1

Finish project

Now you can deploy your API by running docker-compose up in your terminal and your app will be served at http://0.0.0.0:8023 and swagger documentation will be available at http://0.0.0.0:8023/docs

All code from this article is in this repository https://github.com/mrstask/fastapi-crud