Python Web App Structure

Last update: 16 October 2023

In addition to code that works, I believe that having a code that is well structured, easy to read and maintain is equally important. In this article, we will explore a method to achieve this goal.

Although the structure we discuss is applicable to any framework, for the purpose of the illustration, we will build an API to manage users using FastAPI.

You can find the final code here 👉 https://github.com/pythops/python-webapp

Big picture

Here is the structure that we will explore through this article

Starter boilerplate

Let's start with a simple boilerplate first, and we'll build on the top of it as we progress in developing the app.

.
├── app
│  ├── __init__.py
│  └── api
│     ├── controllers.py
│     ├── routes.py
│     └── schemas.py
├── Justfile
├── poetry.lock
├── pypoetry.toml
├── pyproject.toml
├── run.py
└── tests

`init.py`

This is where we define the application factory.

from fastapi import FastAPI

def init_app():
    api = FastAPI()

    from app.api import routes

    api.include_router(routes.router)

    return api

Then we call it in the run.py

from app import init_app

app = init_app()

To run the api, we'll need an ASGI web server. We'll use uvicorn in our case.

$ uvicorn run:app --port 8080

`schemas.py`

This is where we define all the schemas for the routes. This serves for validation and serialization of the requests and the responses.

from pydantic import BaseModel


class User(BaseModel):
    username: str
    email: str

`routes.py`

This is where we define all the routes. This should be as simple as possible and should NOT contain any logic as the logic goes to the controllers.

from fastapi import APIRouter, status

from app.api import controllers, schemas


router = APIRouter()


@router.post("/users", status_code=status.HTTP_204_NO_CONTENT)
async def create_new_user(user: schemas.User):
    await controllers.create_new_user(username=user.username, email=user.email)


@router.get("/users", status_code=status.HTTP_200_OK)
async def get_users() -> list[schemas.User]:
    return await controllers.get_users()

`controllers.py`

This where the logic/processing happens.

from app.models import User


async def create_new_user(username: str, email: str):
    await User.create(username=username, email=email)


async def get_users() -> list:
    users = await User.get_all()
    return [await user.to_dict() for user in users]

Handling Configuration

The configuration for the app should be defined in the config.py module

app
├── __init__.py
├── api
│  ├── controllers.py
│  ├── routes.py
│  └── schemas.py
└── config.py <<< NEW!

In the config.py module, we setup the configuration as follows:

import os


class Config:
    DB_URI = os.getenv(
        "DB_URI", "postgresql+asyncpg://pythops:pythops@127.0.0.1:5432/pythops"
    )


config = Config()

ℹ️ Note

The app configuration should be changeable from various sources, such as environment variables. In case where we have constants in the configuration, it is recommended to relocate them to a consts module `app/consts.py`.

Then we can access the configuration in any module inside the app :

from app.config import config

Handling the exceptions

To handle the exceptions properly, we'll need two modules:

app/exceptions.py
app/errors.py

app
├── __init__.py
├── api
│  ├── controllers.py
│  ├── routes.py
│  └── schemas.py
├── config.py
├── errors.py <<< NEW!
└── exceptions.py <<< NEW!

`exceptions.py`

This is where we define the custom exceptions.

class BaseException(Exception):
    def __init__(self, **kwargs):
        super().__init__()
        self.kwargs = kwargs


class UserAlreadyExists(BaseException):
    message = "User Already Exists"

The BaseException gives us the option to add contextual information when we raise the exception which could be helpful for debugging purposes. In our case, we could add the username to get more insights into the affected user.

raise UserAlreadyExists(username=username)

`errors.py`

This module will catch the exceptions to return a clean and readable response to the client.

If we try to create a user that exists already, we will return a 409 error code.

from fastapi.responses import JSONResponse

from app import exceptions


def error_handler(app):
    @app.exception_handler(exceptions.UserAlreadyExists)
    async def http_409(request, exception):
        return JSONResponse(status_code=409, content={"error": str(exception)})

Then we need to register the errors in the app factory

# app/__init__.py

def init_app():
    ...

    api = FastAPI()

    from app.errors import error_handler

    error_handler(api)

    ...

Using a database

Database initialization

One way to integrate a database is to use an ORM. For our case, we'll be using SQLAlchemy as we'll need a SQL database.

First, we create the app/database.py module under the app folder

app
├── __init__.py
├── api
│  ├── controllers.py
│  ├── routes.py
│  └── schemas.py
├── config.py
├── database.py <<< NEW!
├── errors.py
└── exceptions.py

`database.py`

The database module should be structured this way:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import declarative_base


Base = declarative_base()


class Database:
    def __init__(self):
        self._session: AsyncSession | None = None

    def __getattr__(self, name):
        return getattr(self._session, name)

    def init(self, uri):
        engine = create_async_engine(uri, future=True)
        self._session = async_sessionmaker(engine)()


db = Database()

in the app/__init__.py we initialize the database session:

from app.database import db
form app.config import config


def init_app():
    ...
    db.init(uri=config.DB_URI)
    ..

Then we can call the database session where we need it:

from app.database import db

Database models

The last part is to create the app/models.py module which serves as a place for defining the data structure.

app
├── __init__.py
├── api
│  ├── controllers.py
│  ├── routes.py
│  └── schemas.py
├── config.py
├── database.py
├── errors.py
├── exceptions.py
└── models.py <<< NEW!

`models.py`

Here we define a User model with some class methods to avoid repetition, and keep the database logic inside the models module.

from uuid import uuid4
from sqlalchemy import Column, String
from sqlalchemy.future import select

from app.database import Base, db
from app.exceptions import UserAlreadyExist


class User(Base):
    __tablename__ = "user"

    user_id = Column(String, primary_key=True, nullable=False)
    email = Column(String, nullable=False)
    username = Column(String, nullable=False, unique=True)

    @classmethod
    async def create(cls, email: str, username: str):
        user = await cls.get(username=username)
        if user:
            raise UserAlreadyExist(username=username)
        new_user = cls(
            user_id=str(uuid4()),
            email=email,
            username=username,
        )
        try:
            db.add(new_user)
            await db.commit()
        except Exception:
            await db.rollback()
            raise
        return user

    @classmethod
    async def get(cls, username: str) -> list:
        query = select(cls).where(cls.username == username)
        result = await db.execute(query)
        user = result.one().User
        return user

    @classmethod
    async def get_all(cls) -> list:
        query = select(cls)
        result = await db.execute(query)
        users = result.all()
        return [user.User for user in users]

    async def to_dict(self):
        return {
            "username": self.username,
            "email": self.email,
        }

Using Cache

In case we need to use caching in the app. For our case, we'll go with Redis .

We create the app/cache.py module first

app
├── __init__.py
├── api
│  ├── controllers.py
│  ├── routes.py
│  └── schemas.py
├── cache.py <<< NEW!
├── config.py
├── database.py
├── errors.py
├── exceptions.py
└── models.py

`cache.py`

import redis.asyncio as redis
from redis.asyncio.client import Redis


class Cache:
    def __init__(self):
        self._client: Redis | None = None

    def __getattr__(self, name):
        return getattr(self._client, name)

    def init(self, host: str, port: int, password: str, db: str):
        self._client = redis.from_url(
            f"redis://{host}",
            port=port,
            password=password,
            db=db,
            socket_connect_timeout=2.0,
            encoding="utf-8",
            decode_responses=True,
        )


cache = Cache()

In the app/__init__.py we initialize the cache service :

from app.cache import cache

def init_app():
    ...
    cache.init(...)
    ...

Now we can access the cache as follows:

from app.cache import cache

ℹ️ Note

Redis does not impose any schema for the data but if you want to model and validate the data for Redis you might want to consider using [redis-orm-python](https://github.com/redis/redis-om-python)

Using an http client

When the app requires calling external HTTP services, we initialize an HTTP client during startup and utilize it as needed.

The structure of http client is as follows:

app
├── __init__.py
├── api
│  ├── controllers.py
│  ├── routes.py
│  └── schemas.py
├── cache.py
├── config.py
├── database.py
├── errors.py
├── exceptions.py
├── http  <<< NEW!
│  ├── __init__.py
│  ├── client.py
│  └── exceptions.py
└── models.py

`client.py`

This is where we define the HTTP methods we'll be using. We also define the retry and the timeout policy.

`exceptions.py`

This where we define the exceptions in the case the HTTP call fails for any raison.

In the app/__init__.py we initialize the http client as follows

from app.http import http

def init_app():
    ...
    http.init()
    ...

Now we can import the http client in any module.

from app.http import http

response = http.get(url="https://pythops.com")

Calling external services

When using external services, we can put them inside app/services directory with all their dependencies (exceptions, serializers etc). For instance

app
├── __init__.py
...
└── services << NEW!
   └── foo
      ├── __init__.py
      ├── exceptions.py
      ├── serializers.py
      └── service.py

Conclusion

With this structure, it's really easy to navigate through the code, which makes it easy to maintain and that's a huge advantage when you're building something big. Feel free to add your suggestions or comments by opening a Github issue, I'll be curious to know what do you think about this approach.

PythOps

Python Web App Structure

Big picture

Starter boilerplate

__init__.py

schemas.py

routes.py

controllers.py

Handling Configuration

ℹ️ Note

Handling the exceptions

exceptions.py

errors.py

Using a database

Database initialization

database.py

Database models

models.py

Using Cache

cache.py

ℹ️ Note

Using an http client

client.py

exceptions.py

Calling external services

Conclusion

`init.py`

`schemas.py`

`routes.py`

`controllers.py`

`exceptions.py`

`errors.py`

`database.py`

`models.py`

`cache.py`

`client.py`

`exceptions.py`