Created FastAPI - SOLID Principles and Design Patterns (markdown)

+https://medium.com/@lautisuarez081/fastapi-best-practices-and-design-patterns-building-quality-python-apis-31774ff3c28a

+

+# FastAPI: SOLID Principles and Design Patterns | Medium

+In recent years, FastAPI has emerged as one of the most popular frameworks for building Python APIs thanks to its simplicity, speed and support for static typing. However, to get the most out of this powerful tool, it is essential to follow some good practices and design patterns that allow us to write clean, maintainable and scalable code.

+

+In this article, I will show you how to apply some of the SOLID principles and design patterns such as DAO (Data Access Object), Service Layer, and Dependency Injection to build robust and efficient APIs with FastAPI.

+

+SOLID Principles Applied to FastAPI

+-----------------------------------

+

+**1\. Single Responsibility Principle (SRP)**

+---------------------------------------------

+

+> Each module or class should have responsibility for only one part of the functionality provided by the software, and this responsibility should be encapsulated in its entirety by the class.

+

+What does this mean? For example, in a FastAPI application, a routing function (endpoint) should focus on receiving a request, delegating business logic to a specific service, and returning a response. Let’s see this in code:

+

+**Code without SRP:**

+

+```python

+from fastapi import APIRouter

+from app.models.user import UserCreate, UserRead

+from app.db import database

+

+router = APIRouter()

+

+@router.post("/users", response_model=UserRead)

+async def create_user(user: UserCreate):

+ # Data validation

+ if not user.email or not user.password:

+ raise ValueError("Email and password are required.")

+

+ # Check if the user already exists

+ existing_user = database.fetch_one("SELECT * FROM users WHERE email = :email", {"email": user.email})

+ if existing_user:

+ raise ValueError("User already exists.")

+

+ # Create a new user in the database

+ new_user_id = database.execute("INSERT INTO users (email, password) VALUES (:email, :password)", {

+ "email": user.email,

+ "password": user.password

+ })

+

+ # Get new user details

+ new_user = database.fetch_one("SELECT * FROM users WHERE id = :id", {"id": new_user_id})

+

+ return new_user

+```

+

+

+In this example, the create user endpoint **does multiple tasks**:

+

+\- Input data validation (validate email and password).

+\- Verify if the user already exists in the database.

+\- Create a new user in the database.

+\- Retrieve and return the details of the new user.

+

+This mix of responsibilities makes the code more difficult to maintain and scalable. Any change in business logic or how data is accessed will require modifications to this same block of code, increasing the chances of errors.

+

+**Code with SRP:**

+

+```python

+# ------- REPOSITORY FILE -------

+from app.models.user import UserCreate, UserDB

+from app.db import database

+

+class UserRepository:

+ def __init__(self, db_session):

+ self.db_session = db_session

+

+ async def get_user_by_email(self, email: str) -> UserDB:

+ query = "SELECT * FROM users WHERE email = :email"

+ return await self.db_session.fetch_one(query, {"email": email})

+

+ async def add_user(self, user_data: UserCreate) -> int:

+ query = "INSERT INTO users (email, password) VALUES (:email, :password) RETURNING id"

+ values = {"email": user_data.email, "password": user_data.password}

+ new_user_id = await self.db_session.execute(query, values)

+ return new_user_id

+

+ async def get_user_by_id(self, user_id: int) -> UserDB:

+ query = "SELECT * FROM users WHERE id = :id"

+ return await self.db_session.fetch_one(query, {"id": user_id})

+

+

+# ------- SERVICE FILE -------

+from app.models.user import UserCreate, UserRead

+from app.repositories.user_repository import UserRepository

+

+class UserService:

+ def __init__(self, user_repository: UserRepository):

+ self.user_repository = user_repository

+

+ async def validate_user_data(self, user_data: UserCreate) -> None:

+ if not user_data.email or not user_data.password:

+ raise ValueError("Email and password are required.")

+

+ async def check_user_exists(self, email: str) -> None:

+ existing_user = await self.user_repository.get_user_by_email(email)

+ if existing_user:

+ raise ValueError("User already exists.")

+

+ async def create_user(self, user_data: UserCreate) -> UserRead:

+ # Business Logic Validation

+ await self.validate_user_data(user_data)

+ await self.check_user_exists(user_data.email)

+ new_user_id = await self.user_repository.add_user(user_data)

+ return await self.user_repository.get_user_by_id(new_user_id)

+

+

+# ------- USER ROUTER FILE -------

+from fastapi import APIRouter, Depends

+from app.models.user import UserCreate, UserRead

+from app.services.user_service import UserService

+from app.routers.dependencies import get_user_service

+

+router = APIRouter()

+

+@router.post("/users", response_model=UserRead)

+async def create_user(user: UserCreate, user_service: UserService = Depends(get_user_service)):

+ return await user_service.create_user(user)

+```

+

+

+Now, in this example, you can clearly see the unique responsibility of each module.

+

+* **User Repository:** handles all database related operations, such as getting or inserting users. Any change in the database structure or the way data is handled will be done here.

+* **User Service:** Contains the business logic related to the users, such as validations, business rules, etc. It is the intermediary between the repository and the FastAPI routes.

+* **User Router:** The function of the endpoint is very simple: receive an HTTP request, delegate the logic to the corresponding service, and return the response. It does not care about the internal details of how the data or business logic is handled.

+

+By applying SRP in your FastAPI application, you not only make your code cleaner and easier to maintain, but also to establish a **solid foundation for the future expansion** of its application.

+

+2\. **Dependency Inversion Principle (DIP)**

+--------------------------------------------

+

+This principle sets out 2 rules:

+

+> 1\. High-level modules should not depend on low-level modules. Both must depend on abstractions.

+>

+> 2\. Abstractions should not depend on details. Details must depend on abstractions.

+

+**Returning to the previous example:** The UserService class depends directly on the concrete implementation of UserRepository. This is a DIP violation because UserService (high-level module) depends on UserRepository (low-level module). If tomorrow you decide to change the way UserRepository handles data persistence (for example, migrating from SQL to NoSQL), you would also have to modify UserService.

+

+To apply the DIP correctly, we must introduce an abstraction (an interface or base class) that defines the contract for the operations that UserService needs to interact with the user repository. In this way, UserService will depend on the abstraction and not on the concrete implementation.

+

+We will call this class IUserRepository, referring to the fact that it is an interface:

+

+```

+from abc import ABC, abstractmethod

+from app.models.user import UserCreate, UserRead

+class IUserRepository(ABC):

+ @abstractmethod

+ async def get_user_by_email(self, email: str) -> UserRead:

+ pass

+ @abstractmethod

+ async def add_user(self, user_data: UserCreate) -> int:

+ pass

+ @abstractmethod

+ async def get_user_by_id(self, user_id: int) -> UserRead:

+ pass

+```

+

+

+This interface defines the methods that any concrete implementation of a user repository must have. This ensures that UserService can work with any repository that implements this interface, independent of how it handles data.

+

+Now we must make UserRepository implement this interface:

+

+```

+from app.models.user import UserCreate, UserRead

+from app.db import database

+from app.repositories.user_repository_interface import IUserRepository

+class UserRepository(IUserRepository):

+ def __init__(self, db_session):

+ self.db_session = db_session

+ async def get_user_by_email(self, email: str) -> UserRead:

+ query = "SELECT * FROM users WHERE email = :email"

+ return await self.db_session.fetch_one(query, {"email": email})

+ async def add_user(self, user_data: UserCreate) -> int:

+ query = "INSERT INTO users (email, password) VALUES (:email, :password) RETURNING id"

+ values = {"email": user_data.email, "password": user_data.password}

+ new_user_id = await self.db_session.execute(query, values)

+ return new_user_id

+ async def get_user_by_id(self, user_id: int) -> UserRead:

+ query = "SELECT * FROM users WHERE id = :id"

+ return await self.db_session.fetch_one(query, {"id": user_id})

+```

+

+

+Implementing the interface ensures that it provides all the necessary functionalities that UserService might need.

+

+Finally, we have to make our UserService class depend on the interface and not on the concrete class:

+

+```

+from app.models.user import UserCreate, UserRead

+from app.repositories.user_repository_interface import IUserRepository

+class UserService:

+ def __init__(self, user_repository: IUserRepository):

+ self.user_repository = user_repository

+ async def validate_user_data(self, user_data: UserCreate) -> None:

+ if not user_data.email or not user_data.password:

+ raise ValueError("Email and password are required.")

+ async def check_user_exists(self, email: str) -> None:

+ existing_user = await self.user_repository.get_user_by_email(email)

+ if existing_user:

+ raise ValueError("User already exists.")

+ async def create_user(self, user_data: UserCreate) -> UserRead:

+ await self.validate_user_data(user_data)

+ await self.check_user_exists(user_data.email)

+ new_user_id = await self.user_repository.add_user(user_data)

+ new_user = await self.user_repository.get_user_by_id(new_user_id)

+ return new_user

+```

+

+

+Applying the Dependency Inversion Principle (DIP) improves your application architecture by decoupling business logic from implementation details. In this example, UserService becomes **more flexible, maintainable and easier to test** by relying on the IUserRepository interface instead of a concrete implementation suc**Story**h as UserRepository.

+

+Applied design patterns

