How to use PostgreSQL to perform CRUD operations in FastAPI

In the previous tutorial, we saw how to perform CRUD operations using dummy data inside your project, but that can’t help us in real-world issues. Today I will take you through using the PostgreSQL database to perform CRUD operations in FastAPI RESTful services.

FastAPI really works well with both SQL and NoSQL databases, you can use SQL databases with the help of SQLAlchemy which is the Python SQL toolkit and Object Relational Mapper. It gives developers the full power and flexibility of SQL.

To start our project create a folder where you will install the dependencies and create a virtual environment. I will still be using the song API I created in the previous tutorial.

1. Create a virtual environment

It’s good to create a virtual environment so that you avoid installing programs globally. Create a virtual environment and activate

python3 -m venv env
source env/bin/activate

2. Install FastAPI

To start using FastAPI, we need to install it on our system with the following command.

pip install fastapi

You can pip freeze > requirements.txt to see all the dependencies

#requirements.txt
anyio==3.6.2
fastapi==0.88.0
idna==3.4
pydantic==1.10.2
sniffio==1.3.0
starlette==0.22.0
typing_extensions==4.4.0

Starlette is a lightweight ASGI framework/toolkit which is ideal for building async web services in Python. Startlette gives you the following and among many others:

• Websocket support
• In-process background tasks
• Test client built on httpx
• Session and cookie support
• CORS, GZip, static files, and Streaming responses

Pydantic is used for data validations and it uses Python-type annotations. It enforces type hints at runtime and provides user-friendly errors when data is invalid.

3. Install Uvicorn

Uvicorn is an ASGI web server for Python code implementation. You need ASGI for production. Install Uvicorn with the following command.

pip install "uvicorn[standard]"

The above command will install Uvicorn with Cython-based dependencies. Cython is used in managing event loops with uvloop which is a fast, drop-in replacement of the built-in asyncio event loop.

4. Testing the routes

The first thing we need to do is to import pydantic into our project schema.py file. Pydantic enables us to specify our models’ hints.

#schema.py 
from pydantic import BaseModel # import pydantic 

app = FastAPI

class song(BaseModel):
     title: str
     artist: str
     year: int
     description: str | None = None # None shows that the attribute is not necessary supplied. It can be blank.

As an example let’s create a request to see how pydantic simplifies our view.

#main.py
from fastapi import FastAPI
from songs import Song

app = FastAPI()


# Project root endpoint
@app.get('/')
async def root():
    return {'Message': 'Welcome to my Songs Library'}

# Create a song
@app.post('/songs') # endpoint
def create_Song(song:Song): # creating a schema
    return song

To run this code use uvicorn main:app--releoad.

Uvicorn is the ASGI server, Main is the root of a project main.py and reload is to make sure the server reloads every time we introduce changes.

5. Install Python psycopg2

psycopg2 is a PostgreSQL database adapter. So because we are using PostgreSQL here, we need to install psycopg2 into our environment. If you don’t have PostgreSQL installed already make sure you check it here. How to install PostgreSQL 15 on Debian 11.

pip install psycopg2 # needs a c compiler 
or 
pip install psycopg2-binary

6. Install SQLAlchemy

SQLAlchemy will handle ourdatabase connections with our PostgreSQL. It provides Object Relational Mapper (ORM) to enable us write our SQL queries.

pip install sqlalchemy

When installation is complete, go ahead and create a database.py file. i like separating everything so that I will be in a position to scale the application and also for simplicity purposes.

Inside the database.py paste the following code.

#database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# connection database url for SQLAlchemy
SQLALCHEMY_DATABASE_URL = "postgresql://fastapiuser:[email protected]/db"

# create SQLAlchemy engine 
engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)
# Create a sessionLocal class which acts as a database session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# create a decrative base that returns a class
Base = declarative_base()

7. Create SQLAlchemy models from base model

In the project folder create a models.py file. This is where we will place all the SQLAlchemy models class inheriting from the Base class we declared above. Add the following code to the models.py file.

from database import Base
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship


class Artist(Base):
    __tablename__ = 'artists'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    songs = relationship('Song', back_populates='artist')


class Song(Base):
    __tablename__ = 'songs'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    year = Column(Integer, index=True)
    description = Column(String, index=True)
    artist_id = Column(Integer, ForeignKey('artists.id'))

    artist = relationship('Artist', back_populates='songs')

    # represent an object as a string

    def __repr__(self):
        return f'<Song title = {self.title} year={self.year}>'

