Open in app

Sign in

Write

Sign in

FastAPI CRUD app with Raw SQL | Part 2

Adnan Kaya
11 min readJul 7, 2023

--

In this article we will continue building CRUD app with raw SQL with FastAPI. In the Part 1 article we started the project and added 2 endpoints to our application. We also added a context manager to connect, run SQL queries. We used this context manager in our custom functions.

Table of Contents:

  1. Introduction to FastAPI and psycopg2
  2. Pros and Cons of Raw SQL
  3. Getting Started: Setting up the Project
  4. Connecting to PostgreSQL Database
  5. Building the CRUD Endpoints with Raw SQL
  6. Error Handling and Data Validation
  7. Testing the FastAPI CRUD App
  8. Conclusion

5. Building the CRUD Endpoints with Raw SQL

Creating Models & Schemas

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

from pydantic import BaseModel, Field


class NewsSchema(BaseModel):
    created_by: str = Field(..., min_length=3, max_length=140)
    context: str = Field(..., min_length=3, max_length=4096)
    published_date: str = Field(..., min_length=3, max_length=32)


class NewsDB(NewsSchema):
    id: int
  • The NewsSchema class is defined as a subclass of BaseModel from the Pydantic library. It represents the schema or structure of a news item.
  • Inside the NewsSchema class, there are three fields defined using the Field class from Pydantic:
  • created_by field is of type string (str) and has minimum and maximum length constraints of 3 and 140 characters, respectively.
  • context field is also of type string (str) and has minimum and maximum length constraints of 3 and 4096 characters, respectively.
  • published_date field is of type string (str) and has minimum and maximum length constraints of 3 and 32 characters, respectively.
  • The NewsDB class is defined as a subclass of NewsSchema. It represents the database model for a news item, including the additional id field.
  • The id field is of type integer (int) and represents the unique identifier for a news item in the database.

By defining these models, we can use them to validate incoming data, serialize/deserialize JSON, and enforce constraints on the structure and types of the data.

Writing SELECT query in the database.py file


# previous codes ...

def select_t_news():
    with PgDatabase() as db:
        db.cursor.execute(f"""SELECT id, created_by, context, published_date 
                         FROM {t_news};""")
        objects = [
            {
                "id": data[0],
                "created_by": data[1],
                "context":data[2],
                "published_date":str(data[3])
            }
            for data in db.cursor.fetchall()
        ]
    return objects
  • The select_t_news function is defined to retrieve data from the t_news table.
  • 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 selects data from the t_news table. The selected columns are id, created_by, context, and published_date.
  • After executing the SQL query, the fetchall method is called on db.cursor to retrieve all the selected rows as a list of tuples. Each tuple represents a row from the result set.
  • A list comprehension is used to iterate over the rows and create a list of dictionaries (objects). Each dictionary represents a news item, with keys "id", "created_by", "context", and "published_date". The values are extracted from the corresponding elements in the tuples. The "published_date" value is converted to a string using str().
  • Finally, the objects list is returned, containing the selected news items as dictionaries.

Writing SELECT by ID query in the database.py file

# previous codes ...

def select_t_news_by_id(id: int) -> dict:
    with PgDatabase() as db:
        db.cursor.execute(f"""
        SELECT id, created_by, context, published_date FROM {t_news}
        WHERE id={id};
                        """)
        data = db.cursor.fetchone()
        if data is None:
            return None

    return {
        "id": data[0],
        "created_by": data[1],
        "context": data[2],
        "published_date": str(data[3])
    }
  • The select_t_news_by_id function is defined to retrieve a specific news item from the t_news table based on the provided id.
  • The query includes a WHERE clause that filters the rows based on the id column matching the provided id.
  • After executing the SQL query, the fetchone method is called on db.cursor to retrieve the first row that matches the query. The result is stored in the data variable, which will be a tuple representing the selected row.
  • If data is None, it means no row was found with the provided id, so None is returned.
  • If data contains a valid row, a dictionary is created with keys "id", "created_by", "context", and "published_date". The values are extracted from the corresponding elements in the tuple. The "published_date" value is converted to a string using str().
  • Finally, the dictionary representing the selected news item is returned.

Writing INSERT query in the database.py file

# previous imports ...
# internals
from app.models import NewsDB, NewsSchema # new

# previous codes ...

