Building a barebone Web API service in Python without a web framework

Introduction

Have you ever wondered how Python web frameworks work under the hood? Are you interested in just playing with python and not the complexities of its web frameworks such as Django, Flask, FastAPI, and so on? Do you know that with just Python, you can have some functionalities built? Do you want to explore the popular Python WSGI HTTP Server for UNIX, Gunicorn 'Green Unicorn'? If these sound interesting to you, welcome onboard! We will be exploiting the capabilities of gunicorn to serve our "sketchy", barebone, and "not-recommended-for-production" web API service with the following features:

Getting users added to the app
Allowing users to check balances, deposit and withdraw money to other app users and out of the app.

NOTE: This application does not incorporate any real database. It stores data in memory.

Assumptions

It is assumed you have a basic understanding of Python, and how the web works.

Source code

The entire source code for this article can be accessed via:

Sirneij / peer_to_peer

Barebone web api in python without any framework

A barebone web API service, powered by gunicorn, that uses query paramenters to modify an in-memory data.

CAVEAT: The application may sometimes misbehave. It's barebone, as written.

Starting locally

Clone this project and change directory into it:

You first need to have a copy of this project. To do so, open up your terminal (PowerShell in Windows or WSL terminal) and run the following commands:

$: git clone https://github.com/Sirneij/peer_to_peer.git

$: cd peer_to_peer

Create and activate virtual environment

Create and activate python virtual environment. You are free to use any python package of choice but I opted for ven:

$: python3 -m venv virtualenv

$: source virtualenv/bin/activate.fish

I used the .fish version of activate because my default shell is fish.

Install dpendencies and run the application

This app does not need fancy framework, the only required dependency is gunicorn which serves the application. Other dependencies are not required, they…

View on GitHub

Implementation

Step 1: Create a new project

As with every project, we need to create it. Open up your terminal/cmd/PowerShell and navigate to the directory where the project will be housed. Then create the project. For me, I used poetry to bootstrap a typical python project like so:

sirneij@pop-os ~/D/Projects> poetry new peer_to_peer

I named the project peer_to_peer. Change the directory into it. If poetry was used, you should have a file structure like:

.
├── peer_to_peer
│   ├── __init__.py
├── pyproject.toml
├── README.md
├── tests
│   ├── __init__.py

To build this app, we need one dependency, gunicorn. It'll help serve that entire application. Let's install it in our app's virtual environment.

(virtualenv) sirneij@pop-os ~/D/P/peer_to_peer (main)> pip install gunicorn

Step 2: Create important files for the project

Create some files, server.py, models.py, urls.py, and handlers.py, inside the peer_to_peer folder.

sirneij@pop-os ~/D/P/peer_to_peer (main)> touch peer_to_peer/server.py peer_to_peer/urls.py peer_to_peer/models.py peer_to_peer/handlers.py

server.py does exactly what its name implies. It is the entry point of the app. models.py will hold the app's "model" or more appropriately, in-memory database. urls.py routes all requests to the appropriate logic. handlers.py houses the logic of each route.

Step 3: Design the app's database

Let's employ the database-first approach by defining the kind of data our app needs. It will be a Python class with a single field which is a list of dictionaries, _user_data.

# peer_to_peer/models.py

from typing import Any


class User:
    _user_data: list[dict[str, Any]] = []

    @property
    def user_data(self) -> list[dict[str, Any]]:
        return self._user_data

    def set_user_data(self, data):
        self._user_data = self._user_data.append(data)

    def return_user(self, username) -> dict[str, Any] | None:
        for d in self._user_data:
            if username in d.values():
                return d
        return None

This field was made "private" but a "getter" property, user_data, gets its value for the "outside world" to use. We also defined a "setter", set_user_data method that appends data to it. There was a return_user method which searches the "database" for a user via the "unique" username (or name) and returns such user in case it's found. Pretty basic!

Let's proceed to the content of the server.py.

Step 4: Write the app's entry script

# peer_to_peer/server.py

from typing import Iterator

from peer_to_peer.models import User
from peer_to_peer.urls import url_handlers


def app(environ, start_reponse) -> Iterator[bytes]:
    user = User()
    return iter([url_handlers(environ, start_reponse, user)])

The simple script above is the app's entry point. It's a two-liner housed by the app function. This function takes the environ and start_reponse arguments, the requirements for gunicorn apps as defined here. The environ serves as the "request" object which contains all the details of all incoming requests to the app. As for the start_reponse, it defines the response's status code and headers. This function returns an Iterator of bytes. For data consistency, we passed only one instance of the User model defined previously to the url_hadndlers, housed in the urls.py file. The content of which is shown below:

# peer_to_peer/urls.py
import json

from peer_to_peer.handlers import (
    add_money,
    add_user,
    check_balance,
    index,
    transfer_money_out,
    transfer_money_to_user,
)
from peer_to_peer.models import User


