Refactor database sessions into FastAPI dependency
This commit is contained in:
@@ -0,0 +1,17 @@
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from database import engine
|
||||
|
||||
|
||||
def get_session():
|
||||
with Session(engine, expire_on_commit=False) as session:
|
||||
yield session
|
||||
|
||||
|
||||
SessionDep = Annotated[
|
||||
Session,
|
||||
Depends(get_session),
|
||||
]
|
||||
@@ -2,11 +2,11 @@ from fastapi import FastAPI, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
from database import engine
|
||||
from models import Todo
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
from dependencies import SessionDep
|
||||
from models import Todo
|
||||
|
||||
app = FastAPI()
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
@@ -32,20 +32,20 @@ def health():
|
||||
return {"status":"ok"}
|
||||
|
||||
@app.get("/todos", response_model=list[todoresponse])
|
||||
def todos_get():
|
||||
with Session(engine) as session:
|
||||
def todos_get(session: SessionDep):
|
||||
statement = select(Todo)
|
||||
todos = session.scalars(statement).all()
|
||||
return todos
|
||||
|
||||
@app.post("/todos", response_model=todoresponse)
|
||||
def todos_create(todo_data:todocreate):
|
||||
with Session(engine) as session:
|
||||
def todos_create(
|
||||
todo_data: todocreate,
|
||||
session: SessionDep,
|
||||
):
|
||||
try:
|
||||
new_todo = Todo(task = todo_data.task)
|
||||
new_todo = Todo(task=todo_data.task)
|
||||
session.add(new_todo)
|
||||
session.commit()
|
||||
session.refresh(new_todo)
|
||||
return new_todo
|
||||
except SQLAlchemyError:
|
||||
session.rollback()
|
||||
@@ -53,8 +53,11 @@ def todos_create(todo_data:todocreate):
|
||||
|
||||
|
||||
@app.patch("/todos/{todo_id}", response_model=todoresponse)
|
||||
def todos_patch(todo_id:int, update_data:todoupdate):
|
||||
with Session(engine) as session:
|
||||
def todos_patch(
|
||||
todo_id: int,
|
||||
update_data: todoupdate,
|
||||
session: SessionDep,
|
||||
):
|
||||
try:
|
||||
todo = session.get(Todo, todo_id)
|
||||
if todo is None:
|
||||
@@ -64,7 +67,6 @@ def todos_patch(todo_id:int, update_data:todoupdate):
|
||||
if update_data.completion is not None:
|
||||
todo.completion = update_data.completion
|
||||
session.commit()
|
||||
session.refresh(todo)
|
||||
return todo
|
||||
except SQLAlchemyError:
|
||||
session.rollback()
|
||||
@@ -72,8 +74,10 @@ def todos_patch(todo_id:int, update_data:todoupdate):
|
||||
|
||||
|
||||
@app.delete("/todos/{todo_id}", response_model=todoresponse)
|
||||
def todos_delete(todo_id:int):
|
||||
with Session(engine) as session:
|
||||
def todos_delete(
|
||||
todo_id: int,
|
||||
session: SessionDep,
|
||||
):
|
||||
try:
|
||||
todo = session.get(Todo, todo_id)
|
||||
if todo is None:
|
||||
|
||||
Reference in New Issue
Block a user