API Request Method  - Post Todos (create and save todo)

Code to make is properly:

TodoValidator.py

# Valdator class for Todos table entry for ensuring right data entry.

from pydantic import BaseModel, Field

# We have added id as part of validation

# As it will be done by table and fastAPI by itself

# As id is primary key and have incremental nature.

# Even sqlalchemy will help us in doing it automatically.

class TodoRequest(BaseModel):

    title: str = Field(min_length=3)

    description: str = Field(min_length=3, max_length=100)

    priority: int = Field(gt=0, lt=6)

    complete: bool

main.py:

# This file it to perform all the activity which we required for Todo application.

# Depends means dependency injection

from fastapi import FastAPI, Depends, HTTPException, Path

import models

from models import Todos

from database import engine, SessionLocal

from typing import Annotated

from sqlalchemy.orm import Session

from starlette import status

from TodoValidator import TodoRequest

app = FastAPI()

# it will execute only when todos.db is not existing.

# side effect if you enhance the table in model file then it will not update

# it in database automatically.

models.Base.metadata.create_all(bind=engine)

 

# creating db dependency

def get_db():

    # We need to fetch this SessionLocal before each request made to database.

    db = SessionLocal()

    try:

        # Only the code prio to and including the

        # yield statement will execute before sending the response.

        yield db

    finally:

        # It will be executed after sending the response.

        # It makes fastAPI quicker as we can fetch information

        # from the database and return it to client

        # and then can close the db connection.

        # It makes the connection extreamly safe.

        # As it will open the database connect only at the time of usage.

        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

# Creating asynchronous API Endpoint for database interaction.

@app.get("/", status_code=status.HTTP_200_OK)

#async def read_all(db: Annotated[Session, Depends(get_db)]):

async def read_all(db: db_dependency):

    return db.query(Todos).all()

@app.get("/todo/{todo_id}", status_code=status.HTTP_200_OK)

async def read_todo_by_id(db: db_dependency, todo_id: int = Path(gt=0)):

    todo_model= db.query(Todos).filter(Todos.id == todo_id).first()

    if todo_model is not None:

        return todo_model

    raise HTTPException(status_code=404, detail="id not found in table Todo")

@app.post("/todo",status_code=status.HTTP_201_CREATED)

async def create_todo(db: db_dependency, todo_request: TodoRequest):

    todo_model = Todos(**todo_request.model_dump())

    db.add(todo_model)

    db.commit()

Results: