641 lines
22 KiB
Python
641 lines
22 KiB
Python
import os
|
||
import secrets
|
||
from datetime import datetime, timedelta, date
|
||
from typing import Optional, Dict
|
||
from decimal import Decimal
|
||
|
||
from fastapi import FastAPI, Request, Depends, HTTPException, Form, status
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.templating import Jinja2Templates
|
||
from starlette.middleware.sessions import SessionMiddleware
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .database import get_db
|
||
from .models import Tenant, User, Product, Event, Order, OrderItem, Invitation
|
||
from .auth import hash_password, verify_password
|
||
|
||
app = FastAPI(title="Schmankerl – Das moderne Brotzeitportal")
|
||
|
||
# Session Middleware für sichere Login-Cookies
|
||
app.add_middleware(
|
||
SessionMiddleware,
|
||
secret_key=os.getenv("SECRET_KEY", "schmankerl-secret-key-waiswuaschd-2026!"),
|
||
session_cookie="schmankerl_session",
|
||
max_age=86400 * 30 # 30 Tage
|
||
)
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static")
|
||
templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
|
||
|
||
|
||
def resolve_tenant(request: Request, db: Session) -> Optional[Tenant]:
|
||
"""Ermittelt den Mandanten anhand der Subdomain (Host-Header)."""
|
||
host = request.headers.get("host", "").split(":")[0].lower()
|
||
parts = host.split(".")
|
||
|
||
slug = "demo"
|
||
if len(parts) >= 4 and parts[-3:] == ["waiswuaschd", "de", ""]:
|
||
slug = parts[0]
|
||
elif len(parts) >= 3 and parts[1] == "dev" and parts[2] == "waiswuaschd":
|
||
slug = parts[0]
|
||
elif parts and parts[0] not in ["localhost", "127", "dev"]:
|
||
slug = parts[0]
|
||
|
||
tenant = db.query(Tenant).filter(Tenant.slug == slug, Tenant.is_active == True).first()
|
||
if not tenant:
|
||
tenant = db.query(Tenant).filter(Tenant.slug == "demo").first()
|
||
return tenant
|
||
|
||
|
||
def get_current_user(request: Request, db: Session) -> Optional[User]:
|
||
"""Liest den eingeloggten Benutzer aus der Session."""
|
||
user_id = request.session.get("user_id")
|
||
if not user_id:
|
||
return None
|
||
return db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
||
|
||
|
||
def require_admin(request: Request, db: Session) -> User:
|
||
"""Stellt sicher, dass ein Administrator eingeloggt ist."""
|
||
current_user = get_current_user(request, db)
|
||
if not current_user:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Anmeldung erforderlich")
|
||
if current_user.role != "admin":
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Nur für Administratoren zugänglich")
|
||
return current_user
|
||
|
||
|
||
def seed_demo_users_if_needed(tenant: Tenant, db: Session):
|
||
"""Stellt sicher, dass die Standard-Accounts für die Demo existieren."""
|
||
demo_user = db.query(User).filter(User.tenant_id == tenant.id, User.email == "maxi@waiswuaschd.de").first()
|
||
if not demo_user:
|
||
demo_user = User(
|
||
tenant_id=tenant.id,
|
||
email="maxi@waiswuaschd.de",
|
||
username="maxi",
|
||
first_name="Maxi",
|
||
last_name="Mustermann",
|
||
password_hash=hash_password("demo123"),
|
||
role="user"
|
||
)
|
||
db.add(demo_user)
|
||
|
||
admin_user = db.query(User).filter(User.tenant_id == tenant.id, User.email == "admin@waiswuaschd.de").first()
|
||
if not admin_user:
|
||
admin_user = User(
|
||
tenant_id=tenant.id,
|
||
email="admin@waiswuaschd.de",
|
||
username="admin",
|
||
first_name="Franz",
|
||
last_name="Xaver",
|
||
password_hash=hash_password("admin123"),
|
||
role="admin"
|
||
)
|
||
db.add(admin_user)
|
||
db.commit()
|
||
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def index(request: Request, db: Session = Depends(get_db)):
|
||
tenant = resolve_tenant(request, db)
|
||
if not tenant:
|
||
return templates.TemplateResponse(request, "not_found_tenant.html", context={})
|
||
|
||
current_user = get_current_user(request, db)
|
||
if not current_user:
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
now = datetime.now()
|
||
|
||
# Aktuelle offene Runde suchen
|
||
event = db.query(Event).filter(
|
||
Event.tenant_id == tenant.id,
|
||
Event.status == "open"
|
||
).order_by(Event.deadline.asc()).first()
|
||
|
||
# Wenn keine offene, das neueste anzeigen
|
||
if not event:
|
||
event = db.query(Event).filter(Event.tenant_id == tenant.id).order_by(Event.id.desc()).first()
|
||
|
||
# Fristprüfung & Countdown
|
||
is_closed = True
|
||
seconds_remaining = 0
|
||
if event:
|
||
is_closed = (event.status != "open") or (event.deadline < now)
|
||
if not is_closed:
|
||
seconds_remaining = max(0, int((event.deadline - now).total_seconds()))
|
||
|
||
# Alle aktiven Produkte des Mandanten
|
||
products = db.query(Product).filter(
|
||
Product.tenant_id == tenant.id,
|
||
Product.is_active == True
|
||
).order_by(Product.sort_order.asc(), Product.id.asc()).all()
|
||
|
||
# Bestehende Bestellung des Users in diesem Event
|
||
existing_order = None
|
||
user_items: Dict[int, int] = {}
|
||
if event:
|
||
existing_order = db.query(Order).filter(
|
||
Order.tenant_id == tenant.id,
|
||
Order.event_id == event.id,
|
||
Order.user_id == current_user.id
|
||
).first()
|
||
if existing_order:
|
||
for it in existing_order.items:
|
||
user_items[it.product_id] = it.quantity
|
||
|
||
# Bestellungen insgesamt zählen
|
||
order_count = 0
|
||
if event:
|
||
order_count = db.query(Order).filter(Order.event_id == event.id).count()
|
||
|
||
return templates.TemplateResponse(request, "index.html", context={
|
||
"tenant": tenant,
|
||
"current_user": current_user,
|
||
"event": event,
|
||
"is_closed": is_closed,
|
||
"seconds_remaining": seconds_remaining,
|
||
"products": products,
|
||
"existing_order": existing_order,
|
||
"user_items": user_items,
|
||
"order_count": order_count,
|
||
"now": now
|
||
})
|
||
|
||
|
||
@app.get("/login", response_class=HTMLResponse)
|
||
def login_page(request: Request, db: Session = Depends(get_db), error: Optional[str] = None, success: Optional[str] = None):
|
||
tenant = resolve_tenant(request, db)
|
||
if tenant and tenant.slug == "demo":
|
||
seed_demo_users_if_needed(tenant, db)
|
||
|
||
current_user = get_current_user(request, db)
|
||
if current_user:
|
||
return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
return templates.TemplateResponse(request, "login.html", context={
|
||
"tenant": tenant,
|
||
"error": error,
|
||
"success": success,
|
||
"current_user": None
|
||
})
|
||
|
||
|
||
@app.post("/login")
|
||
def login(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
clean_email = email.strip().lower()
|
||
|
||
user = db.query(User).filter(User.tenant_id == tenant.id, User.email == clean_email).first()
|
||
if not user or not verify_password(password, user.password_hash):
|
||
return templates.TemplateResponse(request, "login.html", context={
|
||
"tenant": tenant,
|
||
"error": "Ungültige E-Mail-Adresse oder falsches Passwort.",
|
||
"current_user": None
|
||
}, status_code=400)
|
||
|
||
# Session setzen
|
||
request.session["user_id"] = user.id
|
||
return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.post("/login/quick")
|
||
def quick_demo_login(
|
||
request: Request,
|
||
role: str = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
if not tenant:
|
||
raise HTTPException(status_code=404, detail="Mandant nicht gefunden")
|
||
|
||
seed_demo_users_if_needed(tenant, db)
|
||
|
||
target_email = "admin@waiswuaschd.de" if role == "admin" else "maxi@waiswuaschd.de"
|
||
user = db.query(User).filter(User.tenant_id == tenant.id, User.email == target_email).first()
|
||
if not user:
|
||
raise HTTPException(status_code=500, detail="Demo-Benutzer konnte nicht geladen werden")
|
||
|
||
request.session["user_id"] = user.id
|
||
if role == "admin":
|
||
return RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)
|
||
return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.get("/logout")
|
||
def logout(request: Request):
|
||
request.session.clear()
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
# --- EINLADUNGS- & BEITRITTS-FLOW (/join) ---
|
||
|
||
@app.get("/join", response_class=HTMLResponse)
|
||
def join_page(request: Request, token: str, db: Session = Depends(get_db)):
|
||
tenant = resolve_tenant(request, db)
|
||
if not tenant:
|
||
raise HTTPException(status_code=404, detail="Mandant nicht gefunden")
|
||
|
||
invitation = db.query(Invitation).filter(
|
||
Invitation.token == token,
|
||
Invitation.tenant_id == tenant.id
|
||
).first()
|
||
|
||
error = None
|
||
if not invitation:
|
||
error = "Dieser Einladungslink ist ungültig oder gehört zu einem anderen Team."
|
||
elif invitation.used_at is not None:
|
||
error = "Dieser Einladungslink wurde bereits eingelöst. Bitte melde dich mit deinem Konto an."
|
||
elif invitation.expires_at < datetime.now():
|
||
error = "Dieser Einladungslink ist leider abgelaufen. Bitte frage deinen Admin nach einem neuen Link."
|
||
|
||
return templates.TemplateResponse(request, "join.html", context={
|
||
"tenant": tenant,
|
||
"invitation": invitation,
|
||
"token": token,
|
||
"error": error,
|
||
"current_user": None
|
||
})
|
||
|
||
|
||
@app.post("/join")
|
||
def process_join(
|
||
request: Request,
|
||
token: str = Form(...),
|
||
first_name: str = Form(...),
|
||
last_name: str = Form(...),
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
invitation = db.query(Invitation).filter(
|
||
Invitation.token == token,
|
||
Invitation.tenant_id == tenant.id
|
||
).first()
|
||
|
||
if not invitation or invitation.used_at is not None or invitation.expires_at < datetime.now():
|
||
raise HTTPException(status_code=400, detail="Ungültiger oder abgelaufener Einladungslink")
|
||
|
||
clean_email = email.strip().lower()
|
||
existing_user = db.query(User).filter(
|
||
User.tenant_id == tenant.id,
|
||
User.email == clean_email
|
||
).first()
|
||
|
||
if existing_user:
|
||
return templates.TemplateResponse(request, "join.html", context={
|
||
"tenant": tenant,
|
||
"invitation": invitation,
|
||
"token": token,
|
||
"error": "Diese E-Mail-Adresse ist bereits im Team registriert. Bitte melde dich an.",
|
||
"current_user": None
|
||
}, status_code=400)
|
||
|
||
# Neuen Benutzer anlegen
|
||
username = clean_email.split("@")[0]
|
||
new_user = User(
|
||
tenant_id=tenant.id,
|
||
email=clean_email,
|
||
username=username,
|
||
first_name=first_name.strip(),
|
||
last_name=last_name.strip(),
|
||
password_hash=hash_password(password),
|
||
role=invitation.role
|
||
)
|
||
db.add(new_user)
|
||
|
||
# Einladung als verwendet markieren
|
||
invitation.used_at = datetime.now()
|
||
db.commit()
|
||
db.refresh(new_user)
|
||
|
||
# Direkt einloggen
|
||
request.session["user_id"] = new_user.id
|
||
return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
# --- BESTELLUNGEN (/order) ---
|
||
|
||
@app.post("/order")
|
||
async def place_order(
|
||
request: Request,
|
||
event_id: int = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
current_user = get_current_user(request, db)
|
||
if not current_user:
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
event = db.query(Event).filter(Event.id == event_id, Event.tenant_id == tenant.id).first()
|
||
if not event or event.status != "open":
|
||
raise HTTPException(status_code=400, detail="Diese Bestellrunde ist geschlossen.")
|
||
|
||
if event.deadline < datetime.now():
|
||
raise HTTPException(status_code=400, detail="Die Bestellfrist für diese Runde ist leider abgelaufen!")
|
||
|
||
form_data = await request.form()
|
||
order_items = []
|
||
total_price = Decimal("0.00")
|
||
|
||
products = db.query(Product).filter(Product.tenant_id == tenant.id, Product.is_active == True).all()
|
||
for prod in products:
|
||
qty_str = form_data.get(f"qty_{prod.id}", "0")
|
||
try:
|
||
qty = int(qty_str)
|
||
except ValueError:
|
||
qty = 0
|
||
|
||
if qty > 0:
|
||
item_total = prod.price * qty
|
||
total_price += item_total
|
||
order_items.append(OrderItem(
|
||
product_id=prod.id,
|
||
quantity=qty,
|
||
unit_price=prod.price
|
||
))
|
||
|
||
if not order_items:
|
||
raise HTTPException(status_code=400, detail="Bitte wähle mindestens einen Artikel aus!")
|
||
|
||
# Bestehende Bestellung des Users in diesem Event aktualisieren oder neu anlegen
|
||
existing_order = db.query(Order).filter(
|
||
Order.tenant_id == tenant.id,
|
||
Order.event_id == event.id,
|
||
Order.user_id == current_user.id
|
||
).first()
|
||
|
||
if existing_order:
|
||
db.query(OrderItem).filter(OrderItem.order_id == existing_order.id).delete()
|
||
existing_order.total_price = total_price
|
||
for item in order_items:
|
||
item.order_id = existing_order.id
|
||
db.add(item)
|
||
db.commit()
|
||
order_id = existing_order.id
|
||
else:
|
||
new_order = Order(
|
||
tenant_id=tenant.id,
|
||
event_id=event.id,
|
||
user_id=current_user.id,
|
||
total_price=total_price,
|
||
is_paid=False
|
||
)
|
||
db.add(new_order)
|
||
db.flush()
|
||
for item in order_items:
|
||
item.order_id = new_order.id
|
||
db.add(item)
|
||
db.commit()
|
||
order_id = new_order.id
|
||
|
||
return RedirectResponse(url=f"/order/{order_id}/success", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.get("/order/{order_id}/success", response_class=HTMLResponse)
|
||
def order_success(order_id: int, request: Request, db: Session = Depends(get_db)):
|
||
tenant = resolve_tenant(request, db)
|
||
current_user = get_current_user(request, db)
|
||
if not current_user:
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
order = db.query(Order).filter(Order.id == order_id, Order.tenant_id == tenant.id).first()
|
||
if not order:
|
||
raise HTTPException(status_code=404, detail="Bestellung nicht gefunden")
|
||
|
||
return templates.TemplateResponse(request, "order_success.html", context={
|
||
"tenant": tenant,
|
||
"current_user": current_user,
|
||
"order": order
|
||
})
|
||
|
||
|
||
# --- ADMIN COCKPIT (/admin) ---
|
||
|
||
@app.get("/admin", response_class=HTMLResponse)
|
||
def admin_dashboard(
|
||
request: Request,
|
||
event_id: Optional[int] = None,
|
||
tab: str = "orders",
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
if not tenant:
|
||
return templates.TemplateResponse(request, "not_found_tenant.html", context={})
|
||
|
||
current_user = get_current_user(request, db)
|
||
if not current_user:
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
||
if current_user.role != "admin":
|
||
raise HTTPException(status_code=403, detail="Zugriff nur für Administratoren")
|
||
|
||
# Alle Events des Mandanten
|
||
all_events = db.query(Event).filter(Event.tenant_id == tenant.id).order_by(Event.id.desc()).all()
|
||
|
||
# Ausgewähltes Event bestimmen
|
||
selected_event = None
|
||
if event_id:
|
||
selected_event = db.query(Event).filter(Event.id == event_id, Event.tenant_id == tenant.id).first()
|
||
if not selected_event and all_events:
|
||
selected_event = all_events[0]
|
||
|
||
orders = []
|
||
item_sums = {}
|
||
total_revenue = Decimal("0.00")
|
||
total_paid = Decimal("0.00")
|
||
|
||
if selected_event:
|
||
orders = db.query(Order).filter(Order.event_id == selected_event.id).order_by(Order.created_at.asc()).all()
|
||
for ord_obj in orders:
|
||
total_revenue += ord_obj.total_price
|
||
if ord_obj.is_paid:
|
||
total_paid += ord_obj.total_price
|
||
|
||
for it in ord_obj.items:
|
||
pname = it.product.name if it.product else "Artikel"
|
||
unit = it.product.unit if it.product else "Stk"
|
||
if pname not in item_sums:
|
||
item_sums[pname] = {"quantity": 0, "unit": unit}
|
||
item_sums[pname]["quantity"] += it.quantity
|
||
|
||
# Teammitglieder & Einladungen laden
|
||
team_members = db.query(User).filter(User.tenant_id == tenant.id).order_by(User.role.asc(), User.first_name.asc()).all()
|
||
pending_invitations = db.query(Invitation).filter(
|
||
Invitation.tenant_id == tenant.id,
|
||
Invitation.used_at == None,
|
||
Invitation.expires_at > datetime.now()
|
||
).order_by(Invitation.created_at.desc()).all()
|
||
|
||
return templates.TemplateResponse(request, "admin.html", context={
|
||
"tenant": tenant,
|
||
"current_user": current_user,
|
||
"all_events": all_events,
|
||
"event": selected_event,
|
||
"orders": orders,
|
||
"item_sums": item_sums,
|
||
"total_revenue": total_revenue,
|
||
"total_paid": total_paid,
|
||
"team_members": team_members,
|
||
"pending_invitations": pending_invitations,
|
||
"active_tab": tab,
|
||
"now": datetime.now()
|
||
})
|
||
|
||
|
||
@app.post("/admin/events/new")
|
||
def admin_create_event(
|
||
request: Request,
|
||
title: str = Form(...),
|
||
event_date: str = Form(...),
|
||
deadline_date: str = Form(...),
|
||
deadline_time: str = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
require_admin(request, db)
|
||
|
||
try:
|
||
ev_date = datetime.strptime(event_date.strip(), "%Y-%m-%d").date()
|
||
dl_str = f"{deadline_date.strip()} {deadline_time.strip()}"
|
||
deadline_dt = datetime.strptime(dl_str, "%Y-%m-%d %H:%M")
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=f"Ungültiges Datumsformat: {e}")
|
||
|
||
new_event = Event(
|
||
tenant_id=tenant.id,
|
||
title=title.strip(),
|
||
event_date=ev_date,
|
||
deadline=deadline_dt,
|
||
status="open"
|
||
)
|
||
db.add(new_event)
|
||
db.commit()
|
||
db.refresh(new_event)
|
||
|
||
return RedirectResponse(url=f"/admin?event_id={new_event.id}&tab=event", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.post("/admin/events/{event_id}/edit")
|
||
def admin_edit_event(
|
||
event_id: int,
|
||
request: Request,
|
||
title: str = Form(...),
|
||
event_date: str = Form(...),
|
||
deadline_date: str = Form(...),
|
||
deadline_time: str = Form(...),
|
||
status_val: str = Form(...),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
require_admin(request, db)
|
||
|
||
event = db.query(Event).filter(Event.id == event_id, Event.tenant_id == tenant.id).first()
|
||
if not event:
|
||
raise HTTPException(status_code=404, detail="Event nicht gefunden")
|
||
|
||
try:
|
||
ev_date = datetime.strptime(event_date.strip(), "%Y-%m-%d").date()
|
||
dl_str = f"{deadline_date.strip()} {deadline_time.strip()}"
|
||
deadline_dt = datetime.strptime(dl_str, "%Y-%m-%d %H:%M")
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=f"Ungültiges Datumsformat: {e}")
|
||
|
||
event.title = title.strip()
|
||
event.event_date = ev_date
|
||
event.deadline = deadline_dt
|
||
event.status = status_val
|
||
db.commit()
|
||
|
||
return RedirectResponse(url=f"/admin?event_id={event.id}&tab=event", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.post("/admin/events/{event_id}/toggle-status")
|
||
def admin_toggle_event_status(
|
||
event_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
require_admin(request, db)
|
||
|
||
event = db.query(Event).filter(Event.id == event_id, Event.tenant_id == tenant.id).first()
|
||
if not event:
|
||
raise HTTPException(status_code=404, detail="Event nicht gefunden")
|
||
|
||
event.status = "closed" if event.status == "open" else "open"
|
||
db.commit()
|
||
|
||
return RedirectResponse(url=f"/admin?event_id={event.id}", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.post("/admin/invitations/create")
|
||
def admin_create_invitation(
|
||
request: Request,
|
||
email: Optional[str] = Form(None),
|
||
role: str = Form("user"),
|
||
valid_days: int = Form(14),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
admin = require_admin(request, db)
|
||
|
||
token = secrets.token_urlsafe(20)
|
||
expires_at = datetime.now() + timedelta(days=valid_days)
|
||
clean_email = email.strip().lower() if email and email.strip() else None
|
||
|
||
inv = Invitation(
|
||
tenant_id=tenant.id,
|
||
token=token,
|
||
email=clean_email,
|
||
role=role,
|
||
created_by=admin.id,
|
||
expires_at=expires_at
|
||
)
|
||
db.add(inv)
|
||
db.commit()
|
||
|
||
return RedirectResponse(url="/admin?tab=team", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.post("/admin/invitations/{inv_id}/revoke")
|
||
def admin_revoke_invitation(
|
||
inv_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
tenant = resolve_tenant(request, db)
|
||
require_admin(request, db)
|
||
|
||
inv = db.query(Invitation).filter(Invitation.id == inv_id, Invitation.tenant_id == tenant.id).first()
|
||
if inv:
|
||
db.delete(inv)
|
||
db.commit()
|
||
|
||
return RedirectResponse(url="/admin?tab=team", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.post("/admin/toggle-paid/{order_id}")
|
||
def toggle_paid(order_id: int, request: Request, db: Session = Depends(get_db)):
|
||
current_user = get_current_user(request, db)
|
||
if not current_user:
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
order = db.query(Order).filter(Order.id == order_id).first()
|
||
if order:
|
||
order.is_paid = not order.is_paid
|
||
db.commit()
|
||
return RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)
|
||
|
||
|
||
@app.get("/health")
|
||
def health():
|
||
return {"status": "ok", "app": "Schmankerl"}
|