127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""
|
|
context.py — Tenant and location context resolution.
|
|
load_tenant_context() and load_location_context() are registered
|
|
as before_request hooks in the tenant app factory.
|
|
"""
|
|
|
|
import logging
|
|
from flask import g, session, redirect, url_for, request, abort
|
|
from flask_login import current_user
|
|
from app.models.platform import Tenant
|
|
from app.models.salon import Location, StaffLocation, Staff
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def load_tenant_context():
|
|
"""
|
|
Resolve g.tenant from the authenticated user's session.
|
|
Called on every request in the tenant portal.
|
|
Redirects to a locked page if the tenant is suspended or cancelled.
|
|
Skips public routes (checkin, booking, staff-login, auth).
|
|
"""
|
|
# Skip context resolution for public / auth endpoints
|
|
public_blueprints = {"tenant_auth", "staff_auth", "checkin", "booking", "static"}
|
|
if request.blueprint in public_blueprints:
|
|
g.tenant = None
|
|
return
|
|
|
|
if not current_user.is_authenticated:
|
|
g.tenant = None
|
|
return
|
|
|
|
# Resolve tenant_id from the authenticated principal
|
|
if hasattr(current_user, "tenant_id"):
|
|
tenant_id = current_user.tenant_id
|
|
else:
|
|
g.tenant = None
|
|
return
|
|
|
|
tenant = Tenant.query.get(tenant_id)
|
|
if tenant is None:
|
|
logger.warning("Tenant %s not found for user %s", tenant_id, current_user)
|
|
abort(403)
|
|
|
|
g.tenant = tenant
|
|
|
|
# Enforce subscription status
|
|
if tenant.status == "suspended":
|
|
if request.endpoint != "tenant_auth.suspended":
|
|
return redirect(url_for("tenant_auth.suspended"))
|
|
|
|
if tenant.status == "cancelled":
|
|
if request.endpoint != "tenant_auth.cancelled":
|
|
return redirect(url_for("tenant_auth.cancelled"))
|
|
|
|
logger.debug("Tenant context loaded: %s", tenant.slug)
|
|
|
|
|
|
def load_location_context():
|
|
"""
|
|
Resolve g.location from session or default to the tenant's primary location.
|
|
For tenant_staff, validates that the staff member is assigned to the location.
|
|
Must be called after load_tenant_context().
|
|
"""
|
|
if not hasattr(g, "tenant") or g.tenant is None:
|
|
g.location = None
|
|
return
|
|
|
|
if not current_user.is_authenticated:
|
|
g.location = None
|
|
return
|
|
|
|
tenant_id = g.tenant.id
|
|
|
|
# Try to load from session
|
|
location_id = session.get("active_location_id")
|
|
|
|
if location_id:
|
|
location = Location.query.filter_by(
|
|
id=location_id,
|
|
tenant_id=tenant_id,
|
|
is_active=True,
|
|
).filter(Location.deleted_at.is_(None)).first()
|
|
else:
|
|
location = None
|
|
|
|
# Fall back to primary location
|
|
if location is None:
|
|
location = Location.query.filter_by(
|
|
tenant_id=tenant_id,
|
|
is_primary=True,
|
|
is_active=True,
|
|
).filter(Location.deleted_at.is_(None)).first()
|
|
|
|
if location is None:
|
|
# Last resort: first active location
|
|
location = Location.query.filter_by(
|
|
tenant_id=tenant_id,
|
|
is_active=True,
|
|
).filter(Location.deleted_at.is_(None)).first()
|
|
|
|
if location:
|
|
session["active_location_id"] = location.id
|
|
|
|
# For tenant_staff: enforce location assignment
|
|
if location and hasattr(current_user, "get_id"):
|
|
user_id_str = current_user.get_id()
|
|
if user_id_str and user_id_str.startswith("staff:"):
|
|
staff_id = int(user_id_str.split(":")[1])
|
|
assigned = StaffLocation.query.filter_by(
|
|
staff_id=staff_id,
|
|
location_id=location.id,
|
|
tenant_id=tenant_id,
|
|
).first()
|
|
if not assigned:
|
|
logger.warning(
|
|
"Staff %s attempted access to unassigned location %s",
|
|
staff_id, location.id,
|
|
)
|
|
abort(403)
|
|
|
|
g.location = location
|
|
logger.debug(
|
|
"Location context loaded: %s",
|
|
location.name if location else "None",
|
|
)
|