FastAPI CRUD app with Raw SQL | Part 1

9 min readJul 7, 2023

In this article, we will explore how to build a FastAPI CRUD (Create, Read, Update, Delete) application using raw SQL queries with the psycopg2 library. We will compare the advantages and disadvantages of using raw SQL in a FastAPI application versus using ORM libraries. By the end of this article, you will have a solid understanding of how to leverage raw SQL in FastAPI and make an informed decision on whether it is the right approach for your project.

1. Introduction:

FastAPI is a modern, fast, web framework for building APIs with Python that has gained popularity for its excellent performance and ease of use. It leverages Python 3.7+ type hints and other modern features to provide highly efficient code and automatic documentation generation. FastAPI is built on top of Starlette, a powerful ASGI framework, which allows it to handle high loads and scale effectively.

When it comes to interacting with databases in FastAPI, one common approach is to use the psycopg2 library for PostgreSQL database management. Psycopg2 is a widely used and well-established Python adapter for PostgreSQL that provides a simple and efficient way to work with PostgreSQL databases.

2. Pros and Cons of Raw SQL:

Raw SQL refers to the practice of writing SQL queries directly in your application code, without the use of Object-Relational Mapping (ORM) libraries. Let’s explore some pros and cons of using raw SQL in a FastAPI application:

Pros:

Flexibility: Raw SQL gives you full control over your queries, allowing you to take advantage of database-specific features and optimizations. You can write complex queries tailored to your specific needs.
Performance: By crafting optimized SQL queries, you can often achieve better performance compared to using ORM libraries. You have direct control over the execution plan and can fine-tune it for better efficiency.
Migration: When your database schema changes, using raw SQL can make it easier to apply migration scripts, as you have direct control over the SQL statements executed.

Cons:

Code Maintainability: Writing raw SQL queries in your application code can make it harder to maintain and understand the logic, especially as the complexity of your application grows.
Boilerplate Code: Raw SQL requires more code to handle the mapping between query results and Python objects manually. This can result in additional development effort and potential sources of bugs.
Portability: Using raw SQL ties your application closely to a specific database engine, making it more challenging to switch to a different database if needed.

3. Getting Started: Setting up the Project

Create news_fastapi folder and open it with your favorite code editor. Open up your terminal and navigate to news_fastapi directory. It is recommended to create a virtual environment for python projects.

$ cd news_fastapi
# create virtual environment using venv module(-m) and name it as venv
$ python3.11 -m venv venv
# activate venv
$ source venv/bin/activate
# windows users can activate like the following
$ venv\Scripts\activate

Install FastAPI

# after activating venv use pip to install fastapi
(venv) $ pip install fastapi

Also install uvicorn to work as the server:

(venv) $ pip install "uvicorn[standard]"

Create app and tests folders under the news_fastapi folder. For now project files and folders structure is like the following.

news_fastapi
├──  requirements.txt
├── app
└── tests
└── venv

Create main.py under the app folder and add the following codes:

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def index():
    return {'message': 'Welcome to news app!'}

This code sets up a basic FastAPI application with a single route defined for the root path (“/”). Here’s a breakdown of what each part does:

from fastapi import FastAPI: This line imports the FastAPI class from the fastapi module. FastAPI is the main class that we'll use to create our application.
app = FastAPI(): This line creates an instance of the FastAPI class and assigns it to the app variable. We'll use this app object to define our routes and run the application.
@app.get('/'): This is a decorator that tells FastAPI that the following function should be executed when a GET request is made to the root path ("/").
async def index():: This defines the function index as an asynchronous function. The async keyword allows us to use await inside the function.
return {'message': 'Welcome to news app!'}: This line returns a dictionary with a single key-value pair, where the key is "message" and the value is "Welcome to news app!". FastAPI automatically serializes the dictionary into JSON format and sends it as the response.

To summarize, this code sets up a FastAPI application with a single route (/) that responds with a JSON message when accessed via a GET request. The message is a simple greeting, "Welcome to news app!".

Let’s run the app

uvicorn app.main:app --reload --workers 1 --host 0.0.0.0 --port 8000

