API Request Methods - Get Todos from Database
API Request Methods - Get Todos from Database
Fetching all records:
Fetch Todos and dependency inversion.
code:
# This file it to perform all the activity which we required for Todo application.
# Depends means dependency injection
from fastapi import FastAPI, Depends
import models
from models import Todos
from database import engine, SessionLocal
from typing import Annotated
from sqlalchemy.orm import Session
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("/")
#async def read_all(db: Annotated[Session, Depends(get_db)]):
async def read_all(db: db_dependency):
return db.query(Todos).all()
Fetching only one record:
Code:# 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
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")
Result:
Fetched specific record:
Comments
Post a Comment