158 lines
5.0 KiB
Python
158 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import models
|
|
from ..db import get_session
|
|
from ..schemas import (
|
|
HouseholdCreate,
|
|
HouseholdRead,
|
|
HouseholdUpdate,
|
|
LoginIdentityCreate,
|
|
LoginIdentityRead,
|
|
UserCreate,
|
|
UserRead,
|
|
UserUpdate,
|
|
)
|
|
|
|
router = APIRouter(tags=["households"])
|
|
|
|
|
|
@router.post("/households", response_model=HouseholdRead)
|
|
def create_household(
|
|
payload: HouseholdCreate, session: Session = Depends(get_session)
|
|
) -> models.Household:
|
|
household = models.Household(
|
|
name=payload.name, default_currency=payload.default_currency.upper()
|
|
)
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
return household
|
|
|
|
|
|
@router.get("/households", response_model=list[HouseholdRead])
|
|
def list_households(session: Session = Depends(get_session)) -> list[models.Household]:
|
|
return list(
|
|
session.scalars(select(models.Household).order_by(models.Household.created_at)).all()
|
|
)
|
|
|
|
|
|
@router.get("/households/{household_id}", response_model=HouseholdRead)
|
|
def get_household(household_id: str, session: Session = Depends(get_session)) -> models.Household:
|
|
household = session.get(models.Household, household_id)
|
|
if not household:
|
|
raise HTTPException(status_code=404, detail="Household not found")
|
|
return household
|
|
|
|
|
|
@router.patch("/households/{household_id}", response_model=HouseholdRead)
|
|
def update_household(
|
|
household_id: str,
|
|
payload: HouseholdUpdate,
|
|
session: Session = Depends(get_session),
|
|
) -> models.Household:
|
|
household = session.get(models.Household, household_id)
|
|
if not household:
|
|
raise HTTPException(status_code=404, detail="Household not found")
|
|
if payload.name is not None:
|
|
household.name = payload.name
|
|
if payload.default_currency is not None:
|
|
household.default_currency = payload.default_currency.upper()
|
|
session.add(household)
|
|
session.commit()
|
|
session.refresh(household)
|
|
return household
|
|
|
|
|
|
@router.delete("/households/{household_id}")
|
|
def delete_household(household_id: str, session: Session = Depends(get_session)) -> dict[str, bool]:
|
|
household = session.get(models.Household, household_id)
|
|
if not household:
|
|
raise HTTPException(status_code=404, detail="Household not found")
|
|
session.delete(household)
|
|
session.commit()
|
|
return {"deleted": True}
|
|
|
|
|
|
@router.post("/users", response_model=UserRead)
|
|
def create_user(payload: UserCreate, session: Session = Depends(get_session)) -> models.User:
|
|
if not session.get(models.Household, payload.household_id):
|
|
raise HTTPException(status_code=404, detail="Household not found")
|
|
user = models.User(
|
|
household_id=payload.household_id,
|
|
display_name=payload.display_name,
|
|
email=payload.email,
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.get("/users", response_model=list[UserRead])
|
|
def list_users(household_id: str, session: Session = Depends(get_session)) -> list[models.User]:
|
|
return list(
|
|
session.scalars(
|
|
select(models.User)
|
|
.where(models.User.household_id == household_id)
|
|
.order_by(models.User.display_name)
|
|
).all()
|
|
)
|
|
|
|
|
|
@router.patch("/users/{user_id}", response_model=UserRead)
|
|
def update_user(
|
|
user_id: str,
|
|
payload: UserUpdate,
|
|
session: Session = Depends(get_session),
|
|
) -> models.User:
|
|
user = session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(user, field, value)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.delete("/users/{user_id}")
|
|
def delete_user(user_id: str, session: Session = Depends(get_session)) -> dict[str, bool]:
|
|
user = session.get(models.User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
session.delete(user)
|
|
session.commit()
|
|
return {"deleted": True}
|
|
|
|
|
|
@router.post("/login-identities", response_model=LoginIdentityRead)
|
|
def create_login_identity(
|
|
payload: LoginIdentityCreate,
|
|
session: Session = Depends(get_session),
|
|
) -> models.LoginIdentity:
|
|
if not session.get(models.User, payload.user_id):
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
identity = models.LoginIdentity(**payload.model_dump())
|
|
session.add(identity)
|
|
session.commit()
|
|
session.refresh(identity)
|
|
return identity
|
|
|
|
|
|
@router.get("/login-identities", response_model=list[LoginIdentityRead])
|
|
def list_login_identities(
|
|
user_id: str, session: Session = Depends(get_session)
|
|
) -> list[models.LoginIdentity]:
|
|
return list(
|
|
session.scalars(
|
|
select(models.LoginIdentity)
|
|
.where(models.LoginIdentity.user_id == user_id)
|
|
.order_by(models.LoginIdentity.created_at)
|
|
).all()
|
|
)
|