Creating a Python REST API project using aiohttp

We’ll see how to create a complete production level Python REST API project using aiohttp framework. We Will then see how to deploy it on Azure using Docker container.

In the application:

We will have Get/POST/PUT/DELETE and a startup method.

We will be using config file to keep our config keys like connection string.

We will be using a Logger file for logging our responses.

We will show how to prepare requirement.txt file which contains the name of the required packages for our project, We’ll see how we can keep an SDK file also.

We will be using Visual Studio Code for it, using Python version 3.9

So lets begin….

0 – Create a folder with name of project example – APIDEMO, then open this folder in Visual Studio.

1 – Create a folder with name Controller

2- Create a confit file with name config.ini

Keep below elements inside it as key value:

[DEFAULT]
mykey1=abc
mykey2=def

Below is the stucture after you would complete the files creation:

3 – Create a file inside Controller folder with name TestController.py

Below is the code for controller

from datetime import datetime
import json
from aiohttp import web
from ConfigurationManager import ConfigurationManager

configuration_manager = ConfigurationManager.get_instance()
mykey = configuration_manager.get_app_settings("mykey1")

class TestController():

    def __init__(self):
        app = web.Application()
        app.add_routes([web.get('/getUser', TestController.get_data)])
        app.add_routes([web.put('/updateUser', TestController.update_user)])
        app.add_routes([web.post('/addUser', TestController.add_user)])
        app.add_routes([web.delete('/deleteUser', TestController.delete_user)])
        app.add_routes([web.get('/', TestController.startup)])
        web.run_app(app, port=5000)

    async def get_data(request):
        try:
            response_obj = { 'request_type' : 'Get request received',

            'username':'test' }
            return web.Response(text=json.dumps(response_obj), status=200)

        except Exception as ex:
            return web.Response(text=str(ex), status=500)

    async def update_user(request):
        try:
            response_obj = { 'request_type' : 'Update request received' }
            return web.Response(text=json.dumps(response_obj), status=200)

        except Exception as ex:
            return web.Response(text=str(ex), status=500)


    async def add_user(request):
        try:
            user = request.query['name']
            print("Creating new user with name: " , user)
            response_obj = { 'status' : 'Post request succeed' }
            return web.Response(text=json.dumps(response_obj), status=200)
        except Exception as e:
            ## Bad path where name is not set
            response_obj = { 'status' : 'failed', 'reason': str(e) }
            ## return failed with a status code of 500 i.e. 'Server Error'
            return web.Response(text=json.dumps(response_obj), status=500)

    async def delete_user(request):
        try:
            response_obj = { 'request_type' : 'Delete request received' }
            return web.Response(text=json.dumps(response_obj), status=200)

        except Exception as ex:
            return web.Response(text=str(ex), status=500)


    def startup(request):
        return web.Response(text="App is running....." +
          '\n\n\ntime is = ' + str(datetime.now()) +
          '\n\n\nmy config key value is = ' + mykey)

4 – Create a Logger file inside main folder and write below code inside it:

import enum

class Logger(enum.Enum):
    INFORMATION = 1
    ERROR = 2

    @staticmethod
    def log_message(messageType: object, message: str):
        logMessage = messageType.name + " : " + message
        print(logMessage)

5 – Create the main file App.py inside main folder. It contains the startup main method where it will call the TestController.

Write below code there.

from Controller.TestController import TestController
from Logger import Logger

class App():
    def __init__():
        Logger.log_message(Logger.INFORMATION, "Starting Python REST API... ")


if __name__ == "__main__":
    TestController()

We will need a configuration manager to read config file. So we will create a file ConfigurationManager.py and write below code in it:

import configparser


class ConfigurationManager:
    __configuration = None
    __instance = None

    def __init__(self):
        if (self.__configuration == None):
            self.__configuration = configparser.ConfigParser()
            self.__configuration.read('config.ini')

    @staticmethod
    def get_instance():
        if (ConfigurationManager.__instance == None):
            ConfigurationManager.__instance = ConfigurationManager()

        return ConfigurationManager.__instance

    def get_app_settings(self, key):
        if (key != None):
            return self.__configuration.get('DEFAULT', key)
        return None

To keep information about packages that we have used in the project, we will create a file call requirements.txt its equivalent to package.json file in Node projects.

If you have a SDK/wheel file keep its path reference inside the requirements.txt

Our requirements.txt contains only one package reference, so its like below: