Code Scalability and Routing
Creating Authentication file
As we need to keep authentication part separate from our main application working so that we will not mess-up with our main application logic and it's security settings and authentication criteria as well as logics.
auth.py - Basic authentication code stub.
# This file is to define authenticator and authorization for the Todo application.
from fastapi import FastAPI,status
app = FastAPI()
@app.get("/auth", status_code=status.HTTP_200_OK)
async def get_user():
return {'user': 'authenticated_user'}
Create Routers Package
Here we want to use both and auth.py and main.py file to run together and to allow routers to run smoothly and want to scale the project code. To make this possible we need to use router facility or FastAPI.
Here we are going to have a main file is a route of our application and routes will sit on this main file and will help in performing the tasks on call from main file.
To make it possible we need to module our main program files and need to have a router folder which will have all relevant router files for different tasks for relevant application.
We need to inherit APIRouter to allow main.py file to route to auth.py file for task accomplishment.
New auth.py and main.py file content and package location and folder structure.
Folder structure:
auth.py file content:
# This file is to define authenticator and authorization for the Todo application.
from fastapi import APIRouter,status
router = APIRouter()
@router.get("/auth", status_code=status.HTTP_200_OK)
async def get_user():
return {'user': 'authenticated_user'}
main.py file content:
# 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
from routers import auth
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)
app.include_router(auth.router)
# 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()
@app.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(db: db_dependency,
todo_request: TodoRequest,
todo_id: int = Path(gt=0)
):
todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
if todo_model is None:
raise HTTPException(status_code=404, detail='Id not found in Todo table')
# todo_model must be the same which we are creating above in this method
# As sqlalchemy will use it to update the table.
# As if we are using the same model then it will think it as a new record
# will try to insert it with new id as autoincremented one.
# or might create an identical record with same id.
todo_model.title = todo_request.title
todo_model.description = todo_request.description
todo_model.priority = todo_request.priority
todo_model.complete = todo_request.complete
# As a result we are modifying the todo_model value before adding to todo table.
db.add(todo_model)
db.commit()
@app.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(db: db_dependency, todo_id: int = Path(gt=0)):
todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
if todo_model is None:
raise HTTPException(status_code=404, detail="Id not found in Todo table")
db.query(Todos).filter(Todos.id == todo_id).delete()
db.commit()
Enhancing Application for Scalability by using router
todos.py file content:
# This file will contain todos router details for main file to execute on call.
from fastapi import APIRouter, Depends, HTTPException, Path
from models import Todos
from database import SessionLocal
from typing import Annotated
from sqlalchemy.orm import Session
from starlette import status
from TodoValidator import TodoRequest
router = APIRouter()
# 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.
@router.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()
@router.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")
@router.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()
@router.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(db: db_dependency,
todo_request: TodoRequest,
todo_id: int = Path(gt=0)
):
todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
if todo_model is None:
raise HTTPException(status_code=404, detail='Id not found in Todo table')
# todo_model must be the same which we are creating above in this method
# As sqlalchemy will use it to update the table.
# As if we are using the same model then it will think it as a new record
# will try to insert it with new id as autoincremented one.
# or might create an identical record with same id.
todo_model.title = todo_request.title
todo_model.description = todo_request.description
todo_model.priority = todo_request.priority
todo_model.complete = todo_request.complete
# As a result we are modifying the todo_model value before adding to todo table.
db.add(todo_model)
db.commit()
@router.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(db: db_dependency, todo_id: int = Path(gt=0)):
todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
if todo_model is None:
raise HTTPException(status_code=404, detail="Id not found in Todo table")
db.query(Todos).filter(Todos.id == todo_id).delete()
db.commit()
main.py content:
# This file it to perform all the activity which we required for Todo application.
# Depends means dependency injection
from fastapi import FastAPI
import models
from database import engine
from routers import auth, todos
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)
app.include_router(auth.router)
app.include_router(todos.router)
Result:
All tasks in new application refactoring arrangement is still running smoothly and we have even achieved our goal of scalability.
Comments
Post a Comment