The command uvicorn app.main:app refers to:

main: the file main.py (the Python "module") under the app folder.
app: the object created inside of main.py with the line app = FastAPI().
--reload: make the server restart after code changes. Only use for development.
--workers 1 : worker amount
--host : host ip address
--port : port number

Open your browser and navigate to http://0.0.0.0:8000/ address. You will see the json response message {“message”:”Welcome to news app!”}

4. Connecting to PostgreSQL Database

To read the environment variables from .env file I will use python-dotenv package which is already installed with uvicorn

Create database.py file under the app folder.

import os
from pathlib import Path
import dotenv

BASE_DIR = Path(__file__).resolve().parent.parent
dotenv.load_dotenv(BASE_DIR / ".env")

BASE_DIR is defined using the Path class from pathlib. It represents the base directory of the project by resolving the path of the current file (__file__) and going two levels up (parent.parent).
The dotenv.load_dotenv() function is called with the path to the .env file (BASE_DIR / ".env") as the argument. This function loads the environment variables from the .env file and makes them available in the application.

Make sure you have a .env file in the base directory of your project, and it contains the necessary environment variables. The dotenv.load_dotenv() function loads the environment variables from this file into the application's environment.

Let’s create a context manager base class to use in database operations. app/database.py

import os
from pathlib import Path
import dotenv
from abc import ABC, abstractmethod # new

BASE_DIR = Path(__file__).resolve().parent.parent
dotenv.load_dotenv(BASE_DIR / ".env")

class Database(ABC):
    """
    Database context manager
    """

    def __init__(self, driver) -> None:
        self.driver = driver

    @abstractmethod
    def connect_to_database(self):
        raise NotImplementedError()

    def __enter__(self):
        self.connection = self.connect_to_database()
        self.cursor = self.connection.cursor()
        return self

    def __exit__(self, exception_type, exc_val, traceback):
        self.cursor.close()
        self.connection.close()

The Database class is defined as a subclass of the ABC class, which stands for Abstract Base Class. It serves as a base class for other classes and can define abstract methods that must be implemented by its subclasses.
The __init__ method is the constructor of the Database class. It takes a driver parameter, which represents the database driver. The self.driver attribute is set to the provided driver value.
This connect_to_database method is marked as an abstract method using the @abstractmethod decorator. Subclasses of Database must implement this method, which should establish a connection to the specific database.
The __enter__ method is a special method used in Python's context manager protocol. It is invoked when entering the context (using the with statement). Here, it establishes a database connection by calling the connect_to_database method and assigns the connection to self.connection. It also creates a cursor object (self.cursor) to execute SQL queries.
The __enter__ method returns self, which allows you to use the instance of the Database class as a context manager within the with block.
The __exit__ method is another special method used in the context manager protocol. It is called when exiting the context (leaving the with block). Here, it closes the cursor (self.cursor) and the database connection (self.connection).
The __exit__ method receives three arguments: exception_type, exc_val, and traceback. These are used to handle any exceptions that occurred within the context. However, in our code, there is no explicit exception handling implemented.

To summarize, this code defines a Database class that acts as a context manager for working with a database. It enforces the implementation of a connect_to_database method in subclasses and provides the necessary setup and cleanup logic for a database connection and cursor.

Install psycopg2 package to work with PostgreSQL

pip install psycopg2

I assume that you installed postgreSQL and created a database named as t_news

Now let’s continue editing app/database.py

# previous imports
import psycopg2

# previous codes ...

class PgDatabase(Database):
    """PostgreSQL Database context manager"""

    def __init__(self) -> None:
        self.driver = psycopg2
        super().__init__(self.driver)

    def connect_to_database(self):
        return self.driver.connect(
            host=os.getenv("DB_HOST"),
            port=os.getenv("DB_PORT"),
            user=os.getenv("DB_USERNAME"),
            password=os.getenv("DB_PASSWORD"),
            database=os.getenv("DB_NAME")
        )

