288 lines
9.8 KiB
Python
288 lines
9.8 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, status
|
|
from sqlalchemy import func, select
|
|
|
|
from app.deps import DbSession, VerifiedUser
|
|
from app.models import Category, ListMember, Market, ShoppingList, User
|
|
from app.permissions import OwnedList, ReadableList
|
|
from app.schemas import MessageOut
|
|
from app.list_view import build_view
|
|
from app.schemas_shopping import (
|
|
CategoryOut,
|
|
ListCreateIn,
|
|
ListOut,
|
|
ListSnapshot,
|
|
ListUpdateIn,
|
|
MarketOut,
|
|
MemberOut,
|
|
MemberRoleIn,
|
|
)
|
|
from app.security import utcnow
|
|
from app.users import months_setting
|
|
|
|
router = APIRouter(prefix="/api/lists", tags=["lists"])
|
|
|
|
MAX_LISTS_PER_USER = 100
|
|
|
|
|
|
ARCHIVE_DELETE_MONTHS_DEFAULT = 12
|
|
|
|
|
|
def _delete_due(db, lst: ShoppingList) -> datetime | None:
|
|
"""Wann würde die Liste automatisch gelöscht?
|
|
|
|
Maßgeblich ist ausschließlich das Archivdatum des EIGENTÜMERS. Dass
|
|
ein eingeladenes Mitglied die Liste weggeräumt hat, sagt nichts
|
|
darüber, ob sie noch gebraucht wird.
|
|
"""
|
|
owner = db.get(ListMember, (lst.id, lst.owner_id))
|
|
if owner is None or owner.archived_at is None:
|
|
return None
|
|
months = months_setting(
|
|
db, "archive_delete_months", ARCHIVE_DELETE_MONTHS_DEFAULT
|
|
)
|
|
if not months:
|
|
return None
|
|
return owner.archived_at + timedelta(days=months * 30)
|
|
|
|
|
|
def _to_out(
|
|
db, lst: ShoppingList, role: str, may_share: bool = False,
|
|
archived_at: datetime | None = None,
|
|
) -> ListOut:
|
|
count = db.scalar(
|
|
select(func.count()).select_from(ListMember).where(ListMember.list_id == lst.id)
|
|
)
|
|
return ListOut(
|
|
id=lst.id, name=lst.name, owner_id=lst.owner_id, rev=lst.rev,
|
|
role=role, may_share_public=(role == "owner") or may_share,
|
|
member_count=count or 0, created_at=lst.created_at,
|
|
archived=archived_at is not None,
|
|
archived_at=archived_at,
|
|
delete_due_at=_delete_due(db, lst) if role == "owner" else None,
|
|
)
|
|
|
|
|
|
@router.get("", response_model=list[ListOut])
|
|
def my_lists(
|
|
db: DbSession,
|
|
user: VerifiedUser,
|
|
archived: bool | None = Query(
|
|
default=False,
|
|
description="False = nur aktive, True = nur archivierte, "
|
|
"nicht gesetzt = alle",
|
|
),
|
|
):
|
|
rows = db.execute(
|
|
select(
|
|
ShoppingList,
|
|
ListMember.role,
|
|
ListMember.may_share_public,
|
|
ListMember.archived_at,
|
|
)
|
|
.join(ListMember, ListMember.list_id == ShoppingList.id)
|
|
.where(ListMember.user_id == user.id, ShoppingList.deleted_at.is_(None))
|
|
.order_by(ShoppingList.created_at)
|
|
).all()
|
|
|
|
if archived is True:
|
|
rows = [r for r in rows if r[3] is not None]
|
|
elif archived is False:
|
|
rows = [r for r in rows if r[3] is None]
|
|
|
|
return [_to_out(db, lst, role, may, at) for lst, role, may, at in rows]
|
|
|
|
|
|
@router.post("", response_model=ListOut, status_code=status.HTTP_201_CREATED)
|
|
def create_list(payload: ListCreateIn, db: DbSession, user: VerifiedUser):
|
|
owned = db.scalar(
|
|
select(func.count()).select_from(ShoppingList).where(
|
|
ShoppingList.owner_id == user.id, ShoppingList.deleted_at.is_(None)
|
|
)
|
|
)
|
|
if (owned or 0) >= MAX_LISTS_PER_USER:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT,
|
|
f"Höchstens {MAX_LISTS_PER_USER} eigene Listen.",
|
|
)
|
|
|
|
lst = ShoppingList(name=payload.name.strip(), owner_id=user.id, rev=1)
|
|
db.add(lst)
|
|
db.flush()
|
|
db.add(ListMember(
|
|
list_id=lst.id, user_id=user.id, role="owner", may_share_public=True
|
|
))
|
|
db.commit()
|
|
return _to_out(db, lst, "owner")
|
|
|
|
|
|
@router.get("/{list_id}", response_model=ListOut)
|
|
def read_list(lst: ReadableList, db: DbSession, user: VerifiedUser):
|
|
member = db.get(ListMember, (lst.id, user.id))
|
|
return _to_out(db, lst, member.role, member.may_share_public, member.archived_at)
|
|
|
|
|
|
@router.post("/{list_id}/archive", response_model=ListOut)
|
|
def archive_list(lst: ReadableList, db: DbSession, user: VerifiedUser):
|
|
"""Nimmt die Liste aus der eigenen Übersicht.
|
|
|
|
Wirkt nur für die aufrufende Person. Eine gemeinsame Liste
|
|
wegzuräumen, weil einer sie nicht mehr braucht, wäre ein Übergriff
|
|
auf die anderen.
|
|
"""
|
|
member = db.get(ListMember, (lst.id, user.id))
|
|
if member.archived_at is None:
|
|
member.archived_at = utcnow()
|
|
db.commit()
|
|
return _to_out(db, lst, member.role, member.may_share_public, member.archived_at)
|
|
|
|
|
|
@router.post("/{list_id}/unarchive", response_model=ListOut)
|
|
def unarchive_list(lst: ReadableList, db: DbSession, user: VerifiedUser):
|
|
member = db.get(ListMember, (lst.id, user.id))
|
|
member.archived_at = None
|
|
# Zähler zurücksetzen, sonst wandert die Liste beim nächsten
|
|
# Durchlauf sofort wieder ins Archiv.
|
|
member.last_opened_at = utcnow()
|
|
|
|
# Beim Eigentümer zusätzlich die Vorwarnungen zurücknehmen: Die Frist
|
|
# beginnt von vorn, also soll auch wieder gewarnt werden.
|
|
if lst.owner_id == user.id:
|
|
lst.warn_7d_at = None
|
|
lst.warn_1d_at = None
|
|
|
|
db.commit()
|
|
return _to_out(db, lst, member.role, member.may_share_public, None)
|
|
|
|
|
|
@router.get("/{list_id}/snapshot", response_model=ListSnapshot)
|
|
def snapshot(lst: ReadableList, db: DbSession, user: VerifiedUser):
|
|
"""Gesamtstand in einem Aufruf: Liste, gruppierte Ansicht, Märkte,
|
|
Warengruppen. Ersetzt vier einzelne Abfragen."""
|
|
member = db.get(ListMember, (lst.id, user.id))
|
|
markets = db.scalars(
|
|
select(Market)
|
|
.where(Market.list_id == lst.id, Market.deleted_at.is_(None))
|
|
.order_by(Market.sort_order, Market.name)
|
|
).all()
|
|
categories = db.scalars(
|
|
select(Category)
|
|
.where(Category.list_id == lst.id, Category.deleted_at.is_(None))
|
|
.order_by(Category.sort_order, Category.name)
|
|
).all()
|
|
|
|
return ListSnapshot(
|
|
list=_to_out(db, lst, member.role, member.may_share_public, member.archived_at),
|
|
view=build_view(db, lst),
|
|
markets=[MarketOut.model_validate(m) for m in markets],
|
|
categories=[CategoryOut.model_validate(c) for c in categories],
|
|
)
|
|
|
|
|
|
@router.patch("/{list_id}", response_model=ListOut)
|
|
def rename_list(payload: ListUpdateIn, lst: OwnedList, db: DbSession, user: VerifiedUser):
|
|
lst.name = payload.name.strip()
|
|
lst.rev += 1
|
|
db.commit()
|
|
return _to_out(db, lst, "owner")
|
|
|
|
|
|
@router.delete("/{list_id}", response_model=MessageOut)
|
|
def delete_list(lst: OwnedList, db: DbSession):
|
|
# Soft Delete: offline-Clients sollen das Verschwinden noch mitbekommen.
|
|
lst.deleted_at = utcnow()
|
|
lst.rev += 1
|
|
db.commit()
|
|
return MessageOut(detail="Liste gelöscht.")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Mitgliedschaften
|
|
# Einladungen per Mail und QR-Code folgen in Phase 5. Hier lassen sich
|
|
# bestehende Mitgliedschaften nur ansehen, ändern und beenden.
|
|
# --------------------------------------------------------------------------
|
|
|
|
@router.get("/{list_id}/members", response_model=list[MemberOut])
|
|
def list_members(lst: ReadableList, db: DbSession):
|
|
rows = db.execute(
|
|
select(ListMember, User)
|
|
.join(User, User.id == ListMember.user_id)
|
|
.where(ListMember.list_id == lst.id)
|
|
.order_by(ListMember.joined_at)
|
|
).all()
|
|
return [
|
|
MemberOut(
|
|
user_id=u.id, email=u.email, display_name=u.display_name,
|
|
role=m.role, may_share_public=m.may_share_public,
|
|
joined_at=m.joined_at,
|
|
)
|
|
for m, u in rows
|
|
]
|
|
|
|
|
|
@router.put("/{list_id}/members/{user_id}", response_model=MemberOut)
|
|
def set_member_role(
|
|
user_id: str, payload: MemberRoleIn, lst: OwnedList, db: DbSession,
|
|
user: VerifiedUser,
|
|
):
|
|
if user_id == lst.owner_id:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT,
|
|
"Die Rolle des Eigentümers lässt sich nicht ändern. "
|
|
"Übertrage stattdessen die Liste.",
|
|
)
|
|
member = db.get(ListMember, (lst.id, user_id))
|
|
if member is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Kein Mitglied dieser Liste")
|
|
|
|
member.role = payload.role
|
|
if payload.may_share_public is not None:
|
|
member.may_share_public = payload.may_share_public
|
|
lst.rev += 1
|
|
db.commit()
|
|
|
|
target = db.get(User, user_id)
|
|
return MemberOut(
|
|
user_id=target.id, email=target.email, display_name=target.display_name,
|
|
role=member.role, may_share_public=member.may_share_public,
|
|
joined_at=member.joined_at,
|
|
)
|
|
|
|
|
|
@router.delete("/{list_id}/members/{user_id}", response_model=MessageOut)
|
|
def remove_member(user_id: str, list_id: str, db: DbSession, user: VerifiedUser):
|
|
"""Der Eigentümer entzieht anderen den Zugriff; jedes Mitglied kann
|
|
sich selbst entfernen (bequemer über POST /api/lists/{id}/leave).
|
|
|
|
Der Eigentümer kann sich nicht selbst entfernen - die Liste stünde
|
|
sonst ohne Verwaltung da. Dafür gibt es den Eigentümerwechsel."""
|
|
lst = db.scalar(
|
|
select(ShoppingList).where(
|
|
ShoppingList.id == list_id, ShoppingList.deleted_at.is_(None)
|
|
)
|
|
)
|
|
own = db.get(ListMember, (list_id, user.id))
|
|
if lst is None or own is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Liste nicht gefunden")
|
|
|
|
if user_id != user.id and own.role != "owner":
|
|
raise HTTPException(
|
|
status.HTTP_403_FORBIDDEN, "Nur der Eigentümer kann andere entfernen."
|
|
)
|
|
if user_id == lst.owner_id:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT,
|
|
"Der Eigentümer kann die Liste nicht verlassen. Lösche sie oder "
|
|
"übertrage sie zuvor.",
|
|
)
|
|
|
|
member = db.get(ListMember, (list_id, user_id))
|
|
if member is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Kein Mitglied dieser Liste")
|
|
|
|
db.delete(member)
|
|
lst.rev += 1
|
|
db.commit()
|
|
return MessageOut(detail="Mitgliedschaft beendet.")
|