Adopting the Repository Pattern for Enhanced Backend Development With FastAPI

This article will not guide you through the process of setting up a virtual environment and installing FastAPI to build a to-do application. Instead, it will offer a new perspective, If you have experience with using FastAPI, for creating backend systems. In the past, you may have performed database queries within your API view or business domain module. However, it is recommended to stop this practice. The repository pattern is a better option as it promotes writing cleaner code and avoids potential issues. In this article, I will explain the reasons for using this design pattern and the advantages it brings.

Introduction

A repository pattern is an interface where data access logic is stored. To put it simply, the beautiful idea behind using a repository pattern is to decouple the data access layer (database) from the service access layer (domain) of the application. CRUD (create, read, update, delete, etc) operations are done through methods without having to worry about database connections, commands, etc.

We will be building a simple URL shortener for the sake of this article. You will find the complete code of the project on my GitHub; you must watch closely to see how incredibly awesome RP (repository pattern) is.

In our models, we have the following code:

# Stdlib Imports
from datetime import datetime

# SQLAlchemy Imports
from sqlalchemy import Column, String, Integer, DateTime

# Own Imports
from config.database import Base, DATABASE_ENGINE


async def create_tables():
    Link.metadata.create_all(bind=DATABASE_ENGINE)


class Link(Base):
    __tablename__ = "links"

    id = Column(Integer, primary_key=True, index=True)
    original = Column(String)
    shortened = Column(String(4))
    date_created = Column(DateTime, default=datetime.now)
    date_modified = Column(DateTime, onupdate=datetime.now)

    def __str__(cls) -> str:
        """
        `__str__` is a special method that returns a string representation of an object.

        :param cls: The class object
        :return: The shortened version of the original link
        """
        return cls.shortened.__str__()

For the sake of the readers that are new to FastAPI, I will quickly go over what create_tables and Link does. The async method create_tables is responsible for creating the database tables of the defined links class. We are defining the necessary fields (columns) we want to have in our links table.

Next would be the module called repository.py, this is where we are going to create the interface where our data access logic is stored. We have the following code:

# Stdlib Imports
from typing import List

# SQLAlchemy Imports
from sqlalchemy.orm import Session

# Own Imports
from config.deps import get_db
from shortener.models import Link


class LinkRepository:
    """Repository responsible for performing operations (CRUD, etc) on links table."""

    def __init__(self) -> None:
        self.db: Session = get_db().__next__()

    async def create(self, original: str, shortened: str) -> Link:
        """
        This method is responsible for creating a new link object.

        :param original: original url link
        :param shortened: shortened url link

        :return: the link object
        """

        link = Link(original=original, shortened=shortened)

        self.db.add(link)
        self.db.commit()
        self.db.refresh(link)

        return link

    async def get(self, skip: int, end: int) -> List[Link]:
        """
        This method retrieves a list of links objects.

        :param skip: The number of links to skip
        :type skip: int

        :param end: The number of links to retrieve
        :type end: int

        :return: A list of link objects
        """

        links = self.db.query(Link).offset(skip).limit(end).all()
        return links

    async def get_code(self, code: str) -> Link:
        """
        This method retrieves a link object that matches the code.

        :param code: The shortened code of the link
        :type code: str

        :return: The link object
        """

        link = self.db.query(Link).filter_by(shortened=code).first()
        return link


link_repository = LinkRepository()

To begin using our repository class; we need to initialize it.

def __init__(self) -> None:
    self.db: Session = get_db().__next__()

The purpose of this method is to initialize an instance of the class by creating a database session and storing it in an instance variable self.db. By doing this, the database session will be created and stored in the instance of the class when the class is instantiated and will be available for use throughout the lifetime of the instance.

get_db is a function that is responsible for creating a database session, yielding it, and rolling back the transaction If an exception occurs during the database operation, the function will catch the exception and call the db.rollback() method to rollback the transaction. This ensures that any changes made to the database during the transaction are undone in case of an error; and finally the block of the function will close the session using the db.close() method, regardless of whether an exception was raised or not.

The code for this function is as follows:

# Own Imports
from config.database import SessionLocal


def get_db():
    """
    This function creates a database session,
    yield it to the get_db function, rollback the transaction
    if there's an exception and then finally closes the session.

    Yields:
        db: scoped database session
    """

    db = SessionLocal()
    try:
        yield db
    except Exception:
        db.rollback()
    finally:
        db.close()

The create method is responsible for creating a new Link object with the given original URL link and shortened code. After which it stores it in the link variable. Then, the method adds the link to the database by calling self.db.add(link).

Next, the method commits the changes to the database by calling self.db.commit(). The refresh method is then called with the link object as an argument to refresh the state of the link in the database. Finally, the method returns the link object to the caller.

The get method is responsible for retrieving a list of Link objects from the database. The method creates a query to retrieve Link objects from the database using the self.db.query(Link) statement. The offset method is then called on the query to specify the number of links to skip, and the limit method is called to specify the number of links to retrieve. Finally, the all method is called on the query to retrieve all the links that match the criteria.

The get_code is responsible for retrieving a single Link object. The method creates a query to retrieve a Link object from the database using the self.db.query(Link) statement. The filter_by method is then called on the query to specify the criteria for retrieving the link, in this case, the shortened code. Finally, the first method is called on the query to retrieve the first link that matches the criteria.

In summary, the methods provide a convenient and asynchronous way of doing a certain thing in the database, be it; creating and storing new Link objects in the database, or to retrieve a list of Link objects from the database or to retrieve a single Link object while ensuring that the database session is properly managed and the transaction is committed if there are no exceptions.

Now that you understand what is going on, let’s move on to our services access layer; this is where the domain logic is stored. We have the following code:

# Stdlib Imports
import random
import string

# FastAPI Imports
from fastapi.responses import RedirectResponse

# Own Imports
from shortener.repository import link_repository, Link


async def shorten_link() -> str:
    """
    This function returns a random string of 4 characters.

    :return: A string of 4 random letters.
    """
    
    shrt_str = "".join(random.choice(string.ascii_letters) for i in range(4))
    return shrt_str


async def create_shortened_link(original: str) -> Link:
    """
    This function creates a shortened link for the given original link.

    :param original: str - the original link that we want to shorten
    :type original: str

    :return: A Link object
    """

    shortened_link = await shorten_link()
    link = await link_repository.create(original, shortened_link)
    return link


async def redirect_to_original_link(code: str) -> str:
    """
    This function takes a code and returns the original link.

    :param code: str - the code that was generated by the shortener
    :type code: str

    :return: redirect to original link
    """
    
    link = await link_repository.get_code(code)
    return RedirectResponse(link.original)

If you take a look at the functions create_shortened_link and redirect_to_original_link, we are accessing the create and get_code methods in our repository interface. We have eliminated writing database queries in our service layer, making our code clean and quick to debug.

Advantages of Using Repository Pattern

Using repository design patterns offers several advantages, including:

1. Facilitating the testing of your application's logic
2. Reducing the occurrence of duplicated database operations
3. Central management of the database layer, enabling the implementation of access policies with ease
4. The ability to define strong annotations for domain entities.

Final Thoughts

In this article, we explained the benefits of using the repository design pattern in software development. The repository pattern separates the data access layer from the service access layer in an application. This leads to a cleaner and more organized code base, and reduces duplicated database operations.

We used a simple URL shortener as an example to illustrate the implementation of the repository pattern. The code demonstrates how to create a database table and how to store data access logic in a repository class. Find the complete code to this example here.