def insert_t_news(payload: NewsSchema, *args, **kwargs) -> NewsDB:
    with PgDatabase() as db:
        db.cursor.execute(f"""
        INSERT INTO {t_news}(created_by, context, published_date) 
        VALUES('{payload.created_by}', 
                '{payload.context}', 
                '{payload.published_date}'
                ) 
        RETURNING id;
                    """)
        db.connection.commit()
        inserted_id = db.cursor.fetchone()[0]

        obj = select_t_news_by_id(inserted_id)
    return obj
  • Inside the with block, the execute method of the db.cursor object is called to execute a SQL query that inserts a new row into the t_news table. The query includes the VALUES clause with the values extracted from the payload object.
  • The RETURNING clause is used to retrieve the id of the newly inserted row.
  • After executing the SQL query, the changes are committed to the database using db.connection.commit() to make the insertion persistent.
  • The fetchone method is called on db.cursor to retrieve the id value of the inserted row. It is stored in the inserted_id variable.
  • The select_t_news_by_id function is called with the inserted_id to retrieve the inserted news item.
  • Finally, the inserted news item is returned.

Writing UPDATE query in the database.py file

# previous codes
def update_t_news_by_id(id: int, payload: NewsSchema):
    with PgDatabase() as db:
        db.cursor.execute(f"""
        UPDATE {t_news}
        SET created_by='{payload.created_by}', 
            context='{payload.context}', 
            published_date='{payload.published_date}'
        WHERE id='{id}'
        RETURNING id;
                        """)
        db.connection.commit()
        result = db.cursor.fetchone()
        if not result:
            return None
        updated_id = result[0]
        obj = select_t_news_by_id(updated_id)
    return obj

Writing DELETE query in the database.py file

# previous codes...
def delete_t_news_by_id(id: int):
    with PgDatabase() as db:
        db.cursor.execute(f"""
        DELETE FROM {t_news}
        WHERE id={id};
                        """)
        db.connection.commit()
        res = db.cursor.statusmessage
    if res == "DELETE 1":
        return True
    return False
  • The statusmessage attribute of the db.cursor object is checked to determine the result of the delete operation. If the status message is "DELETE 1", it means that one row was successfully deleted, so True is returned.
  • If the status message is not "DELETE 1", it means that no row was deleted or an error occurred, so False is returned.

Writing CRUD views

Create views.py file under the app folder

create_news

from fastapi import APIRouter, HTTPException, status
from psycopg2.errors import DatetimeFieldOverflow
#  internals
from app.database import (insert_t_news,)
from app.models import NewsDB, NewsSchema

router = APIRouter()


@router.post('/', response_model=NewsDB, status_code=status.HTTP_201_CREATED)
async def create_news(payload: NewsSchema):
    try:
        res = insert_t_news(payload)
        return res
    except DatetimeFieldOverflow:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Formats are : month-day-year hour:minute:seconds or year-month-day hour:minute:seconds"
        )
  • The APIRouter class from FastAPI is imported to create the router.
  • The HTTPException class from FastAPI is imported to handle exceptions.
  • The DatetimeFieldOverflow exception class from psycopg2.errors is imported. It specifically handles exceptions related to datetime field overflows.
  • The insert_t_news function from the app.database module is imported. It is used to insert a news item into the database.
  • The NewsDB and NewsSchema models from the app.models module are imported. They are used as the response model and payload model, respectively.
  • An instance of APIRouter is created and assigned to the router variable.
  • The @router.post decorator is used to define the POST route for creating a news item. It specifies the root path ("/") as the endpoint.
  • The response_model parameter is set to NewsDB, which represents the response model for this route. It ensures that the response will be serialized according to the NewsDB model.
  • The status_code parameter is set to status.HTTP_201_CREATED, which indicates that the response status code will be 201 Created.
  • The route function is defined with the name create_news. It takes a payload parameter of type NewsSchema, representing the data sent in the request body.
  • Inside the function, an attempt is made to insert the news item into the database by calling insert_t_news(payload). The result is stored in the res variable.
  • If the insertion is successful, the res variable is returned as the response.
  • If a DatetimeFieldOverflow exception occurs during the insertion, it means that the provided datetime value exceeds the field's capacity. In this case, a HTTPException is raised with a status code of 400 Bad Request and a detail message indicating the supported datetime formats.

The code sets up an API route to handle the creation of news items, with appropriate response models and error handling for datetime field overflows.

