FastAPI Simple CRUD - SqlAlchemy

FastAPI Simple CRUD - SqlAlchemy

Install Fastapi

https://github.com/tiangolo/fastapi

pip install fastapi

C:\fastAPI\myapp>pip install fastapi

C:\fastAPI\myapp>pip install "uvicorn[standard]"

Create main.py

crud/main.py

 
//crud/main.py
FastAPI Simple CRUD - SqlAlchemy


from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return "todo"

@app.post("/todo")
def create_todo():
    return "create todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "read todo item with id {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "update todo item with id {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "delete todo item with id {id}"

@app.get("/todo")
def read_todo_list():
    return "read todo list"

crud/main.py

 
//crud/main.py
from typing import List
from fastapi import FastAPI, status, HTTPException, Depends
from database import Base, engine, SessionLocal
from sqlalchemy.orm import Session
import models
import schemas

Base.metadata.create_all(engine) # Create the database

# Initialize app
app = FastAPI()

# Helper function to get database session
def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

@app.get("/")
def root():
    return "todo"

@app.post("/todo", response_model=schemas.ToDo, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDoCreate, session: Session = Depends(get_session)):

    tododb = models.ToDo(task = todo.task)

    session.add(tododb)
    session.commit()
    session.refresh(tododb)

    return tododb

@app.get("/todo/{id}", response_model=schemas.ToDo)
def read_todo(id: int, session: Session = Depends(get_session)):

    todo = session.query(models.ToDo).get(id) # get item with the given id

    # check if id exists. If not, return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

@app.put("/todo/{id}", response_model=schemas.ToDo)
def update_todo(id: int, task: str, session: Session = Depends(get_session)):

    todo = session.query(models.ToDo).get(id)     # get given id

    if todo:
        todo.task = task
        session.commit()

    # check if id exists. If not, return 404 not found response
    if not todo:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo

@app.delete("/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(id: int, session: Session = Depends(get_session)):

    # get the given id
    todo = session.query(models.ToDo).get(id)

    # if todo item with given id exists, delete it from the database. Otherwise raise 404 error
    if todo:
        session.delete(todo)
        session.commit()
    else:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return None

@app.get("/todo", response_model = List[schemas.ToDo])
def read_todo_list(session: Session = Depends(get_session)):

    todo_list = session.query(models.ToDo).all() # get all todo items

    return todo_list 	

Database sqlalchemy

https://www.sqlalchemy.org/

C:\fastAPI\crud_todo>pip install sqlalchemy
fastAPI\crud_todo\database.py

 
//fastAPI\crud_todo\database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Create a sqlite engine instance
engine = create_engine("sqlite:///fastapidb.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

# Create SessionLocal class from sessionmaker factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

fastAPI\crud_todo\models.py

 
//fastAPI\crud_todo\models.py
from sqlalchemy import Column, Integer, String
from database import Base

# Define ToDo class from Base
class ToDo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(256))

fastAPI\crud_todo\schemas.py

 
//fastAPI\crud_todo\schemas.py
from pydantic import BaseModel

# Create ToDo Schema (Pydantic Model)
class ToDoCreate(BaseModel):
    task: str

# Complete ToDo Schema (Pydantic Model)
class ToDo(BaseModel):
    id: int
    task: str

    class Config:
        orm_mode = True

run the FastAPI app

C:\fastAPI\crud_todo>uvicorn main:app --reload

with uvicorn using the file_name:app_instance
open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs