CRUD APIs
When you boil down what any web application does, it can be said that most of the time it is performing either Create, Read, Update and Delete (CRUD) operations.
It's always best to begin with the Create function, to get an initial record added. Notice this function uses the POST method.
Create
Create a new file that will house the APIs for the blog part of the project. Here starting with adding the API for creating new blog posts.
from fastapi import Depends, status, APIRouter
from sqlmodel import Session
from db.schemas.blog_schemas import BlogPostIn, BlogPost
from db.db import get_session_db
router = APIRouter()
@router.post('/blog/create', status_code=status.HTTP_201_CREATED)
async def create_blogpost_api(blogpost_in: BlogPostIn, session: Session = Depends(get_session_db)) -> BlogPost:
new_blogpost = BlogPost.from_orm(blogpost_in)
session.add(new_blogpost)
session.commit()
session.refresh(new_blogpost)
return new_blogpost
Stepping through the create_blogpost_api function, it's quite self-explanatory.
You define the function and decorate it with @router.post, provide a context path, for example /blog/create and the expected HTTP response 201.
The function expects data input that conforms to the BlogPostIn schema, which is where the validations come into play. Moreover, a session to the database is included that will conform the theBlogPostschema. The rest is simple using, the ORM a new record is created and added to the database, and the API with return a JSON response of the record.
This will need plumbing into the main application, for example add the importfrom api import health_api, blog_apisand include the router:
def configure_routing():
main_app.include_router(health_api.router, prefix=settings.CONF_API_V1_STR, tags=["health"])
main_app.include_router(blog_apis.router, prefix=settings.CONF_API_V1_STR, tags=["blog_apis"]) # NEW
The example comes from the schema_extra defined in db/schemas/blog_schemas.py,
so adding a few samples is easy. Try out the API via Swagger to add a new record in the database.
For SQLite, open the database file:
To view records in the blogpost table:
To quit:
Read
With a database record or two added in the blogpost table, you can now add an
API to return all records. This function also accepts an integer to limit to number of results, remember to
import select from sqlmodel. Moreover, the results will be ordered by the newest first, and records need to have published == True to be included in the results:
from sqlmodel import Session, select
@router.get('/blog', status_code=status.HTTP_200_OK)
async def get_blogposts_api(limit: int | None = None, session: Session = Depends(get_session_db)) -> list:
query = select(BlogPost).where(BlogPost.published == True).limit(limit).order_by(BlogPost.created_date.desc())
return session.exec(query).all()
Additionally, another function is needed which includes a path parameter for selecting specific records by
its id.
@router.get('/blog/{id}', response_model=BlogPostOut, status_code=status.HTTP_200_OK)
async def get_blogpost_by_id_api(id: UUID, session: Session = Depends(get_session_db)) -> BlogPost:
post = session.get(BlogPost, id)
if post:
return post
else:
raise HTTPException(status_code=404, detail=f"No blog post found with id {id}.")
You'll need to import the BlogPostOut schema, UUID, and HTTPException. Here is the complete version of api/blog_apis.py at this point:
from fastapi import Depends, status, APIRouter, HTTPException
from pydantic.types import UUID
from sqlmodel import Session, select
from db.schemas.blog_schemas import BlogPostIn, BlogPostOut, BlogPost
from db.db import get_session_db
router = APIRouter()
@router.post('/blog/create', status_code=status.HTTP_201_CREATED)
async def create_blogpost_api(blogpost_in: BlogPostIn, session: Session = Depends(get_session_db)) -> BlogPost:
new_blogpost = BlogPost.from_orm(blogpost_in)
session.add(new_blogpost)
session.commit()
session.refresh(new_blogpost)
return new_blogpost
@router.get('/blog', status_code=status.HTTP_200_OK)
async def get_blogposts_api(limit: int | None = None, session: Session = Depends(get_session_db)) -> list:
query = select(BlogPost).limit(limit).order_by(BlogPost.created_date.desc())
return session.exec(query).all()
@router.get('/blog/{id}', response_model=BlogPostOut, status_code=status.HTTP_200_OK)
async def get_blogpost_by_id_api(id: UUID, session: Session = Depends(get_session_db)) -> BlogPost:
post = session.get(BlogPost, id)
if post:
return post
else:
raise HTTPException(status_code=404, detail=f"No blog post found with id {id}.")
Depending on whether your using TLS or not, via https://localhost/docs or http://localhost:8000/docs you should now be able to added sample records, retrieve them, and using an id from the returned results, retrieve a specific record.
Update
FastAPI recommends to use the PUT method for updating records. This function will use theidas PATH parameter.
@router.put('/blog/{id}', response_model=BlogPost, status_code=status.HTTP_202_ACCEPTED)
async def update_blogpost_api(id: UUID, data: BlogPostIn, session: Session = Depends(get_session_db)) -> BlogPost:
post = session.get(BlogPost, id)
if post:
post.title = data.title
post.teaser = data.teaser
post.content = data.content
post.cover_image = data.cover_image
session.commit()
return post
else:
raise HTTPException(status_code=404, detail=f"No blog post found with id {id}.")
Once again, you can get an id using the "Get Blogposts Api":
And test updating a record via http://localhost:8000/docs.
Delete
To complete the CRUD operations, this final function is to delete a record by id:
@router.delete('/blog/{id}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_blogpost_by_id_api(id: UUID, session: Session = Depends(get_session_db)) -> None:
post = session.get(BlogPost, id)
if post:
session.delete(post)
session.commit()
else:
raise HTTPException(status_code=404, detail=f"No blog post found with id {id}.")
You will see how these relatively simple APIs will provide the core operations for almost any application you build.