The PgDatabase class is defined as a subclass of the Database class. It specifically represents a PostgreSQL database context manager.
The __init__ method initializes the PgDatabase object. It sets the self.driver attribute to psycopg2, which is the PostgreSQL driver library. Then, it calls the __init__ method of the parent Database class using super().__init__(self.driver), passing the psycopg2 driver as an argument.
The connect_to_database method is implemented here to establish a connection to the PostgreSQL database using the psycopg2 driver. It retrieves the connection details from environment variables (os.getenv) such as DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD, and DB_NAME. These environment variables should be set with the appropriate database configuration.
The connect method of psycopg2 is called with the retrieved connection details to establish the connection. The method returns the connection object.
Let’s create the .env file and add the followings(Please change the values by your credentials):

export DB_HOST=localhost
export DB_PORT=5432
export DB_USERNAME=developer
export DB_PASSWORD=developer
export DB_NAME=db_news

To summarize, this code defines the PgDatabase class, a subclass of Database, specifically tailored for PostgreSQL database management. It sets up the PostgreSQL driver (psycopg2) and implements the connect_to_database method to establish a connection to the database using the driver and environment variables for the connection details.

Creating Tables

Add the following function and table name variable in the app/database.py


# previous codes

t_news = "t_news"

def create_tables():
    with PgDatabase() as db:
        db.cursor.execute(f"""CREATE TABLE {t_news} (
            id SERIAL PRIMARY KEY,
            published_date TIMESTAMPTZ,
            created_date TIMESTAMPTZ DEFAULT NOW(),
            created_by VARCHAR(140),
            context TEXT NOT NULL
            );
        """)
        db.connection.commit()
        print("Tables are created successfully...")

The create_tables function is defined to create the required table(s) in the database.
It uses a context manager by utilizing the with statement with an instance of the PgDatabase class (with PgDatabase() as db:). This ensures that the database connection is properly managed and automatically closed after executing the block of code.
Inside the with block, the execute method of the db.cursor object is called to execute a SQL query that creates a table. The table name is expected to be stored in the t_news variable (assuming it's defined in the code). The CREATE TABLE statement defines the table structure with columns such as id, published_date, created_date, created_by, and context.
After executing the CREATE TABLE statement, the changes are committed to the database using db.connection.commit(). This ensures that the table creation operation is persistent.
Finally, a success message is printed to the console, indicating that the table(s) were created successfully.

Dropping Tables

Let’s add another function in database.py to drop the tables if the table already exists. app/database.py


def drop_tables():
    with PgDatabase() as db:
        db.cursor.execute(f"DROP TABLE IF EXISTS {t_news} CASCADE;")
        db.connection.commit()
        print("Tables are dropped...")

Dropping and Creating Tables via Endpoint

Let’s create an endpoint to drop and create tables. Open the app/main.py and add the following codes.

from fastapi import FastAPI, status # new
from fastapi.exceptions import HTTPException # new
# internals
from app.database import drop_tables, create_tables # new

app = FastAPI()

# previous codes ...

@app.post('/initdb')
async def initdb():
    try:
        drop_tables()
        create_tables()
        return {"message": "Tables dropped and created!"}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Error {e}"
        )

The import statements are added to import the necessary modules for handling exceptions (fastapi.exceptions.HTTPException) and the drop_tables and create_tables functions from the app.database module.
The /initdb route is defined as an HTTP POST route using the @app.post decorator. It is responsible for dropping and creating the database tables.
Inside the route function, a try-except block is used to handle any potential exceptions that may occur during the dropping and creation of tables.
In the try block, the drop_tables function is called to drop the existing tables, and then the create_tables function is called to create new tables.
If the operations complete without any exceptions, a JSON response is returned with a success message: {"message": "Tables dropped and created!"}.
If an exception occurs during the process, the except block is triggered. A HTTPException is raised with a status code of 400 (Bad Request) and a detailed error message that includes the specific error (f"Error {e}").

Now let’s test the endpoint. Open your terminal to use curl or use postman, thunder client to send POST request to this endpoint.

curl -X POST http://0.0.0.0:8000/initdb

You will get the response message as

{"message":"Tables dropped and created!"}

In the next article we are going to build CRUD endpoints and database queries.