FastAPI

#load tdm environment
module use /anvil/projects/tdm/opt/core/
module load tdm
module load python/seminar

#make a directory
mkdir tdm-fastapi-example
#go to the directory
cd tdm-fastapi-example

# import SQLAlchemy parts
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# database URL for SQLAlchemy
# open a file with the SQLite database in the same directory
SQLALCHEMY_DATABASE_URL = "sqlite:///./tdm-fastapi-example.db"

# create SQLAlchemy engine (home base for the database)
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
# by default, SQLite only accepts one thread
# but in FastAPI, using (def) functions can interact with the database for the same request, hence multiple threads
# so `False` for checking the same thread is needed

# instance of the SessionLocal class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# returns a class to create each database model/class
Base = declarative_base()

# import SQLAlchemy parts
from sqlalchemy import Boolean, Column, Date, Double, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

# import Base from the file database.py
from .database import Base

# create User class
class User(Base):
    # the name of the table
    __tablename__ = "users"

    # Class attributes with each attribute being a column in the table
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)

    # relationship is a "magic" attribute that conatains values from another table
    # expense table is related to this owner table
    expenses = relationship("Expense", back_populates="owner")

# create Expense class
class Expense(Base):
    # the name of the table
    __tablename__ = "expenses"

    # Class attributes
    id = Column(Integer, primary_key=True, index=True)
    category = Column(String, index=True)
    date = Column(Date, index=True)
    amount = Column(Double, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    # owner table is related to this expenses table
    owner = relationship("User", back_populates="expenses")

from pydantic import BaseModel
from datetime import date

class ExpenseBase(BaseModel):
    category: str
    date: date
    amount: float
    description: str | None = None
class ExpenseCreate(ExpenseBase):
    pass
class Expense(ExpenseBase):
    id: int
    owner_id: int
    class Config:
        orm_mode = True

class UserBase(BaseModel):
    name: str
class UserCreate(UserBase):
    pass
class User(UserBase):
    id: int
    expenses: list[Expense] = []
    class Config:
        orm_mode = True

from sqlalchemy.orm import Session

# import models (SQLAlchemy models) and schemas (Pydantic models/schemas) files
from . import models, schemas

# Utility Functions

# Read a single user by ID
def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id==user_id).first()

# read a single user by name
def get_user_by_name(db: Session, name: str):
    return db.query(models.User).filter(models.User.name==name).first()

# Read multiple users
def get_users(db: Session, skip: int=0, limit: int=10):
    return db.query(models.User).offset(skip).limit(limit).all()

# Read multiple expenses
def get_expenses(db: Session, skip: int=0, limit: int=10):
    return db.query(models.Expense).offset(skip).limit(limit).all()

# Create a SQLAlchemy model instance
def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(name=user.name)

    # add the instance to the database system
    db.add(db_user)
    # commit the changes to the database session to be saved
    db.commit()
    # refresh the instance to contain any new data from the database
    db.refresh(db_user)
    return db_user

def create_user_expense(db: Session, expense: schemas.ExpenseCreate, user_id: int):
    db_expense = models.Expense(**expense.dict(), owner_id=user_id)

    # add the instance to the database session
    db.add(db_expense)
    # commit the changes to the database session to be saved
    db.commit()
    # refresh the instance to contain any new data from the database
    db.refresh(db_expense)
    return db_expense

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency: use the same session for the request and then close it
def get_db():
    db=SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session=Depends(get_db)):
    db_user = crud.get_user_by_name(db, user.name)

    if db_user:
        raise HTTPException(status_code=400, detail="Name already registered")

    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int=0, limit: int=10, db: Session=Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session=Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.post("/users/{user_id}/expenses/", response_model=schemas.Expense)
def create_expense_for_user(user_id: int, expense: schemas.ExpenseCreate, db: Session=Depends(get_db)):
    return crud.create_user_expense(db=db, expense=expense, user_id=user_id)

@app.get("/expenses/", response_model=list[schemas.Expense])
def read_expenses(skip: int=0, limit: int=10, db: Session=Depends(get_db)):
    return crud.get_expenses(db, skip=skip, limit=limit)

python3 -m uvicorn tdm-fastapi-example.main:app --reload