The __tablename__ attribute tells SQLAlchemy the name of the table to use in the database for each of these models.

We use = when representing SQLAlchemy models.

8. Create Pydantic Models

Pydantic models or schemas create attributes of the entire database structure, it will show how data will be created and read. In the schema.py we created earlier add add the following code.

# entire schema.py file
from pydantic import BaseModel

class Song(BaseModel):
    id: int
    title: str
    artist: str
    year: int
    description: str | None = None
    artist_id: int

    class Config: # provides configuration to pydantic
        orm_mode = True

class Artist(BaseModel):
    name: str

    class Config:
        orm_mode = True

Class Config provides configuration to pydantic. Not the difference between pydantic models and SQLAlchemy models. For pydantic we use : while SQLAlchemy we use =

Pydantic’s orm_mode will tell the Pydantic model to read the data even if it is not a dict, but an ORM model (or any other arbitrary object with attributes).

Test Database Connection

In order to test whether we can connect to our PostgreSQL database, we can create_db.py file. I am creating this file because I don’t want to use the python shell to test my connection. This is a better and easier way to do the test. Inside paste the following code.

#create_db.py
from database import Base, engine
from models import Song, Artist

Base.metadata.create_all(engine)

This is what you are going to see if you don’t run into errors if you run python create_db.py in your terminal

2022-12-01 20:48:36,735 INFO sqlalchemy.engine.Engine select pg_catalog.version()
2022-12-01 20:48:36,735 INFO sqlalchemy.engine.Engine [raw sql] {}
2022-12-01 20:48:36,736 INFO sqlalchemy.engine.Engine select current_schema()
2022-12-01 20:48:36,736 INFO sqlalchemy.engine.Engine [raw sql] {}
2022-12-01 20:48:36,737 INFO sqlalchemy.engine.Engine show standard_conforming_strings
2022-12-01 20:48:36,737 INFO sqlalchemy.engine.Engine [raw sql] {}
2022-12-01 20:48:36,737 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2022-12-01 20:48:36,738 INFO sqlalchemy.engine.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2022-12-01 20:48:36,738 INFO sqlalchemy.engine.Engine [generated in 0.00021s] {'name': 'artists'}
2022-12-01 20:48:36,740 INFO sqlalchemy.engine.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2022-12-01 20:48:36,740 INFO sqlalchemy.engine.Engine [cached since 0.001525s ago] {'name': 'songs'}
2022-12-01 20:48:36,741 INFO sqlalchemy.engine.Engine 
CREATE TABLE artists (
        id SERIAL NOT NULL, 
        name VARCHAR, 
        PRIMARY KEY (id)
)

2022-12-01 20:48:36,741 INFO sqlalchemy.engine.Engine [no key 0.00018s] {}
2022-12-01 20:48:36,758 INFO sqlalchemy.engine.Engine CREATE INDEX ix_artists_name ON artists (name)
2022-12-01 20:48:36,758 INFO sqlalchemy.engine.Engine [no key 0.00024s] {}
2022-12-01 20:48:36,763 INFO sqlalchemy.engine.Engine CREATE INDEX ix_artists_id ON artists (id)
2022-12-01 20:48:36,764 INFO sqlalchemy.engine.Engine [no key 0.00028s] {}
2022-12-01 20:48:36,769 INFO sqlalchemy.engine.Engine 
CREATE TABLE songs (
        id SERIAL NOT NULL, 
        title VARCHAR, 
        year INTEGER, 
        description VARCHAR, 
        artist_id INTEGER, 
        PRIMARY KEY (id), 
        FOREIGN KEY(artist_id) REFERENCES artists (id)
)

