topic15.py

# CRUD with Database (CREATE - Insert Data)
# . What is CREATE?
# . Flow of CREATE API
# . Testing with Swagger UI & DB
# . Interview Questions



from fastapi import Depends, FastAPI
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker
from starlette.exceptions import HTTPException

from topic6 import todos

app = FastAPI()  # object

DATABASE_URL = "sqlite:///./test.db"

# Engine create (DB connection)
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

# Session (DB operations ke liye)
sessionlocal = sessionmaker(bind=engine)

# Base(model ke liye)
Base = declarative_base()


# Table (Model)
class Todo(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    completed = Column(String)


# Table Create
Base.metadata.create_all(bind=engine)


# Dependency (DB session provide karega)
def get_db():
    db = sessionlocal()
    try:
        yield db
    finally:
        db.close()

# Creating a todos API (post)
@app.post("/todos")
def home(title: str, db: Session = Depends(get_db)):
    todo = Todo(title=title, completed="False")
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return {"message": "Todo Created", "data": todo}

# Reads all Data
@app.get("/todos")
def get_todos(db: Session = Depends(get_db)):
    todos = db.query(Todo).all()
    return {
        "Total":len(todos),
        "data": todos
    }

@app.get("/todos/{todo_id}")
def get_todo(todo_id = int, db:Session=Depends(get_db)):
    todo = db.query(Todo).filter(Todo.id == todo_id).first() #TODO is a model class, where we are getting its already written query

    if not todo:
        raise HTTPException(status_code=404, detail="Todo Not Found")
    return todo


# Update
@app.put("/todos/{todo_id}")
def update_todo(todo_id:int, title:str, db:Session = Depends(get_db)):
    todo = db.query(Todo).filter(Todo.id == todo_id).first()

    if not todo:
        raise HTTPException(status_code=404, detail="Todo Not Found")

    todo.title = title
    db.refresh(todo)

    db.commit()
    return{
        "message":"Todo Updated",
        "data":todo
    }


# Delete - Remove Data using API
@app.delete("/todos/{todo_id}")
def delete_todo(todo_id:int, db: Session=Depends(get_db)):
    todo = db.query(Todo).filter(Todo.id == todo_id).first()

    if not todo:
        raise HTTPException(status_code=404, detail="Todo Not Found")

    db.delete(todo)
    db.commit()

    return{
        "message":"Todo Deleted Successfully"
    }