4 Layers Architecture

At this point, your code is running as expected. But, if you take a closer look, it's not "clean" enough.

Not "clean"

Here are some of the not-clean points of the current code:

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = sqlite3.connect(DATABASE)
        g._database = db

It’s not advisable to directly utilize a global variable for storing the database instance and accessing it throughout the codebase. Although the get_db singleton function and the flask.g context mitigate this issue to some extent.

db = sqlite3.connect(DATABASE)
g._database = db
cursor = db.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS books (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    author TEXT NOT NULL,
    published_at TEXT NOT NULL,
    description TEXT NOT NULL,
    isbn TEXT NOT NULL,
    total_pages INTEGER NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )''')
cursor.close()

It’s not a good practice to initialize a database in the business code. Database operations belong to the infrastructure layer, not the business logic. These are distinct concerns and should be separated accordingly.

@app.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
    query = "SELECT * FROM books WHERE id = ?"
    books = execute_query(query, (book_id,))
    if not books:
        return {"error": "Record not found"}, 404
    return jsonify(books[0])

It’s not recommended to write SQL queries directly in your route handlers. The cost could be significant if you decide to switch to a new ORM framework in the future.

Separation of Concerns

Clean Arch From https://blog.cleancoder.com/uncle-bob/2012/08/13/the-clean-architecture.html

The Clean Architecture is a software architectural pattern introduced by Robert C. Martin, in his book "Clean Architecture: A Craftsman's Guide to Software Structure and Design." It emphasizes the separation of concerns within a software system, with the primary goal of making the system more maintainable, scalable, and testable over its lifecycle.

4 Layers Architecture varies somewhat in the details, but it's similar to the Clean Architecture. They all have the same objective, which is the separation of concerns. They all achieve this separation by dividing the software into layers.

4 layers

Adapter Layer: Responsible for routing and adaptation of frameworks, protocols, and front-end displays (web, mobile, and etc).
Application Layer: Primarily responsible for obtaining input, assembling context, parameter validation, invoking domain layer for business processing. The layer is open-ended, it's okay for the application layer to bypass the domain layer and directly access the infrastructure layer.
Domain Layer: Encapsulates core business logic and provides business entities and business logic operations to the Application layer through domain services and domain entities' methods. The domain is the core of the application and does not depend on any other layers.
Infrastructure Layer: Primarily handles technical details such as CRUD operations on databases, search engines, file systems, RPC for distributed services, etc. External dependencies need to be translated through gateways here before they can be used by the Application and Domain layers above.

Refactoring

Let's do the code refactoring with 4 layers architecture.

The new folder structure looks like this:

projects/lr_rest_books_py
├── LICENSE
├── README.md
├── books
│   ├── __init__.py
│   ├── adapter
│   │   ├── __init__.py
│   │   ├── router.py
│   │   └── util.py
│   ├── application
│   │   ├── __init__.py
│   │   ├── executor
│   │   │   ├── __init__.py
│   │   │   └── book_operator.py
│   │   └── wire_helper.py
│   ├── domain
│   │   ├── __init__.py
│   │   ├── gateway
│   │   │   ├── __init__.py
│   │   │   └── book_manager.py
│   │   └── model
│   │       ├── __init__.py
│   │       └── book.py
│   └── infrastructure
│       ├── __init__.py
│       ├── config
│       │   ├── __init__.py
│       │   └── config.py
│       └── database
│           ├── __init__.py
│           └── sqlite.py
├── main.py
├── requirements.txt
└── test.db

Move model/book.py into domain/model/book.py:

Note: This tutorial ignores the books/ prefix in paths for convenience.

from dataclasses import dataclass
from datetime import datetime

@dataclass
class Book:
    id: int
    title: str
    author: str
    published_at: str
    description: str
    isbn: str
    total_pages: int
    created_at: datetime
    updated_at: datetime

domain folder is where all your core business rules reside. Non-business code stay away from this folder.

Create domain/gateway/book_manager.py:

from abc import ABC, abstractmethod
from typing import List, Optional

from ..model import Book


class BookManager(ABC):
    @abstractmethod
    def create_book(self, b: Book) -> int:
        pass

    @abstractmethod
    def update_book(self, id: int, b: Book) -> None:
        pass

    @abstractmethod
    def delete_book(self, id: int) -> None:
        pass

    @abstractmethod
    def get_book(self, id: int) -> Optional[Book]:
        pass

    @abstractmethod
    def get_books(self) -> List[Book]:
        pass

Python does not have a built-in construct called "interface" like some other programming languages. However, Python allows you to achieve similar functionality using Abstract Base Classes (ABCs) from the abc module.

The BookManager abstract class defines the business capabilities of the domain entity Book. It is abstract, and the implementations are done in the infrastructure layer. These methods also allow the application layer to utilize them for business logic.

Create infrastructrue/database/sqlite.py:

import sqlite3
from typing import List, Optional

from ...domain.gateway import BookManager
from ...domain.model import Book


class SQLitePersistence(BookManager):

    def __init__(self, file_name: str):
        self._file_name = file_name
        self._create_table()

    def _create_table(self):
        conn, cursor = self._connect()
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS books (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            author TEXT NOT NULL,
            published_at TEXT NOT NULL,
            description TEXT NOT NULL,
            isbn TEXT NOT NULL,
            total_pages INTEGER NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
          )
       ''')
        conn.commit()

    def _connect(self):
        # You cannot use a SQLite connection object across multiple threads in Flask.
        # Flask is designed to be multithreaded for handling multiple requests simultaneously.
        conn = sqlite3.connect(self._file_name)
        return conn, conn.cursor()

    def create_book(self, b: Book) -> int:
        conn, cursor = self._connect()
        cursor.execute('''
            INSERT INTO books (title, author, published_at, description, isbn, total_pages) VALUES (?, ?, ?, ?, ?, ?)
        ''', (b.title, b.author, b.published_at, b.description, b.isbn, b.total_pages))
        conn.commit()
        return cursor.lastrowid or 0

    def update_book(self, id: int, b: Book) -> None:
        conn, cursor = self._connect()
        cursor.execute('''
            UPDATE books SET title=?, author=?, published_at=?, description=?, isbn=?, total_pages=?,updated_at=DATETIME('now') WHERE id=?
        ''', (b.title, b.author, b.published_at, b.description, b.isbn, b.total_pages, id))
        conn.commit()

    def delete_book(self, id: int) -> None:
        conn, cursor = self._connect()
        cursor.execute('''
            DELETE FROM books WHERE id=?
        ''', (id,))
        conn.commit()

    def get_book(self, id: int) -> Optional[Book]:
        _, cursor = self._connect()
        cursor.execute('''
            SELECT * FROM books WHERE id=?
        ''', (id,))
        result = cursor.fetchone()
        if result:
            return Book(*result)
        return None

    def get_books(self) -> List[Book]:
        _, cursor = self._connect()
        cursor.execute('''
            SELECT * FROM books
        ''')
        results = cursor.fetchall()
        return [Book(*result) for result in results]

