feat: add multi-tenant system with properties, organizations, and public booking
Implement complete multi-property architecture: - Properties (groups of spaces) with public/private visibility - Property managers (many-to-many) with role-based permissions - Organizations with member management - Anonymous/guest booking support via public API (/api/public/*) - Property-scoped spaces, bookings, and settings - Frontend: property selector, organization management, public booking views - Migration script and updated seed data Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -11,6 +11,7 @@ from app.db.session import get_db
|
||||
from app.models.user import User
|
||||
|
||||
security = HTTPBearer()
|
||||
optional_security = HTTPBearer(auto_error=False)
|
||||
|
||||
|
||||
def get_current_user(
|
||||
@@ -40,13 +41,58 @@ def get_current_user(
|
||||
return user
|
||||
|
||||
|
||||
def get_optional_user(
|
||||
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(optional_security)],
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> User | None:
|
||||
"""Get current user or None for anonymous access."""
|
||||
if credentials is None:
|
||||
return None
|
||||
try:
|
||||
token = credentials.credentials
|
||||
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
|
||||
user_id = payload.get("sub")
|
||||
if user_id is None:
|
||||
return None
|
||||
user = db.query(User).filter(User.id == int(user_id)).first()
|
||||
if user is None or not user.is_active:
|
||||
return None
|
||||
return user
|
||||
except JWTError:
|
||||
return None
|
||||
|
||||
|
||||
def get_current_admin(
|
||||
current_user: Annotated[User, Depends(get_current_user)],
|
||||
) -> User:
|
||||
"""Verify current user is admin."""
|
||||
if current_user.role != "admin":
|
||||
"""Verify current user is admin (superadmin or legacy admin)."""
|
||||
if current_user.role not in ("admin", "superadmin"):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Not enough permissions",
|
||||
)
|
||||
return current_user
|
||||
|
||||
|
||||
def get_current_superadmin(
|
||||
current_user: Annotated[User, Depends(get_current_user)],
|
||||
) -> User:
|
||||
"""Verify current user is superadmin."""
|
||||
if current_user.role not in ("admin", "superadmin"):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Superadmin access required",
|
||||
)
|
||||
return current_user
|
||||
|
||||
|
||||
def get_current_manager_or_superadmin(
|
||||
current_user: Annotated[User, Depends(get_current_user)],
|
||||
) -> User:
|
||||
"""Verify current user is manager or superadmin."""
|
||||
if current_user.role not in ("admin", "superadmin", "manager"):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Manager or admin access required",
|
||||
)
|
||||
return current_user
|
||||
|
||||
115
backend/app/core/permissions.py
Normal file
115
backend/app/core/permissions.py
Normal file
@@ -0,0 +1,115 @@
|
||||
"""Property access permission utilities."""
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.organization_member import OrganizationMember
|
||||
from app.models.property import Property
|
||||
from app.models.property_access import PropertyAccess
|
||||
from app.models.property_manager import PropertyManager
|
||||
from app.models.user import User
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
|
||||
def verify_property_access(
|
||||
db: Session, user: User | None, property_id: int, require_manager: bool = False
|
||||
) -> bool:
|
||||
"""Verify user has access to a property. Raises HTTPException if denied."""
|
||||
prop = db.query(Property).filter(Property.id == property_id).first()
|
||||
if not prop:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Property not found")
|
||||
|
||||
if user is None:
|
||||
# Anonymous - only public properties
|
||||
if not prop.is_public:
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Property is private")
|
||||
if require_manager:
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
return True
|
||||
|
||||
# Superadmin always has access
|
||||
if user.role in ("superadmin", "admin"):
|
||||
return True
|
||||
|
||||
if require_manager:
|
||||
# Manager must own this property
|
||||
if user.role == "manager":
|
||||
pm = db.query(PropertyManager).filter(
|
||||
PropertyManager.property_id == property_id,
|
||||
PropertyManager.user_id == user.id,
|
||||
).first()
|
||||
if pm:
|
||||
return True
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
|
||||
|
||||
# Manager has access to managed properties
|
||||
if user.role == "manager":
|
||||
pm = db.query(PropertyManager).filter(
|
||||
PropertyManager.property_id == property_id,
|
||||
PropertyManager.user_id == user.id,
|
||||
).first()
|
||||
if pm:
|
||||
return True
|
||||
|
||||
# Public property - anyone has access
|
||||
if prop.is_public:
|
||||
return True
|
||||
|
||||
# Check explicit access (user)
|
||||
access = db.query(PropertyAccess).filter(
|
||||
PropertyAccess.property_id == property_id,
|
||||
PropertyAccess.user_id == user.id,
|
||||
).first()
|
||||
if access:
|
||||
return True
|
||||
|
||||
# Check explicit access (organization)
|
||||
org_ids = [
|
||||
m.organization_id
|
||||
for m in db.query(OrganizationMember).filter(OrganizationMember.user_id == user.id).all()
|
||||
]
|
||||
if org_ids:
|
||||
org_access = db.query(PropertyAccess).filter(
|
||||
PropertyAccess.property_id == property_id,
|
||||
PropertyAccess.organization_id.in_(org_ids),
|
||||
).first()
|
||||
if org_access:
|
||||
return True
|
||||
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No access to this property")
|
||||
|
||||
|
||||
def get_manager_property_ids(db: Session, user_id: int) -> list[int]:
|
||||
"""Get list of property IDs managed by user."""
|
||||
return [
|
||||
pm.property_id
|
||||
for pm in db.query(PropertyManager).filter(PropertyManager.user_id == user_id).all()
|
||||
]
|
||||
|
||||
|
||||
def get_user_accessible_property_ids(db: Session, user_id: int) -> list[int]:
|
||||
"""Get all property IDs accessible by user (public + explicitly granted)."""
|
||||
# Public properties
|
||||
public_ids = [
|
||||
p.id
|
||||
for p in db.query(Property).filter(Property.is_public == True, Property.is_active == True).all() # noqa: E712
|
||||
]
|
||||
|
||||
# Directly granted
|
||||
direct_ids = [
|
||||
a.property_id
|
||||
for a in db.query(PropertyAccess).filter(PropertyAccess.user_id == user_id).all()
|
||||
]
|
||||
|
||||
# Org granted
|
||||
org_ids = [
|
||||
m.organization_id
|
||||
for m in db.query(OrganizationMember).filter(OrganizationMember.user_id == user_id).all()
|
||||
]
|
||||
org_property_ids = []
|
||||
if org_ids:
|
||||
org_property_ids = [
|
||||
a.property_id
|
||||
for a in db.query(PropertyAccess).filter(PropertyAccess.organization_id.in_(org_ids)).all()
|
||||
]
|
||||
|
||||
return list(set(public_ids + direct_ids + org_property_ids))
|
||||
Reference in New Issue
Block a user