FastAPI CRUD app with Raw SQL | Part 2
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:
- Introduction to FastAPI and psycopg2
- Pros and Cons of Raw SQL
- Getting Started: Setting up the Project
- Connecting to PostgreSQL Database
- Building the CRUD Endpoints with Raw SQL
- Error Handling and Data Validation
- Testing the FastAPI CRUD App
- 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 ofBaseModel
from the Pydantic library. It represents the schema or structure of a news item. - Inside the
NewsSchema
class, there are three fields defined using theField
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 ofNewsSchema
. It represents the database model for a news item, including the additionalid
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 thet_news
table. - It uses a context manager by utilizing the
with
statement with an instance of thePgDatabase
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, theexecute
method of thedb.cursor
object is called to execute a SQL query that selects data from thet_news
table. The selected columns areid
,created_by
,context
, andpublished_date
. - After executing the SQL query, the
fetchall
method is called ondb.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 usingstr()
. - 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 thet_news
table based on the providedid
. - The query includes a
WHERE
clause that filters the rows based on theid
column matching the providedid
. - After executing the SQL query, the
fetchone
method is called ondb.cursor
to retrieve the first row that matches the query. The result is stored in thedata
variable, which will be a tuple representing the selected row. - If
data
isNone
, it means no row was found with the providedid
, soNone
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 usingstr()
. - 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, theexecute
method of thedb.cursor
object is called to execute a SQL query that inserts a new row into thet_news
table. The query includes theVALUES
clause with the values extracted from thepayload
object. - The
RETURNING
clause is used to retrieve theid
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 ondb.cursor
to retrieve theid
value of the inserted row. It is stored in theinserted_id
variable. - The
select_t_news_by_id
function is called with theinserted_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 thedb.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, soTrue
is returned. - If the status message is not
"DELETE 1"
, it means that no row was deleted or an error occurred, soFalse
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 frompsycopg2.errors
is imported. It specifically handles exceptions related to datetime field overflows. - The
insert_t_news
function from theapp.database
module is imported. It is used to insert a news item into the database. - The
NewsDB
andNewsSchema
models from theapp.models
module are imported. They are used as the response model and payload model, respectively. - An instance of
APIRouter
is created and assigned to therouter
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 toNewsDB
, which represents the response model for this route. It ensures that the response will be serialized according to theNewsDB
model. - The
status_code
parameter is set tostatus.HTTP_201_CREATED
, which indicates that the response status code will be201 Created
. - The route function is defined with the name
create_news
. It takes apayload
parameter of typeNewsSchema
, 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 theres
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, aHTTPException
is raised with a status code of400 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 toList[NewsDB]
, indicating that the response will be a list of news items serialized according to theNewsDB
model. - The
status_code
parameter is set tostatus.HTTP_200_OK
, indicating that the response status code will be200 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, aHTTPException
is raised with a status code of500 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 of400 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 itsid
. The/{id}/
path parameter is specified in the route's URL. - The
response_model
parameter is set toNewsDB
, indicating that the response will be a news item serialized according to theNewsDB
model. - The
status_code
parameter is set tostatus.HTTP_200_OK
, indicating that the response status code will be200 OK
. - The route function is defined with the name
read_news_by_id
, and it takes anid
parameter of typeint
. ThePath
class is used fromfastapi.Path
to define the parameter as a path parameter in the URL. - The
id
parameter is validated using...
andgt=0
in thePath
constructor, which indicates that theid
must be provided and greater than0
. - Inside the function, the
select_t_news_by_id
function is called to retrieve the news item based on the providedid
. The result is stored in theresult
variable. - If the
result
isNone
, indicating that no news item was found with the providedid
, aHTTPException
is raised with a status code of404 Not Found
and a detail message indicating that the news item was not found. - If the
result
is notNone
, 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 itsid
. The/{id}/
path parameter is specified in the route's URL. - The
response_model
parameter is set toNewsDB
, indicating that the response will be a news item serialized according to theNewsDB
model. - The
status_code
parameter is set tostatus.HTTP_200_OK
, indicating that the response status code will be200 OK
. - The route function is defined with the name
update_news_by_id
. It takes apayload
parameter of typeNewsSchema
representing the data sent in the request body, and anid
parameter of typeint
representing the news item'sid
. Both parameters are validated usingPath
with...
andgt=0
to ensure they are provided and greater than0
. - Inside the function, the
update_t_news_by_id
function is called to update the news item in the database based on the providedid
andpayload
. The result is stored in theresult
variable. - If the
result
isNone
, indicating that no news item was found with the providedid
or the update was unsuccessful, aHTTPException
is raised with a status code of404 Not Found
and a detail message indicating that the news item was not found. - If the
result
is notNone
, 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 itsid
. The/{id}/
path parameter is specified in the route's URL. - The
status_code
parameter is set tostatus.HTTP_204_NO_CONTENT
, indicating that the response status code will be204 No Content
. - The route function is defined with the name
delete_news_by_id
, and it takes anid
parameter of typeint
. ThePath
class is used to define theid
parameter as a path parameter in the URL, and...
andgt=0
are used for validation, ensuring that theid
must be provided and greater than0
. - Inside the function, the
delete_t_news_by_id
function is called to delete the news item from the database based on the providedid
. The result is stored in theresult
variable. - If the
result
isFalse
, indicating that no news item was found with the providedid
or the deletion was unsuccessful, aHTTPException
is raised with a status code of404 Not Found
and a detail message indicating that the news item was not found. - If the
result
isTrue
, indicating a successful deletion,result
is returned as the response. However, since the response status code is204 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…