In this tutorial, we will walk through creating a simple yet organized Python API using Flask and PostgreSQL. We will use SQLAlchemy to interact with the database and organize our code into separate modules for maintainability. The goal is to create a clean, scalable structure for building and expanding the application as needed.

In this tutorial, we are going to use the below directory structure:

├── app.py
├── configs
│   ├── db_config.py
├── controllers
│   └── user_controller.py
├── models
│   ├── base_model.py
│   └── user_model.py
├── repository
│   └── user_repository.py
├── requirements.txt
├── serializers
│   └── user_serializer.py
├── services
│   └── user_service.py
└── utils
   └── helper_functions.py

• app.py: The main entry point for the application.
• models: Contains database models.
• controllers: Handles HTTP requests and responses.
• repository: Responsible for database operations.
• services: Contains business logic.
• serializers: Validates and serializes input data.
• utils: Utility functions, such as password handling

Step By Step Guide To Build an API

Step 1: Project Setup

We begin by setting up a Python project and installing the necessary packages. To install Flask, SQLAlchemy, and psycopg2 (the PostgreSQL adapter for Python), open the terminal and run the following command

pip install flask sqlalchemy psycopg2

Step 2: Database Configuration

Create a file in the configs directory and add the code snippet below –

import os
class Config:
   SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'postgresql+psycopg2://username:password@localhost/my_api_db')
   SQLALCHEMY_TRACK_MODIFICATIONS = False

Step 3: Setting Up the Database Models

Now, let’s define our database models. These models typically represent the structure of our database tables. Using an ORM (Object-Relational Mapping) makes managing database queries straightforward and secure, as it automatically helps prevent SQL injection. Create a file in the “models” directory and add the following code snippet:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from configs.db_config import DATABASE_URI
Base = declarative_base()
engine = create_engine(DATABASE_URI)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Now let’s set up our user model. Create another file in the model’s directory and add the code snippet below –

from sqlalchemy import Column, Integer, String
from .base_model import Base
class User(Base):
   __tablename__ = "users"
   id = Column(Integer, primary_key=True, index=True)
   first_name = Column(String, nullable=False)
   last_name = Column(String, nullable=False)
   email = Column(String, unique=True, nullable=False)
   mobile_number = Column(String, unique=True, nullable=False)
   password = Column(String, nullable=False)

Step 4: Controllers Layer

Controllers basically control our data flow throughout the request/response cycle. We set all our routes here so let’s create a file in the controller’s directory and add the code snippet –

from flask import Blueprint, request, jsonify
from sqlalchemy.orm import Session
from models.base_model import SessionLocal
from services.user_service import (
   add_user_service, list_users_service, get_user_service,
   update_user_service, delete_user_service
)
from serializers.user_serializer import UserCreate, UserUpdate, UserResponse
from pydantic import ValidationError
user_bp = Blueprint("user", __name__)
def get_db():
   db = SessionLocal()
   try:
       yield db
   finally:
       db.close()
@user_bp.route("/", methods=["POST"])
def create_user():
   db = next(get_db())
   try:
       data = request.get_json()
       user_data = UserCreate(**data)
       user = add_user_service(db, user_data)
       return jsonify(UserResponse.from_orm(user).dict()), 201
   except ValidationError as e:
       return jsonify({"error": e.errors()}), 400
@user_bp.route("/", methods=["GET"])
def list_users():
   db = next(get_db())
   users = list_users_service(db)
   return jsonify([UserResponse.from_orm(user).dict() for user in users]), 200
@user_bp.route("/<int:user_id>", methods=["GET"])
def get_user(user_id):
   db = next(get_db())
   user = get_user_service(db, user_id)
   if not user:
       return jsonify({"error": "User not found"}), 404
   return jsonify(UserResponse.from_orm(user).dict()), 200
@user_bp.route("/<int:user_id>", methods=["PATCH"])
def update_user(user_id):
   db = next(get_db())
   try:
       data = request.get_json()
       updates = UserUpdate(**data)
       user = update_user_service(db, user_id, updates)
       if not user:
           return jsonify({"error": "User not found"}), 404
       return jsonify(UserResponse.from_orm(user).dict()), 200
   except ValidationError as e:
       return jsonify({"error": e.errors()}), 400