We need to include /news endpoints in the app/main.py like the following


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

app = FastAPI()

# previous codes...

# at the end of the file
app.include_router(views.router, prefix='/news', tags=['news'])

To test create_news make POST request to http://0.0.0.0:8000/news/ address with the body

{
  "created_by":"adnankaya",
  "context":"fastapi tutorial by adnankaya",
  "published_date":"2023-07-14 14:53:21"
}

The response will be

{
  "created_by": "adnankaya",
  "context": "fastapi tutorial by adnankaya",
  "published_date": "2023-07-14 14:53:21+03:00",
  "id": 1
}

read_news

from typing import List # new
from fastapi import APIRouter, HTTPException, status
from psycopg2.errors import DatetimeFieldOverflow, OperationalError # new
#  internals
from app.database import (insert_t_news, select_t_news) # new
from app.models import NewsDB, NewsSchema

# previous codes...

@router.get('/', response_model=List[NewsDB], status_code=status.HTTP_200_OK)
async def read_news():
    try:
        return select_t_news()
    except OperationalError:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="""Check if the database exists, connection is successful or tables exist. To create tables use '/initdb' endpoint"""
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"""Error {e}"""
        )
  • The @router.get decorator is used to define a GET route for retrieving news items. The root path ("/") is used as the endpoint.
  • The response_model parameter is set to List[NewsDB], indicating that the response will be a list of news items serialized according to the NewsDB model.
  • The status_code parameter is set to status.HTTP_200_OK, indicating that the response status code will be 200 OK.
  • The route function is defined with the name read_news and no parameters.
  • Inside the function, an attempt is made to retrieve all news items by calling the select_t_news function.
  • If the retrieval is successful, the retrieved news items are returned as the response.
  • If an OperationalError exception occurs during the retrieval, it indicates a server-side issue related to the database. In this case, a HTTPException is raised with a status code of 500 Internal Server Error and a detail message suggesting to check the database's existence, successful connection, or table existence. It also mentions the /initdb endpoint for creating tables.
  • If any other exception occurs during the retrieval, a generic HTTPException is raised with a status code of 400 Bad Request. The exception details are included in the response's detail message.

This code sets up a GET route to retrieve all news items, with appropriate response models and error handling for operational errors and other exceptions.

To test read_news make GET request to http://0.0.0.0:8000/news/ address

You will get the response like:

[
  {
    "created_by": "adnankaya",
    "context": "fastapi tutorial by adnankaya",
    "published_date": "2023-07-14 14:53:21+03:00",
    "id": 1
  },
  {
    "created_by": "adnankaya",
    "context": "django tutorial by adnankaya",
    "published_date": "2023-06-14 14:53:21+03:00",
    "id": 2
  },
  {
    "created_by": "adnankaya",
    "context": "flask tutorial by adnankaya",
    "published_date": "2022-11-22 14:53:21+03:00",
    "id": 3
  }
]

read_news_by_id

from typing import List
from fastapi import APIRouter, HTTPException, status, Path # new
from psycopg2.errors import DatetimeFieldOverflow, OperationalError
#  internals
from app.database import (insert_t_news, select_t_news, select_t_news_by_id) # new
from app.models import NewsDB, NewsSchema

@router.get('/{id}/', response_model=NewsDB, status_code=status.HTTP_200_OK)
async def read_news_by_id(id: int = Path(..., gt=0)):
    result = select_t_news_by_id(id)
    if not result:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail='News not found')
    return result
  • The @router.get decorator is used to define a GET route for retrieving a news item by its id. The /{id}/ path parameter is specified in the route's URL.
  • The response_model parameter is set to NewsDB, indicating that the response will be a news item serialized according to the NewsDB model.
  • The status_code parameter is set to status.HTTP_200_OK, indicating that the response status code will be 200 OK.
  • The route function is defined with the name read_news_by_id, and it takes an id parameter of type int. The Path class is used from fastapi.Path to define the parameter as a path parameter in the URL.
  • The id parameter is validated using ... and gt=0 in the Path constructor, which indicates that the id must be provided and greater than 0.
  • Inside the function, the select_t_news_by_id function is called to retrieve the news item based on the provided id. The result is stored in the result variable.
  • If the result is None, indicating that no news item was found with the provided id, a HTTPException is raised with a status code of 404 Not Found and a detail message indicating that the news item was not found.
  • If the result is not None, the news item is returned as the response.

