Wish-List-With-Fastapi
May 6, 2022 - ⧖ 6 minAmazon has crossed the line with me just one too many times now so we are looking to drop them like every other Big Tech provider....
However, one key feature of Amazon that has been so useful for us is Lists... We can just maintain a list for each of us and then family members can login anytime and check it out... This really alleviates any last minute gift idea stress right before a birthday or something.
So I need a nice gift list service but I don't want to be locked into one company (like a Target registry or something) and I'd like to host it myself
The internets had a few options but nothing looked/felt like I wanted to I decided to build my own.
The Frontend
I have no idea how to do front end so stay tuned
The Backend
FastAPI for the win on this one... I followed a few examples online and what I was able to build in just a few minutes is pretty impressive thanks to the design of FastAPI.
Some key features are:
- Auto doc generation
- Required typing (which makes #1 possible)
- Built-in api testing in the browser
- Easy integration with sqlalchemy
- Development time so short you won't be done with your coffee before having something up and running!
Database
Starting with a simple database.py
we can create a sqlalchemy session with a base model with about 7 lines of code...
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///wishes.sqlite3"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Model
For my wish list I needed just a simple table:
id | person | item | link | purchased | purchased_by | date_added |
---|---|---|---|---|---|---|
1 | pypeaday | A sweet item | www.mystore.store | False | dad | 2022-05-05 21:55:09 |
2 | pypeaday | A bitter item | www.bitterstore.com | True | Mrs. pypeaday | 2022-05-06 06:55:54 |
The table is simple enough... A unique key, the person who the wish belongs to, the item (or wish), a link to the item, whether it's been purchased or not and by whom, and the date it was added.
To make this model with sqlalchemy we can make a model.py
like so:
from database import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Boolean, Integer, String, Text
class Wishes(Base):
__tablename__ = "Wishes"
id = Column(Integer, primary_key=True, index=True)
person = Column(String(20))
item = Column(Text())
link = Column(Text())
purchased = Column(Boolean())
purchased_by = Column(String(90))
date_added = Column(String(15))
Schema
One of the best things about FastAPI is trivial integration with pydantic. We can define a schema to ensure any data posted is not missing anything!
Make a schema.py
with the following:
from pydantic import BaseModel
import time
from typing import Optional
class wish_schema(BaseModel):
person: str
item: str
link: str
purchased: bool = False
purchased_by: Optional[str] = None
date_added: Optional[str] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
class Config:
orm_mode = True
class patch_schema(BaseModel):
purchased: bool
purchased_by: Optional[str] = None
class Config:
orm_mode = True
I have 2 schemas - one for a wish
which you'll see down below is used to validate any post
requests.
To simplify things for me I made another schema, patch_schema
which I use for the route that updates the table (ie. marking an existing wish as purchased)
Session
One of the last things we need is a Session
So make a session.py
...
from database import SessionLocal, engine
import model
model.Base.metadata.create_all(bind=engine)
def create_get_session():
try:
db = SessionLocal()
yield db
finally:
db.close()
Our routes will depend on this create_get_session
function that will yield a db
object through which we'll udpate our database
Ok just do it already!
So our main.py
will have a few routes in it...
What do we want to support?
- Getting all wishes
- Getting a specific wish
- Updating a specific wish
- Deleting a wish
I think the script is fairly self explanatory but here's a few notes...
- We decorate each function with
@app.<method>
and defineresponse_model
as well asstatus_code
- The functions are defined with
async
(this was my first exposure to this so I can't go in depth on it yet) - The functions all take a
db
which is fromsession.py
and thatdb
depends on thecreate_get_session
function - If the db is being updtes then we type the object used for the update with the appropriate schema (either
wish_schema
orpatch_schema
)
From there we're in true python-land where you can basically guess the methods on db
and you'd probably be right... (like query
, upddate
, delete
etc.)
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from model import Wishes
from schema import wish_schema, patch_schema
from session import create_get_session
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "server is up!"}
@app.get("/wishes", response_model=List[wish_schema], status_code=200)
async def read_wishes(db: Session = Depends(create_get_session)):
wishes = db.query(Wishes).all()
return wishes
@app.post("/wishes", response_model=wish_schema, status_code=201)
async def add_wish(wish: wish_schema, db: Session = Depends(create_get_session)):
new_wish = Wishes(
person=wish.person,
item=wish.item,
link=wish.link,
purchased=wish.purchased,
purchased_by=wish.purchased_by,
date_added=wish.date_added,
)
db.add(new_wish)
db.commit()
return new_wish
@app.get("/wishes/{id}", response_model=wish_schema, status_code=200)
async def get_wish(id: int, db: Session = Depends(create_get_session)):
wish = db.query(Wishes).get(id)
return wish
@app.patch("/wishes/{id}", response_model=wish_schema, status_code=200)
async def update_wish(
id: int, patch: patch_schema, db: Session = Depends(create_get_session)
):
db_wish = db.query(Wishes).get(id)
db_wish.purchased = patch.purchased
db_wish.purchased_by = patch.purchased_by
db.commit()
db.refresh(db_wish)
return db_wish
@app.delete("/wishes/{id}", status_code=200)
async def delete_wish(id: int, db: Session = Depends(create_get_session)):
db_wish = db.query(Wishes).get(id)
if not db_wish:
raise HTTPException(status_code="404", detail="Wish id does not exist")
db.delete(db_wish)
db.commit()
return None
My Code
You can find my repo here.
I'll plan to update and maintain for as long as I use it