@user_bp.route("/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
   db = next(get_db())
   user = delete_user_service(db, user_id)
   if not user:
       return jsonify({"error": "User not found"}), 404
   return jsonify({"message": "User deleted"}), 200

Step 5: Service Layer

This layer contains our core logic—all kinds of things like checking duplicity and hashing of passwords, etc. We usually do that in the service layer. Create a file in the services directory and add the below lines –

from repository.user_repository import (
   create_user, get_all_users, get_user_by_id, update_user, delete_user
)
from utils.helper_functions import hash_password
from sqlalchemy.orm import Session
from serializers.user_serializer import UserCreate, UserUpdate
def add_user_service(db: Session, user_data: UserCreate):
   user_data.password = hash_password(user_data.password)
   return create_user(db, user_data.dict())
def list_users_service(db: Session):
   return get_all_users(db)
def get_user_service(db: Session, user_id: int):
   return get_user_by_id(db, user_id)
def update_user_service(db: Session, user_id: int, updates: UserUpdate):
   if updates.password:
       updates.password = hash_password(updates.password)
   return update_user(db, user_id, updates.dict(exclude_unset=True))
def delete_user_service(db: Session, user_id: int):
   return delete_user(db, user_id)

Step 6: Repository Layer

This layer communicates with the Database. Create a file in the repository directory and add the code snippet below –

from sqlalchemy.orm import Session
from models.user_model import User
def create_user(db: Session, user_data: dict):
   user = User(**user_data)
   db.add(user)
   db.commit()
   db.refresh(user)
   return user
def get_all_users(db: Session):
   return db.query(User).all()
def get_user_by_id(db: Session, user_id: int):
   return db.query(User).filter(User.id == user_id).first()
def update_user(db: Session, user_id: int, updates: dict):
   user = db.query(User).filter(User.id == user_id).first()
   if user:
       for key, value in updates.items():
           setattr(user, key, value)
       db.commit()
       db.refresh(user)
   return user
def delete_user(db: Session, user_id: int):
   user = db.query(User).filter(User.id == user_id).first()
   if user:
       db.delete(user)
       db.commit()
   return user

Step 7: Serializers layer

Serializers are a way to serialize our output. We can control what should be the contents of the output because we don’t want to send the hashed password in the API response. Create a file in the serializers directory and add the code snippet –

from pydantic import BaseModel, Field, EmailStr
class UserBase(BaseModel):
   first_name: str = Field(..., min_length=1, max_length=50)
   last_name: str = Field(..., min_length=1, max_length=50)
   email: str = Field(..., pattern=r"^[^@]+@[^@]+\.[^@]+$")
   mobile_number: str = Field(..., pattern=r"^\+?\d{10,15}$")
class UserCreate(UserBase):
   password: str = Field(..., min_length=6)
class UserUpdate(BaseModel):
   first_name: str
   last_name: str
   email: EmailStr
   mobile_number: str
   password: str
class UserResponse(UserBase):
   id: int
   class Config:
       from_attributes = True

Step 8: Utils layer

The utils layer contains small utility functions that can be reused in our app. Create a file in the utils directory and add the code snippet –

from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
   return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
   return pwd_context.verify(plain_password, hashed_password)

Step 9: Entry point

Finally, create an entry point for our application. This file will serve as the main execution point and should be placed in the root directory. Add the following code to this file

from flask import Flask
from controllers.user_controller import user_bp
from models.base_model import Base, engine
app = Flask(__name__)
Base.metadata.create_all(bind=engine)
app.register_blueprint(user_bp, url_prefix="/users")
if __name__ == "__main__":
   app.run(debug=True)

Step 10: Running our application

Add your database URL in the config file open the terminal and use the command –

python3 app.pyyour application with start to run on http://localhost:5000 and you can access your routes as follows –

POST /users - Create a new user
GET /users - Get all users
GET /users/<user_id> - Get a user by ID
PATCH /users/<user_id> - Update a user by ID
DELETE /users/<user_id> - Delete a user by ID

Conclusion

This Python API is built using Flask for handling web requests and SQLAlchemy for interacting with a PostgreSQL database. The application follows a modular structure, making it easy to maintain and expand as it grows. Here’s a breakdown of the key components:

app.py — This is the main entry point of the app. It sets up the Flask application, registers routes, and connects to the database.
models — This file contains the database models, defining the structure of the tables and handling database operations.
controllers — These are the functions that handle incoming HTTP requests, validate input, and send appropriate responses.
repository — This layer is responsible for directly interacting with the database, performing CRUD (Create, Read, Update, Delete) operations.
services — Here is where the core business logic resides, including processing data and managing interactions with the repository layer.
serializers — These ensure that input data is validated and serialized correctly before passing it to the services layer.
utils — This folder contains utility functions for tasks like password hashing and verification.

This structure keeps the application organized, easy to extend, and clean. By separating concerns into specific files and directories, the application can grow and adapt without becoming unmanageable.