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

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

//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

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83

//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

?

1 2 3 4 5 6 7 8 9 10 11 12 13

//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

?

1 2 3 4 5 6 7 8 9

//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

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14

//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