As you see here, class SQLitePersistence implements all the methods declared in the abstract class BookManager.

Create infrastructrue/config/config.py:

from dataclasses import dataclass


@dataclass
class DBConfig:
    file_name: str


@dataclass
class ApplicationConfig:
    port: int


@dataclass
class Config:
    app: ApplicationConfig
    db: DBConfig

The Config class pulls out some hard-coded items in the codebase, which is a good practice in general.

Create application/executor/book_operator.py:

from typing import List, Optional
from ...domain.model import Book
from ...domain.gateway import BookManager


class BookOperator():

    def __init__(self, book_manager: BookManager):
        self.book_manager = book_manager

    def create_book(self, b: Book) -> Book:
        id = self.book_manager.create_book(b)
        b.id = id
        return b

    def get_book(self, id: int) -> Optional[Book]:
        return self.book_manager.get_book(id)

    def get_books(self) -> List[Book]:
        return self.book_manager.get_books()

    def update_book(self, id: int, b: Book) -> Book:
        self.book_manager.update_book(id, b)
        return b

    def delete_book(self, id: int) -> None:
        return self.book_manager.delete_book(id)

The application layer encapsulates everything and provides interfaces for the topmost adapter layer.

Create application/wire_helper.py:

from ..domain.gateway import BookManager
from ..infrastructure.config import Config
from ..infrastructure.database import SQLitePersistence


class WireHelper:
    def __init__(self, persistence: SQLitePersistence):
        self.persistence = persistence

    @classmethod
    def new(cls, c: Config):
        db = SQLitePersistence(c.db.file_name)
        return cls(db)

    def book_manager(self) -> BookManager:
        return self.persistence

WireHelper helps with DI (dependency injection). If you want advanced DI support, consider the injector framework.

With all these preparation work, finally the router can do its job "cleanly". It doesn't know which database is used under the hood. Concerns are separated now.

Create adapter/router.py:

import logging
from flask import Flask, request, jsonify

from ..application.executor import BookOperator
from ..application import WireHelper
from ..domain.model import Book
from .util import dataclass_from_dict


class RestHandler:
    def __init__(self, logger: logging.Logger, book_operator: BookOperator):
        self._logger = logger
        self.book_operator = book_operator

    def get_books(self):
        try:
            books = self.book_operator.get_books()
            return jsonify(books), 200
        except Exception as e:
            self._logger.error(f"Failed to get books: {e}")
            return jsonify({"error": "Failed to get books"}), 404

    def get_book(self, id):
        try:
            book = self.book_operator.get_book(id)
            if not book:
                return jsonify({"error": f"The book with id {id} does not exist"}), 404
            return jsonify(book), 200
        except Exception as e:
            self._logger.error(f"Failed to get the book with {id}: {e}")
            return jsonify({"error": "Failed to get the book"}), 404

    def create_book(self):
        try:
            b = dataclass_from_dict(Book, request.json)
            book = self.book_operator.create_book(b)
            return jsonify(book), 201
        except Exception as e:
            self._logger.error(f"Failed to create: {e}")
            return jsonify({"error": "Failed to create"}), 404

    def update_book(self, id):
        try:
            b = dataclass_from_dict(Book, request.json)
            book = self.book_operator.update_book(id, b)
            return jsonify(book), 200
        except Exception as e:
            self._logger.error(f"Failed to update: {e}")
            return jsonify({"error": "Failed to update"}), 404

    def delete_book(self, id):
        try:
            self.book_operator.delete_book(id)
            return "", 204
        except Exception as e:
            self._logger.error(f"Failed to delete: {e}")
            return jsonify({"error": "Failed to delete"}), 404


def health():
    return jsonify({"status": "ok"})


def make_router(app: Flask, wire_helper: WireHelper):
    rest_handler = RestHandler(
        app.logger, BookOperator(wire_helper.book_manager()))
    app.add_url_rule('/', view_func=health)
    app.add_url_rule('/books', view_func=rest_handler.get_books)
    app.add_url_rule('/books/<int:id>', view_func=rest_handler.get_book)
    app.add_url_rule('/books', view_func=rest_handler.create_book,
                     methods=['POST'])
    app.add_url_rule('/books/<int:id>', view_func=rest_handler.update_book,
                     methods=['PUT'])
    app.add_url_rule('/books/<int:id>', view_func=rest_handler.delete_book,
                     methods=['DELETE'])

The route() decorator is a shortcut to call app.add_url_rule with the view_func argument.

These are equivalent:

@app.route("/")
def index():
    ...

def index():
    ...

app.add_url_rule("/", view_func=index)

Create adapter/util.py for helper functions:

import dataclasses


def dataclass_from_dict(klass, d):
    class_d = {field.name: field.default if field.default !=
               field.default_factory else None for field in dataclasses.fields(klass)}
    return klass(**{**class_d, **d})

Replace main.py with the following content to make it clean:

from flask import Flask

from books.adapter.router import make_router
from books.application import WireHelper
from books.infrastructure.config import Config, ApplicationConfig, DBConfig

c = Config(
    ApplicationConfig(
        8080
    ),
    DBConfig(
        "test.db"
    )
)
wire_helper = WireHelper.new(c)
app = Flask(__name__)
make_router(app, wire_helper)

Run it again to test endpoints:

flask --app main run --debug

Everything works! Refactoring is achieved. 🎉Great!

❮ Prev Next ❯