def url_handlers(environ, start_reponse, user: User):
    path = environ.get('PATH_INFO')
    if path.endswith('/'):
        path = path[:-1]

    if path == '':
        context = index(environ, user)
        data = json.dumps(context.get('data')) if context.get('data') else json.dumps(context.get('error'))
        status = context['status']

    elif path == '/add-user':
        context = add_user(environ, user)
        data = json.dumps(context.get('data')) if context.get('data') else json.dumps(context.get('error'))
        status = context['status']

    elif path == '/add-money':
        context = add_money(environ, user)
        data = json.dumps(context.get('data')) if context.get('data') else json.dumps(context.get('error'))
        status = context['status']
    elif path == '/check-balance':
        context = check_balance(environ, user)
        data = json.dumps(context.get('data')) if context.get('data') else json.dumps(context.get('error'))
        status = context['status']
    elif path == '/transfer-money-to-user':
        context = transfer_money_to_user(environ, user)
        data = json.dumps(context.get('data')) if context.get('data') else json.dumps(context.get('error'))
        status = context['status']
    elif path == '/transfer-money-out':
        context = transfer_money_out(environ, user)
        data = json.dumps(context.get('data')) if context.get('data') else json.dumps(context.get('error'))
        status = context['status']
    else:
        data, status = json.dumps({'error': '404 Not Found'}), '400 Not FOund'

    data = data.encode('utf-8')
    content_type = 'application/json' if int(status.split(' ')[0]) < 400 else 'text/plain'
    response_headers = [('Content-Type', content_type), ('Content-Length', str(len(data)))]

    start_reponse(status, response_headers)
    return data

If you are familiar with any of the Python web frameworks mentioned earlier, this is equivalent to how requests are being routed. The environ contains, among many other things, the PATH_INFO which represents the URL entered into your browser. From the path info, we tried calling different logic as contained in the handlers.py, to be discussed soon. For each path, we turn the data or error returned into JSON using python's JSON module and then extract the status of the request. Later on, the data were encoded to be utf-8-compliant and the headers were set accordingly. Now to the handlers.py:

# peer_to_peer/handlers.py
from typing import Any
from urllib import parse

from peer_to_peer.models import User


def index(environ, user: User) -> dict[str, Any]:
    """Display the in-memory data to users."""
    request_params = dict(parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query))

    context = {'status': '200 Ok'}
    if request_params and request_params.get('name').replace('\'', '').lower() == 'admin':
        user_data = user.user_data
        for data in user_data:
            if data:
                if data.get('password'):
                    data.pop('password')
        context['data'] = user_data

    elif request_params and request_params.get('name').replace('\'', '').lower():
        current_user = user.return_user(request_params.get('name').replace('\'', '').lower())
        context['data'] = current_user

    else:
        context['data'] = 'You cannot just view this page without a `name` query parameter.'
    return context


def add_user(environ, user: User) -> dict[str, Any]:
    """Use query parameters to add users to the in-memory data."""

    if parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query) == '':
        return {'error': '405 Method Not Allowed', 'status': '405 Method Not Allowed'}
    request_meta_query = dict(parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query))
    if not request_meta_query:
        return {'error': 'Query must be provided.', 'status': '405 Method Not Allowed'}

    user_data = user.user_data
    if not any(d.get('username') == request_meta_query.get('name').replace('\'', '').lower() for d in user_data):
        if all(not data for data in user_data):
            user.set_user_data(
                {
                    'id': 1,
                    'username': request_meta_query.get('name').replace('\'', '').lower(),
                    'password': request_meta_query.get('password'),
                }
            )
        else:
            user.set_user_data(
                {
                    'id': user_data[-1]['id'] + 1,
                    'username': request_meta_query.get('name').replace('\'', '').lower(),
                    'password': request_meta_query.get('password'),
                }
            )

    else:
        return {
            'error': f'A user with username, {request_meta_query.get("name")}, already exists.',
            'status': '409 Conflict',
        }
    context = {'data': user_data[-1], 'status': '200 Ok'}
    return context


def add_money(environ, user: User) -> dict[str, Any]:
    """Use query parameters to add money to a user's account to the in-memory data."""
    if parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query) == '':
        return {'error': '405 Method Not Allowed', 'status': '405 Method Not Allowed'}
    request_meta_query = dict(parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query))
    if not request_meta_query:
        return {'error': 'Query must be provided.', 'status': '405 Method Not Allowed'}

    context = {'status': '200 Ok'}

    user_data = user.return_user(request_meta_query.get('name').replace('\'', '').lower())
    if user_data:
        if user_data['password'] == request_meta_query.get('password'):
            user_data['balance'] = user_data.get('balance', 0.0) + float(request_meta_query.get('amount'))
            context['data'] = user_data
        else:
            return {
                'error': 'You are not authorized to add money to this user\'s balance.',
                'status': '401 Unauthorized',
            }
    else:
        return {'error': 'A user with that name does not exist.', 'status': '404 Not Found'}

    return context


def check_balance(environ, user: User) -> dict[str, Any]:
    """Use query parameters to check a user's account balance to the in-memory data."""
    if parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query) == '':
        return {'error': '405 Method Not Allowed', 'status': '405 Method Not Allowed'}
    request_meta_query = dict(parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query))
    if not request_meta_query:
        return {'error': 'Query must be provided.', 'status': '405 Method Not Allowed'}

    context = {'status': '200 Ok'}

    user_data = user.return_user(request_meta_query.get('name').replace('\'', '').lower())
    if user_data:
        password = request_meta_query.get('password')
        if password:
            if user_data['password'] == password:
                context['data'] = {'balance': user_data.get('balance', 0.0)}
            else:
                return {
                    'error': 'You are not authorized to check this user\'s balance.',
                    'status': '401 Unauthorized',
                }
        else:
            return {
                'error': 'You must provide the user\'s password to check balance.',
                'status': '401 Unauthorized',
            }
    else:
        return {'error': 'A user with that name does not exist.', 'status': '404 Not Found'}
    return context


def transfer_money_to_user(environ, user: User) -> dict[str, Any]:
    """Use query parameters to transfer money from a user to another."""
    if parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query) == '':
        return {'error': '405 Method Not Allowed', 'status': '405 Method Not Allowed'}
    request_meta_query = dict(parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query))
    if not request_meta_query:
        return {'error': 'Query must be provided.', 'status': '405 Method Not Allowed'}

    context = {'status': '200 Ok'}

    user_data = user.return_user(request_meta_query.get('from_name').replace('\'', '').lower())

    if user_data:
        if user_data['password'] == request_meta_query.get('from_password'):
            if request_meta_query.get('amount') and user_data.get('balance', 0.0) >= float(
                request_meta_query.get('amount')
            ):
                beneficiary = user.return_user(request_meta_query.get('to_name').replace('\'', '').lower())
                if beneficiary:
                    beneficiary['balance'] = beneficiary.get('balance', 0.0) + float(request_meta_query['amount'])
                    user_data['balance'] = user_data['balance'] - float(request_meta_query['amount'])
                    context[
                        'data'
                    ] = f'A sum of ${float(request_meta_query["amount"])} was successfully transferred to {request_meta_query.get("to_name")}.'
                else:
                    return {
                        'error': 'The user you want to credit does not exist.',
                        'status': '404 Not Found',
                    }
            else:
                return {
                    'error': 'You either have insufficient funds or did not include `amount` as query parameter.',
                    'status': '404 Not Found',
                }
        else:
            return {
                'error': 'You are not authorized to access this user\'s account.',
                'status': '401 Unauthorized',
            }
    else:
        return {'error': 'A user with that name does not exist.', 'status': '404 Not Found'}
    return context


def transfer_money_out(environ, user: User) -> dict[str, Any]:
    """Use query parameters to transfer money out of this app."""
    if parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query) == '':
        return {'error': '405 Method Not Allowed', 'status': '405 Method Not Allowed'}
    request_meta_query = dict(parse.parse_qsl(parse.urlsplit(environ.get('RAW_URI')).query))
    if not request_meta_query:
        return {'error': 'Query must be provided.', 'status': '405 Method Not Allowed'}

    context = {'status': '200 Ok'}

    user_data = user.return_user(request_meta_query.get('name').replace('\'', '').lower())
    if user_data:
        if user_data.get('password') == request_meta_query.get('password'):
            if request_meta_query.get('amount') and user_data.get('balance') >= float(request_meta_query.get('amount')):
                user_data['balance'] = user_data['balance'] - float(request_meta_query['amount'])
                context[
                    'data'
                ] = f'A sum of ${float(request_meta_query["amount"])} was successfully transferred to {request_meta_query.get("to_bank")}.'
            else:
                return {
                    'error': 'You either have insufficient funds or did not include `amount` as query parameter.',
                    'status': '404 Not Found',
                }
        else:
            return {
                'error': 'You are not authorized to access this user\'s account.',
                'status': '401 Unauthorized',
            }
    else:
        return {'error': 'A user with that name does not exist.', 'status': '404 Not Found'}
    return context

That's lengthy! However, taking a closer look will reveal that these lines are basic python codes. Each route gets and parses the query parameter(s) included by the user(s). If no parameter was included, an appropriate error will be returned. In case there is/are query parameter(s), different simple logic that manipulates the in-memory data we have was implemented. All pretty simple if looked at.

Step 5: Running the app

Having gone through the app's build-up, we can now run it using:

(virtualenv) sirneij@pop-os ~/D/P/peer_to_peer (main)> gunicorn peer_to_peer.server:app --reload -w 5

We are giving the app 5 workers using the -w 5 flag. You can also turn on DEBUG mode by:

(virtualenv) sirneij@pop-os ~/D/P/peer_to_peer (main)> gunicorn peer_to_peer.server:app --reload --log-level DEBUG -w 5

You then can play with the "APIs". A comprehensive work-through of how to send requests and all are available in the project's README on github.

Outro

Enjoyed this article? Consider contacting me for a job, something worthwhile or buying a coffee ☕. You can also connect with/follow me on LinkedIn. Also, it isn't bad if you help share it for wider coverage. I will appreciate it...

Blog