This code sets up a GET route to retrieve a specific news item by its id, with appropriate response models and error handling for the case when the news item is not found.

To test read_news_by_id make GET request to http://0.0.0.0:8000/news/1 address

You will get the response like:

{
  "created_by": "adnankaya",
  "context": "fastapi tutorial by adnankaya",
  "published_date": "2023-07-14 14:53:21+03:00",
  "id": 1
}

update_news_by_id

#  internals
from app.database import (insert_t_news, select_t_news, select_t_news_by_id, 
                          update_t_news_by_id) # new


@router.put('/{id}/', response_model=NewsDB, status_code=status.HTTP_200_OK)
async def update_news_by_id(payload: NewsSchema, id: int = Path(..., gt=0)):
    result = update_t_news_by_id(id, payload)
    if not result:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail='News not found')
    return result
  • The @router.put decorator is used to define a PUT route for updating a news item by its id. The /{id}/ path parameter is specified in the route's URL.
  • The response_model parameter is set to NewsDB, indicating that the response will be a news item serialized according to the NewsDB model.
  • The status_code parameter is set to status.HTTP_200_OK, indicating that the response status code will be 200 OK.
  • The route function is defined with the name update_news_by_id. It takes a payload parameter of type NewsSchema representing the data sent in the request body, and an id parameter of type int representing the news item's id. Both parameters are validated using Path with ... and gt=0 to ensure they are provided and greater than 0.
  • Inside the function, the update_t_news_by_id function is called to update the news item in the database based on the provided id and payload. The result is stored in the result variable.
  • If the result is None, indicating that no news item was found with the provided id or the update was unsuccessful, a HTTPException is raised with a status code of 404 Not Found and a detail message indicating that the news item was not found.
  • If the result is not None, the updated news item is returned as the response.

This code sets up a PUT route to update a specific news item by its id, with appropriate response models and error handling for the case when the news item is not found or the update is unsuccessful.

To test the view make PUT request to http://0.0.0.0:8000/news/1 with the body

{
  "created_by": "adnankaya new",
  "context": "fastapi & raw SQL tutorial by adnankaya",
  "published_date": "2023-07-11 14:53:21+03:00"
}

The response

{
  "created_by": "adnankaya new",
  "context": "fastapi & raw SQL tutorial by adnankaya",
  "published_date": "2023-07-11 14:53:21+03:00",
  "id": 1
}

delete_news_by_id

from app.database import (insert_t_news, select_t_news, select_t_news_by_id, 
                          update_t_news_by_id, delete_t_news_by_id) # new

@router.delete('/{id}/', status_code=status.HTTP_204_NO_CONTENT)
async def delete_news_by_id(id: int = Path(..., gt=0)):
    result = delete_t_news_by_id(id)
    if not result:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail='News not found')
    return result
  • The @router.delete decorator is used to define a DELETE route for deleting a news item by its id. The /{id}/ path parameter is specified in the route's URL.
  • The status_code parameter is set to status.HTTP_204_NO_CONTENT, indicating that the response status code will be 204 No Content.
  • The route function is defined with the name delete_news_by_id, and it takes an id parameter of type int. The Path class is used to define the id parameter as a path parameter in the URL, and ... and gt=0 are used for validation, ensuring that the id must be provided and greater than 0.
  • Inside the function, the delete_t_news_by_id function is called to delete the news item from the database based on the provided id. The result is stored in the result variable.
  • If the result is False, indicating that no news item was found with the provided id or the deletion was unsuccessful, a HTTPException is raised with a status code of 404 Not Found and a detail message indicating that the news item was not found.
  • If the result is True, indicating a successful deletion, result is returned as the response. However, since the response status code is 204 No Content, the response body will be empty.

This code sets up a DELETE route to delete a specific news item by its id, with appropriate response status code and error handling for the case when the news item is not found or the deletion is unsuccessful.

Test the view by making DELETE request to http://0.0.0.0:8000/news/2 address. The response is empty and Status: 204 No Content

In the next article we are going to write automated tests…

Fastapi
Python
Crud
Sql
Rest Api

--

--

Adnan Kaya

Written by Adnan Kaya

19 Followers

Software Engineer #python #django #flask #fastapi . More ? Visit https://adnankaya.github.io/

Help

Status

About

Careers

Press

Blog

Privacy

Terms

Text to speech

Teams