Building a barebone Web API service in Python without a web framework
Tags: python, tutorial, beginners, webdev
Published: 2022-09-10 05:42:03

## 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][1], [Flask][2], [FastAPI][3], 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'][4]? If these sound interesting to you, welcome onboard! We will be exploiting the capabilities of [gunicorn][4] 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:

{% github Sirneij/peer_to_peer %}

## 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][5] to bootstrap a typical python project like so:

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

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

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

```

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

```bash
(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.

```bash
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`.

```python
# 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

```python
# 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][4] apps as defined [here][6]. 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:

```python
# 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][7] 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`:

```python
# 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:

```bash
(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:

```bash
(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][8].

## Outro

Enjoyed this article? Consider [contacting me for a job, something worthwhile or buying a coffee ☕](mailto:sirneij@gmail.com). You can also connect with/follow me on [LinkedIn](https://www.linkedin.com/in/idogun-john-nelson/). Also, it isn't bad if you help share it for wider coverage. I will appreciate it...

[1]: https://www.djangoproject.com/ "Django Web Framework"
[2]: https://flask.palletsprojects.com/en/2.2.x/ "Flask Microframework"
[3]: https://fastapi.tiangolo.com/ "FastAPI framework"
[4]: https://gunicorn.org/ "Gunicorn 'Green Unicorn'"
[5]: https://python-poetry.org/ "Poetry: Python Packaging and Dependencies made easy"
[6]: https://docs.gunicorn.org/en/latest/run.html#gunicorn "Running Gunicorn"
[7]: https://www.json.org/json-en.html "JavaScript Object Notation"
[8]: https://github.com/Sirneij/peer_to_peer "Peer to Peer GitHub Repository"