2022-12-01 20:48:36,769 INFO sqlalchemy.engine.Engine [no key 0.00018s] {}
2022-12-01 20:48:36,785 INFO sqlalchemy.engine.Engine CREATE INDEX ix_songs_year ON songs (year)
2022-12-01 20:48:36,786 INFO sqlalchemy.engine.Engine [no key 0.00020s] {}
2022-12-01 20:48:36,798 INFO sqlalchemy.engine.Engine CREATE INDEX ix_songs_id ON songs (id)
2022-12-01 20:48:36,798 INFO sqlalchemy.engine.Engine [no key 0.00016s] {}
2022-12-01 20:48:36,804 INFO sqlalchemy.engine.Engine CREATE INDEX ix_songs_title ON songs (title)
2022-12-01 20:48:36,804 INFO sqlalchemy.engine.Engine [no key 0.00020s] {}
2022-12-01 20:48:36,810 INFO sqlalchemy.engine.Engine CREATE INDEX ix_songs_description ON songs (description)
2022-12-01 20:48:36,810 INFO sqlalchemy.engine.Engine [no key 0.00035s] {}
2022-12-01 20:48:36,821 INFO sqlalchemy.engine.Engine COMMIT

9. Perform CRUD operations

After we have successfully connected our database, its now time to perform CRUD operations on our API. The operations we will perform are:

• Create songs/artist
• Read songs/artist from the database
• Update given song/artist
• Delete given song/artist

List all songs and Artists(GET)

To be able to list all the songs from the database use the following code

# Retrieve all songs
@app.get('/songs', response_model=List[Song], status_code=200)
def get_songs():
    songs = db.query(models.Song).all()
    return songs

# Retrieve all Artists
@app.get('/artists', response_model=List[Artist], status_code=200)
def get_all_artists():
    artists = db.query(models.Artist).all()
    return artists

Response_model is a parameter of the decorator method such as put,get,post,delete etc. response model is used to convert the output data to its type declaration and to validate data.

The above code when you run on swagger UI will give an empty list because the database is still empty.

List a single song or artist(GET)

If you want to retrieve a single song from the database, use the following code:

# Retrieve a single song
@app.get('/song/{song_id}', response_model=Song, status_code=status.HTTP_200_OK)
def get_one_song(song_id:int):
    song = db.query(models.Song).filter(models.Song.id==song_id).first()
    return song

And for retrieving a single artist, it the same with that of the song. Just change models song to artist.

Create a new song and artist(POST)

Lets start by creating new artist first so that we can not violate the foreign key constrain. Because songs belong to different artists or a single artist. The most important thing with create is the add and commit. Add will add the newly created data to the database but it doesn’t save until you commit the changes.

# Add an Artist
@app.post('/Add_artist', response_model=Artist, status_code=status.HTTP_201_CREATED)
def create_an_artist(artist:Artist):
   new_artist = models.Artist(name=artist.name)

   db.add(new_artist)
   db.commit()
   return new_artist

For adding a new song use the following code, both codes are the same its only the attributes which are different.

# Create a song
@app.post('/create_song', response_model=Song, status_code=status.HTTP_201_CREATED)
def create_a_song(song:Song):
   new_song = models.Song(title=song.title, year=song.year, description=song.description, artist_id=song.artist_id)
   
   # add new song to the db using SessionLocal class
   db.add(new_song)

   # save the new song to the database
   db.commit()
   return new_song

Updating a song or artist(PUT)

To update data, we need to get the id of the song to update and then invoke the db update method then use the commit function to save the changes to the database. Check the code below.

# Update a song
@app.put('/update_song/song_id', response_model=Song, status_code=status.HTTP_200_OK)
def update_a_song(song_id: int, song: Song):
    song_to_be_updated = db.query(models.Song).filter(
        models.Song.id == song_id).first()
    song_to_be_updated.title = song.title  # for updating the title
    song_to_be_updated.year = song.year
    song_to_be_updated.description = song.description
    song_to_be_updated.artist_id = song.artist_id

    db.commit()  # save changes

    return song_to_be_updated

For updating the artist, just change the song’s attributes

Deleting songs or artist(DELETE)

Invoking delete will remove an item from the database. Look at the following code

# delete a song from database
@app.delete('/song/{song_id}')
def delete_a_song(song_id: int):
    song_to_be_deleted = db.query(models.Song).filter(
        models.Song.id == song_id).first()
# raise an exception if the song is not found
    if song_to_be_deleted is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail='That song is not found in the databse')
    db.delete(song_to_be_deleted)  # delete a song instance
    db.commit()  # save to the databse
    return {'message': 'The song has been deleted successfully'}

For delete artist just change the query model to artist.

Conclusion

We have successfully implemented CRUD operations using PostgreSQL, SQLAlchemy on a fastAPI application. I hope you have learned something new today. Get the code from github repository.