+-----------------------

+

+In the examples above, in addition to using SOLID principles, we are using some design patterns. Did you manage to identify which ones?

+

+**The DAO pattern and the Service Layer**, now I will explain them in detail:

+

+**1\. Data Access Object (DAO)**

+--------------------------------

+

+The DAO Pattern is a design pattern used to separate the data access logic from the business logic of the application. Its purpose is to provide an abstraction for CRUD (Create, Read, Update, Delete) operations that are performed on a database or other data source.

+

+In the example we are working on we can **identify the UserRepository class**. This class will be responsible for all interactions with the database related to the User entity.

+

+```

+from app.models.user import UserCreate, UserRead

+from app.db import database

+from app.repositories.user_repository_interface import IUserRepository

+class UserRepository(IUserRepository):

+ def __init__(self, db_session):

+ self.db_session = db_session

+ async def get_user_by_email(self, email: str) -> UserRead:

+ query = "SELECT * FROM users WHERE email = :email"

+ return await self.db_session.fetch_one(query, {"email": email})

+ async def add_user(self, user_data: UserCreate) -> int:

+ query = "INSERT INTO users (email, password) VALUES (:email, :password) RETURNING id"

+ values = {"email": user_data.email, "password": user_data.password}

+ new_user_id = await self.db_session.execute(query, values)

+ return new_user_id

+ async def get_user_by_id(self, user_id: int) -> UserRead:

+ query = "SELECT * FROM users WHERE id = :id"

+ return await self.db_session.fetch_one(query, {"id": user_id})

+```

+

+

+Why should we use this pattern?

+

+* **Data Access Encapsulation:** The DAO provides a dedicated layer to manage all data access operations. This means that any change in the persistence logic (e.g., switching from SQL to NoSQL) is performed only in the DAO layer, without affecting the rest of the application.

+* **Reusability:** The DAO implementation can be reused by different services or components that need to interact with data from the same entity, eliminating code duplication.

+* **Easy Testing:** By separating data access in its own layer, it is easy to create mocks or stubs for unit testing, allowing the business logic to be tested in isolation without depending on the actual database.

+* **Maintainability:** Data access operations are centralized in a DAO class, making it easy to locate and correct persistence-related errors.

+

+2\. **Service Layer**

+---------------------

+

+It is responsible for organizing the business logic of the application. Its purpose is to provide an interface to the presentation layer or controllers (e.g. FastAPI controllers) that encapsulates all relevant business logic.

+

+By separating the business logic into a service layer, a more modular, more maintainable and testable code is achieved. In addition, it facilitates the reuse of business logic in different application contexts.

+

+In the example, we **use this pattern in the UserService class**. This class is responsible for the business logic related to users. This business logic may include validations, business rules, data transformation, among others. It will use the DAO to perform data access operations, but it will not directly manage the database.

+

+```

+from app.models.user import UserCreate, UserRead

+from app.repositories.user_repository_interface import IUserRepository

+class UserService:

+ def __init__(self, user_repository: IUserRepository):

+ self.user_repository = user_repository

+ async def validate_user_data(self, user_data: UserCreate) -> None:

+ if not user_data.email or not user_data.password:

+ raise ValueError("Email and password are required.")

+ async def check_user_exists(self, email: str) -> None:

+ existing_user = await self.user_repository.get_user_by_email(email)

+ if existing_user:

+ raise ValueError("User already exists.")

+ async def create_user(self, user_data: UserCreate) -> UserRead:

+ await self.validate_user_data(user_data)

+ await self.check_user_exists(user_data.email)

+ new_user_id = await self.user_repository.add_user(user_data)

+ new_user = await self.user_repository.get_user_by_id(new_user_id)

+ return new_user

+```

+

+

+Why should we use this pattern?

+

+* **Separation of Business Logic:** Provides a centralized place for all business logic, eliminating duplicate logic in different parts of the application and keeping it separate from the data access logic.

+* **Easy Unit Test:** By encapsulating the business logic in a separate service layer, it is easier to perform unit tests on that business logic without having to worry about the details of the database. This allows creating different Mockups to test different scenarios.

+* **Code Decoupling:** Presentation layers (such as controllers in FastAPI) do not depend directly on the business logic or data access layer. This allows changes to the business logic without affecting the user interface or vice versa.

+* **Flexibility and Extensibility:** Acts as a central point where additional changes, rules or validations can be applied to the business logic without affecting the rest of the application.

+

+Conclusion

+----------

+

+By following these design principles and patterns, you will be able to build more robust, flexible and maintainable APIs using FastAPI. By applying principles such as SOLID and using patterns such as DAO or Service Layer you will not only improve the quality of your code, but also its ability to adapt to changes and grow over time.