Fast API as EndPoint of Postgres

Introduction

Recently, I've watch this video which a youtuber try to solve an interview step by step. But in this case, he didn't share here code.

In my case, I love to try to solve this kind of issues but I've never work with a backend solution as API. For this reason, I found intereseting tryng to repoduce his code (with some modification) to try to achive to get an API to get information about one table.

Here is my repository with the main code

Goal

Try to have a backend-structure (with FastAPI to trying get the information about one table from the database (in this case I'm going to use Postgrest).

Source data

For this purpouse it's going to be use this dataset. In there we could find two csv:

• 1st File: Travel Company Old Clients; Number of observations: 682
• 2nd File: Travel Company New Clients; Number of observations: 1303

We are going to use only Travel Company New Clients.

Instalation

Python Installation

All the libraries in the requirements.txt:

anyio==3.7.0
click==8.1.3
exceptiongroup==1.1.1
fastapi==0.97.0
greenlet==2.0.2
h11==0.14.0
httptools==0.5.0
idna==3.4
Jinja2==3.1.2
MarkupSafe==2.0.1
psycopg2==2.9.6
pydantic==1.10.9
python-dotenv==1.0.0
PyYAML==6.0
sniffio==1.3.0
SQLAlchemy==2.0.16
starlette==0.27.0
typing_extensions==4.6.3
uvicorn==0.22.0
uvloop==0.17.0
watchfiles==0.19.0
websockets==11.0.3

Create a virtual environment

We are going to use a virtual environment to try to keep the track of all the libraries we are going to need:

python -m venv venv

And also we are going to active the environment (Linux):

source venv/bin/activate

Finally to get the requirements.txt you could use this shell:

pip freeze >requirements.txt

Install FastAPI

We are going to need to install this library to deploy our python code as a server. They told us in there documentation how to install:

This library is to deploy our python code:

pip install fastapi

... and also we are going to need our system work as server for this reason they told us to use uvicorn.

pip install "uvicorn[standard]"

Example FastAPI

To try to check is it is working I've done this example, to try if everthing is working fine:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

This script return only a message of hello world. I've save this script as example_1.py. To run in our bash:

uvicorn example_1:app --reload

If it is working properly in bash you are going to have a message like this:

Then, you could go to your favorite browser and in the ip http://127.0.0.1:8000/ you are going to se a message like this:

Also there are a links I could find usefull like:
http://127.0.0.1:8000/docs - where you could see the documentation about your project an test the dependecies

Finally to close your bash server simple Ctrl-C and your server is going to be down.

Install Jinja2

FastAPI could be use with Jinja2 template and we are going to use it to get a template of a query

'''shell
pip install Jinja2
'''

Install SQLAlchemy

We are going to use it to connect to our database and make a querys:

pip install SQLAlchemy

Install Pydantic

To help us to translate of our results in the query to format json

pip install pydantic

Install markupsafe

Library to implements a text object that escapes characters so it is safe to use in HTML and XML

pip install MarkupSafe==2.0.1

Install psycopg2

Adaptar of PostgresSQL to python

pip install psycopg2

Postgres Installation w/docker

Env variables

We are going to use this env variables:

PS_VERSION=15
PS_USER=ps
PS_PASSWORD=ps_2023
PS_DB=RAW

Dockerfile

We are going to create a file with name Dockerfile.postgres like this:

FROM postgres

# make directory to Docker
RUN mkdir -p /home/src

# Copy source data 
COPY ./src /home/src
COPY ./db_script/* /docker-entrypoint-initdb.d

# How to create a main table 
RUN chmod a+r /docker-entrypoint-initdb.d/*

# Expose the port
EXPOSE 6666

Steps to create FastAPI:

Models.py

First script we are going to create is how our model is going to be connected to our database based on the table of Newclients:

import os
from sqlalchemy import Column, String, Integer
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base

# Use environment variable DB_CONNSTR
if os.environ["DB_CONNSTR"] is None:
    DB_CONNSTR='postgresql://ps:ps_2023@localhost:5432/RAW'
else:
    DB_CONNSTR = os.environ["DB_CONNSTR"]

# Create engine to connect our data base
engine = create_engine(DB_CONNSTR) if DB_CONNSTR else None
# Get all the metadata
meta = MetaData(DB_CONNSTR)
# declare metada has an object
Base = declarative_base(metadata=meta)

# Declare tables
TABLE_NAME='newclients'
class Newclients(Base):
    __tablename__ = TABLE_NAME
    num_age = Column(Integer, primary_key=True)
    des_employetype = Column(String, primary_key=True)
    is_graduate = Column(String, primary_key=True)
    imp_annualincome = Column(Integer, primary_key=True)
    num_familymembers = Column(Integer, primary_key=True)
    is_frequentflyer = Column(String, primary_key=True)
    is_evertravelledabroad= Column(String, primary_key=True)

Backend.py

In this script we are going to create a class, with the base of the paramenters:

• select
• where
• order_by
• group_by

This script is going to connect to our database and return a dictionary to get the result of our query.

To achive this point we are going to use Jinja2Templates.

Jinja2Templates

FastApi it could works native with Jinja2Templates. With this in main we could create an SQL script with the style of jinja. This is store in the folder of templates.

In this case, I've stored in github repo because dev.to doesn't let you to use double brackets

If someone knows how it could be copied in here without any problems it could be great.

Python Script

Use the Models.py and also the Jinja2Templates, we could use this to get the connection and the result of the query:

from fastapi.templating import Jinja2Templates
from sqlalchemy import text

from models import engine, TABLE_NAME

# Import templates of Jinja2
templates = Jinja2Templates(directory="templates")

class Sql_conn:

    def _parse(self, query_dict: dict[str]):
        # Use dict to pass variables throught jinja template
        query_dict['table']= TABLE_NAME
        render = templates.TemplateResponse(
                "query_template.j2",{'request': None, 'data': query_dict}
            )
        return render.body.decode('ascii')

    def _query(self, q):
       # Use connection to make a query
        print(q)
        with engine.connect() as con:
            rs = con.execute(text(q))
        return rs
    def _resultdict(self,result):
        # Convert result to dict 
        resultdic = dict()
        for idx, row in enumerate(result):
            row_as_dict = row._mapping
            resultdic.update({idx:row_as_dict})
        return resultdic

    def request_query(self, query_dict: dict[str]):
        q = self._parse(query_dict)
        data = self._query(q)
        datadic = self._resultdict(data)
        return datadic

Api.py

Finally in this script, we are going to use fastapi. In here based on the last script (backend.py), we are going to create our desired api:

from typing import Optional
from fastapi import FastAPI, HTTPException
from sqlalchemy.exc import ProgrammingError
from pydantic import BaseModel


from backend import Sql_conn


# Variables to the model
class Query(BaseModel):
    select: str
    where : Optional[str]=None
    order_by : Optional[str]=None
    group_by : Optional[str]=None
    sort_by : Optional[str]=None

# Create our object of FastAPI
app = FastAPI()

# Based on the method post of analytics..
@app.post("/analytics")
async def query_data(query: Query):
    # ... we create  async function to get the result of a query 
    return Sql_conn().request_query(query.dict())

Dockerfile.python

We are going to create an instance to run our desired dockerfile to run our api.

FROM python:3.9

WORKDIR /app

# Install OS dependencies
RUN apt-get update && apt-get install -qq -y \
    git gcc build-essential libpq-dev --fix-missing --no-install-recommends \ 
    && apt-get clean

# Make sure we are using latest pip
RUN pip install --upgrade pip

# Copy requirements.txt
COPY requirements.txt requirements.txt

# ... install of the requirements
RUN pip install -r requirements.txt

# ... copy all the script into the app
COPY ./py_script/ /app/

# ... select the variable of environment where is going to execute python
ENV PYTHONPATH "${PYTHONPATH}:/app/"

# ... and expose the desired port
EXPOSE 8000

CMD ["uvicorn","api:app","--host","0.0.0.0"]

Docker compose

Finally to get all the enviornment working together I've create this script:

version: '3.9'
services:
  post_db:
    build:
        context: .
        dockerfile: Dockerfile.postgres
    user: postgres
    image: postgres/test:v1
    environment:
      - POSTGRES_USER=${PS_USER}
      - POSTGRES_PASSWORD=${PS_PASSWORD}
      - PG_DATA:/var/lib/postgresql/data/pgdata
      - POSTGRES_DB=${PS_DB}
    healthcheck:
      test: ["CMD-SHELL","pg_isready -U ${PS_USER} ${PS_PASSWORD}"]
      interval: 10s
      timeout: 5s
      retries: 5
    ports:
      - "5432:5432"
    volumes:
      - db-data:/var/lib/postgresql/data
    restart: unless-stopped
  api:
    build:
        context: .
        dockerfile: Dockerfile.python
    environment:
      - DB_CONNSTR=postgresql://${PS_USER}:${PS_PASSWORD}@post_db:5432/${PS_DB}
    ports:
      - 8000:8000
    links:
      - post_db
    restart: unless-stopped

volumes:
  db-data: 

Now we could acces to our http://localhost:8000/docs and check if there is any in action:

Tests

To check if everything is working properly I've made this test. You could access to make this test with tryout button:

{
  "select": "num_age as Age, count(*) as Num_row",
  "group_by": "num_age",
"order_by": "count(*) desc"

}

And also there is our desired result:

Conclusion

FastAPI is a really good packacge to get the information about your database and also there is a lot of features I've not test it yet, but it looks great to have it as a server to get access to your data.

Have a nice day:)