Logging in FastAPI applications

Add logging to your FastAPI endpoints

Want more?

This lesson for enrolled students only. Join the course to unlock it!

You can see the code changes implemented in this lecture below.

If you have purchased the course in a different platform, you still have access to the code changes per lecture here on Teclado. The lecture video and lecture notes remain locked.
Join course for $30

Modified files

storeapi/routers/post.py
--- 
+++ 
@@ -1,3 +1,5 @@
+import logging
+
 from fastapi import APIRouter, HTTPException
 from storeapi.database import comment_table, database, post_table
 from storeapi.models.post import (
@@ -10,48 +12,81 @@

 router = APIRouter()

+logger = logging.getLogger(__name__)
+

 async def find_post(post_id: int):
+    logger.info(f"Finding post with id {post_id}")
+
     query = post_table.select().where(post_table.c.id == post_id)
+
+    logger.debug(query)
+
     return await database.fetch_one(query)


 @router.post("/post", response_model=UserPost, status_code=201)
 async def create_post(post: UserPostIn):
+    logger.info("Creating post")
+
     data = post.model_dump()  # previously .dict()
     query = post_table.insert().values(data)
+
+    logger.debug(query)
+
     last_record_id = await database.execute(query)
     return {**data, "id": last_record_id}


 @router.get("/post", response_model=list[UserPost])
 async def get_all_posts():
+    logger.info("Getting all posts")
+
     query = post_table.select()
+
+    logger.debug(query)
+
     return await database.fetch_all(query)


 @router.post("/comment", response_model=Comment, status_code=201)
 async def create_comment(comment: CommentIn):
+    logger.info("Creating comment")
+
     post = await find_post(comment.post_id)
+
     if not post:
+        logger.error(f"Post with id {comment.post_id} not found")
         raise HTTPException(status_code=404, detail="Post not found")

     data = comment.model_dump()  # previously .dict()
     query = comment_table.insert().values(data)
+
+    logger.debug(query)
+
     last_record_id = await database.execute(query)
     return {**data, "id": last_record_id}


 @router.get("/post/{post_id}/comment", response_model=list[Comment])
 async def get_comments_on_post(post_id: int):
+    logger.info("Getting comments on post")
+
     query = comment_table.select().where(comment_table.c.post_id == post_id)
+
+    logger.debug(query)
+
     return await database.fetch_all(query)


 @router.get("/post/{post_id}", response_model=UserPostWithComments)
 async def get_post_with_comments(post_id: int):
+    logger.info("Getting post and its comments")
+
     post = await find_post(post_id)
+
     if not post:
+        logger.error(f"Post with post id {post_id} not found")
         raise HTTPException(status_code=404, detail="Post not found")

     return {