699 lines
26 KiB
Python
699 lines
26 KiB
Python
"""
|
|
Enhanced Tenant Service
|
|
Business logic layer using repository pattern for tenant operations
|
|
"""
|
|
|
|
import structlog
|
|
from datetime import datetime, timezone
|
|
from typing import Optional, List, Dict, Any
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.repositories import TenantRepository, TenantMemberRepository, SubscriptionRepository
|
|
from app.models.tenants import Tenant, TenantMember, Subscription
|
|
from app.schemas.tenants import (
|
|
BakeryRegistration, TenantResponse, TenantAccessResponse,
|
|
TenantUpdate, TenantMemberResponse
|
|
)
|
|
from app.services.messaging import publish_tenant_created, publish_member_added
|
|
from shared.database.exceptions import DatabaseError, ValidationError, DuplicateRecordError
|
|
from shared.database.base import create_database_manager
|
|
from shared.database.unit_of_work import UnitOfWork
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class EnhancedTenantService:
|
|
"""Enhanced tenant management business logic using repository pattern with dependency injection"""
|
|
|
|
def __init__(self, database_manager=None):
|
|
self.database_manager = database_manager or create_database_manager()
|
|
|
|
async def _init_repositories(self, session):
|
|
"""Initialize repositories with session"""
|
|
self.tenant_repo = TenantRepository(Tenant, session)
|
|
self.member_repo = TenantMemberRepository(TenantMember, session)
|
|
self.subscription_repo = SubscriptionRepository(Subscription, session)
|
|
return {
|
|
'tenant': self.tenant_repo,
|
|
'member': self.member_repo,
|
|
'subscription': self.subscription_repo
|
|
}
|
|
|
|
async def create_bakery(
|
|
self,
|
|
bakery_data: BakeryRegistration,
|
|
owner_id: str,
|
|
session=None
|
|
) -> TenantResponse:
|
|
"""Create a new bakery/tenant with enhanced validation and features using repository pattern"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
async with UnitOfWork(db_session) as uow:
|
|
# Register repositories
|
|
tenant_repo = uow.register_repository("tenants", TenantRepository, Tenant)
|
|
member_repo = uow.register_repository("members", TenantMemberRepository, TenantMember)
|
|
subscription_repo = uow.register_repository("subscriptions", SubscriptionRepository, Subscription)
|
|
|
|
# Prepare tenant data
|
|
tenant_data = {
|
|
"name": bakery_data.name,
|
|
"business_type": bakery_data.business_type,
|
|
"address": bakery_data.address,
|
|
"city": bakery_data.city,
|
|
"postal_code": bakery_data.postal_code,
|
|
"phone": bakery_data.phone,
|
|
"owner_id": owner_id,
|
|
"email": getattr(bakery_data, 'email', None),
|
|
"latitude": getattr(bakery_data, 'latitude', None),
|
|
"longitude": getattr(bakery_data, 'longitude', None),
|
|
"is_active": True
|
|
}
|
|
|
|
# Create tenant using repository
|
|
tenant = await tenant_repo.create_tenant(tenant_data)
|
|
|
|
# Create owner membership
|
|
membership_data = {
|
|
"tenant_id": str(tenant.id),
|
|
"user_id": owner_id,
|
|
"role": "owner",
|
|
"is_active": True
|
|
}
|
|
|
|
owner_membership = await member_repo.create_membership(membership_data)
|
|
|
|
# Get subscription plan from user's registration using standardized auth client
|
|
selected_plan = "starter" # Default fallback
|
|
try:
|
|
from shared.clients.auth_client import AuthServiceClient
|
|
from app.core.config import settings
|
|
|
|
auth_client = AuthServiceClient(settings)
|
|
selected_plan = await auth_client.get_subscription_plan_from_registration(owner_id)
|
|
|
|
logger.info("Retrieved subscription plan from registration",
|
|
tenant_id=tenant.id,
|
|
owner_id=owner_id,
|
|
plan=selected_plan)
|
|
|
|
except Exception as e:
|
|
logger.warning("Could not retrieve subscription plan from auth service, using default",
|
|
error=str(e),
|
|
owner_id=owner_id,
|
|
default_plan=selected_plan)
|
|
|
|
# Create subscription with selected or default plan
|
|
subscription_data = {
|
|
"tenant_id": str(tenant.id),
|
|
"plan": selected_plan,
|
|
"status": "active"
|
|
}
|
|
|
|
subscription = await subscription_repo.create_subscription(subscription_data)
|
|
|
|
logger.info("Subscription created",
|
|
tenant_id=tenant.id,
|
|
plan=selected_plan)
|
|
|
|
# Commit the transaction
|
|
await uow.commit()
|
|
|
|
# Publish event
|
|
try:
|
|
await publish_tenant_created(str(tenant.id), owner_id, bakery_data.name)
|
|
except Exception as e:
|
|
logger.warning("Failed to publish tenant created event", error=str(e))
|
|
|
|
logger.info("Bakery created successfully",
|
|
tenant_id=tenant.id,
|
|
name=bakery_data.name,
|
|
owner_id=owner_id,
|
|
subdomain=tenant.subdomain)
|
|
|
|
return TenantResponse.from_orm(tenant)
|
|
|
|
except (ValidationError, DuplicateRecordError) as e:
|
|
logger.error("Validation error creating bakery",
|
|
name=bakery_data.name,
|
|
owner_id=owner_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error("Error creating bakery",
|
|
name=bakery_data.name,
|
|
owner_id=owner_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to create bakery"
|
|
)
|
|
|
|
async def verify_user_access(
|
|
self,
|
|
user_id: str,
|
|
tenant_id: str
|
|
) -> TenantAccessResponse:
|
|
"""Verify if user has access to tenant with enhanced permissions"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
access_info = await self.member_repo.verify_user_access(user_id, tenant_id)
|
|
|
|
return TenantAccessResponse(
|
|
has_access=access_info["has_access"],
|
|
role=access_info["role"],
|
|
permissions=access_info["permissions"],
|
|
membership_id=access_info.get("membership_id"),
|
|
joined_at=access_info.get("joined_at")
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Error verifying user access",
|
|
user_id=user_id,
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
return TenantAccessResponse(
|
|
has_access=False,
|
|
role="none",
|
|
permissions=[]
|
|
)
|
|
|
|
async def get_tenant_by_id(self, tenant_id: str) -> Optional[TenantResponse]:
|
|
"""Get tenant by ID with enhanced data"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
tenant = await self.tenant_repo.get_by_id(tenant_id)
|
|
if tenant:
|
|
return TenantResponse.from_orm(tenant)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting tenant",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
return None
|
|
|
|
async def get_tenant_by_subdomain(self, subdomain: str) -> Optional[TenantResponse]:
|
|
"""Get tenant by subdomain"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
tenant = await self.tenant_repo.get_by_subdomain(subdomain)
|
|
if tenant:
|
|
return TenantResponse.from_orm(tenant)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting tenant by subdomain",
|
|
subdomain=subdomain,
|
|
error=str(e))
|
|
return None
|
|
|
|
async def get_user_tenants(self, owner_id: str) -> List[TenantResponse]:
|
|
"""Get all tenants owned by a user"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
tenants = await self.tenant_repo.get_tenants_by_owner(owner_id)
|
|
return [TenantResponse.from_orm(tenant) for tenant in tenants]
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting user tenants",
|
|
owner_id=owner_id,
|
|
error=str(e))
|
|
return []
|
|
|
|
async def search_tenants(
|
|
self,
|
|
search_term: str,
|
|
business_type: str = None,
|
|
city: str = None,
|
|
skip: int = 0,
|
|
limit: int = 50
|
|
) -> List[TenantResponse]:
|
|
"""Search tenants with filters"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
tenants = await self.tenant_repo.search_tenants(
|
|
search_term, business_type, city, skip, limit
|
|
)
|
|
return [TenantResponse.from_orm(tenant) for tenant in tenants]
|
|
|
|
except Exception as e:
|
|
logger.error("Error searching tenants",
|
|
search_term=search_term,
|
|
error=str(e))
|
|
return []
|
|
|
|
async def update_tenant(
|
|
self,
|
|
tenant_id: str,
|
|
update_data: TenantUpdate,
|
|
user_id: str,
|
|
session: AsyncSession = None
|
|
) -> TenantResponse:
|
|
"""Update tenant information with permission checks"""
|
|
|
|
try:
|
|
# Verify user has admin access
|
|
access = await self.verify_user_access(user_id, tenant_id)
|
|
if not access.has_access or access.role not in ["owner", "admin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Insufficient permissions to update tenant"
|
|
)
|
|
|
|
# Update tenant using repository with proper session management
|
|
update_values = update_data.dict(exclude_unset=True)
|
|
if update_values:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
updated_tenant = await self.tenant_repo.update(tenant_id, update_values)
|
|
|
|
if not updated_tenant:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Tenant not found"
|
|
)
|
|
|
|
logger.info("Tenant updated successfully",
|
|
tenant_id=tenant_id,
|
|
updated_by=user_id,
|
|
fields=list(update_values.keys()))
|
|
|
|
return TenantResponse.from_orm(updated_tenant)
|
|
|
|
# No updates to apply - get current tenant data
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
tenant = await self.tenant_repo.get_by_id(tenant_id)
|
|
return TenantResponse.from_orm(tenant)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Error updating tenant",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to update tenant"
|
|
)
|
|
|
|
async def add_team_member(
|
|
self,
|
|
tenant_id: str,
|
|
user_id: str,
|
|
role: str,
|
|
invited_by: str,
|
|
session: AsyncSession = None
|
|
) -> TenantMemberResponse:
|
|
"""Add a team member to tenant with enhanced validation"""
|
|
|
|
try:
|
|
# Verify inviter has admin access
|
|
access = await self.verify_user_access(invited_by, tenant_id)
|
|
if not access.has_access or access.role not in ["owner", "admin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Insufficient permissions to add team members"
|
|
)
|
|
|
|
# Create membership using repository
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
|
|
membership_data = {
|
|
"tenant_id": tenant_id,
|
|
"user_id": user_id,
|
|
"role": role,
|
|
"invited_by": invited_by,
|
|
"is_active": True
|
|
}
|
|
|
|
member = await self.member_repo.create_membership(membership_data)
|
|
|
|
# Publish event
|
|
try:
|
|
await publish_member_added(tenant_id, user_id, role)
|
|
except Exception as e:
|
|
logger.warning("Failed to publish member added event", error=str(e))
|
|
|
|
logger.info("Team member added successfully",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
role=role,
|
|
invited_by=invited_by)
|
|
|
|
return TenantMemberResponse.from_orm(member)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except (ValidationError, DuplicateRecordError) as e:
|
|
logger.error("Validation error adding team member",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error("Error adding team member",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to add team member"
|
|
)
|
|
|
|
async def get_team_members(
|
|
self,
|
|
tenant_id: str,
|
|
user_id: str,
|
|
active_only: bool = True
|
|
) -> List[TenantMemberResponse]:
|
|
"""Get all team members for a tenant"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as session:
|
|
# Initialize repositories with session
|
|
await self._init_repositories(session)
|
|
|
|
members = await self.member_repo.get_tenant_members(
|
|
tenant_id, active_only=active_only
|
|
)
|
|
|
|
return [TenantMemberResponse.from_orm(member) for member in members]
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Error getting team members",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
error=str(e))
|
|
return []
|
|
|
|
async def update_member_role(
|
|
self,
|
|
tenant_id: str,
|
|
member_user_id: str,
|
|
new_role: str,
|
|
updated_by: str,
|
|
session: AsyncSession = None
|
|
) -> TenantMemberResponse:
|
|
"""Update team member role"""
|
|
|
|
try:
|
|
# Verify updater has admin access
|
|
access = await self.verify_user_access(updated_by, tenant_id)
|
|
if not access.has_access or access.role not in ["owner", "admin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Insufficient permissions to update member roles"
|
|
)
|
|
|
|
updated_member = await self.member_repo.update_member_role(
|
|
tenant_id, member_user_id, new_role, updated_by
|
|
)
|
|
|
|
if not updated_member:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Member not found"
|
|
)
|
|
|
|
return TenantMemberResponse.from_orm(updated_member)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except (ValidationError, DuplicateRecordError) as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error("Error updating member role",
|
|
tenant_id=tenant_id,
|
|
member_user_id=member_user_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to update member role"
|
|
)
|
|
|
|
async def remove_team_member(
|
|
self,
|
|
tenant_id: str,
|
|
member_user_id: str,
|
|
removed_by: str,
|
|
session: AsyncSession = None
|
|
) -> bool:
|
|
"""Remove team member from tenant"""
|
|
|
|
try:
|
|
# Verify remover has admin access
|
|
access = await self.verify_user_access(removed_by, tenant_id)
|
|
if not access.has_access or access.role not in ["owner", "admin"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Insufficient permissions to remove team members"
|
|
)
|
|
|
|
removed_member = await self.member_repo.deactivate_membership(
|
|
tenant_id, member_user_id, removed_by
|
|
)
|
|
|
|
if not removed_member:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Member not found"
|
|
)
|
|
|
|
return True
|
|
|
|
except HTTPException:
|
|
raise
|
|
except ValidationError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
except Exception as e:
|
|
logger.error("Error removing team member",
|
|
tenant_id=tenant_id,
|
|
member_user_id=member_user_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to remove team member"
|
|
)
|
|
|
|
async def update_model_status(
|
|
self,
|
|
tenant_id: str,
|
|
ml_model_trained: bool,
|
|
user_id: str,
|
|
last_training_date: datetime = None
|
|
) -> TenantResponse:
|
|
"""Update tenant model training status"""
|
|
|
|
try:
|
|
# Verify user has access
|
|
access = await self.verify_user_access(user_id, tenant_id)
|
|
if not access.has_access:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied to tenant"
|
|
)
|
|
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
updated_tenant = await self.tenant_repo.update_tenant_model_status(
|
|
tenant_id, ml_model_trained, last_training_date
|
|
)
|
|
|
|
if not updated_tenant:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Tenant not found"
|
|
)
|
|
|
|
return TenantResponse.from_orm(updated_tenant)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Error updating model status",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to update model status"
|
|
)
|
|
|
|
async def get_tenant_statistics(self) -> Dict[str, Any]:
|
|
"""Get comprehensive tenant statistics"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
# Get tenant statistics
|
|
tenant_stats = await self.tenant_repo.get_tenant_statistics()
|
|
|
|
# Get subscription statistics
|
|
subscription_stats = await self.subscription_repo.get_subscription_statistics()
|
|
|
|
return {
|
|
"tenants": tenant_stats,
|
|
"subscriptions": subscription_stats
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting tenant statistics", error=str(e))
|
|
return {
|
|
"tenants": {},
|
|
"subscriptions": {}
|
|
}
|
|
|
|
async def get_tenants_near_location(
|
|
self,
|
|
latitude: float,
|
|
longitude: float,
|
|
radius_km: float = 10.0,
|
|
limit: int = 50
|
|
) -> List[TenantResponse]:
|
|
"""Get tenants near a geographic location"""
|
|
|
|
try:
|
|
async with self.database_manager.get_session() as db_session:
|
|
await self._init_repositories(db_session)
|
|
tenants = await self.tenant_repo.get_tenants_by_location(
|
|
latitude, longitude, radius_km, limit
|
|
)
|
|
|
|
return [TenantResponse.from_orm(tenant) for tenant in tenants]
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting tenants by location",
|
|
latitude=latitude,
|
|
longitude=longitude,
|
|
error=str(e))
|
|
return []
|
|
|
|
async def deactivate_tenant(
|
|
self,
|
|
tenant_id: str,
|
|
user_id: str,
|
|
session: AsyncSession = None
|
|
) -> bool:
|
|
"""Deactivate a tenant (admin only)"""
|
|
|
|
try:
|
|
# Verify user is owner
|
|
access = await self.verify_user_access(user_id, tenant_id)
|
|
if not access.has_access or access.role != "owner":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only tenant owner can deactivate tenant"
|
|
)
|
|
|
|
deactivated_tenant = await self.tenant_repo.deactivate_tenant(tenant_id)
|
|
|
|
if not deactivated_tenant:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Tenant not found"
|
|
)
|
|
|
|
# Also suspend subscription
|
|
subscription = await self.subscription_repo.get_active_subscription(tenant_id)
|
|
if subscription:
|
|
await self.subscription_repo.suspend_subscription(
|
|
str(subscription.id),
|
|
"Tenant deactivated"
|
|
)
|
|
|
|
logger.info("Tenant deactivated",
|
|
tenant_id=tenant_id,
|
|
deactivated_by=user_id)
|
|
|
|
return True
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Error deactivating tenant",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to deactivate tenant"
|
|
)
|
|
|
|
async def activate_tenant(
|
|
self,
|
|
tenant_id: str,
|
|
user_id: str,
|
|
session: AsyncSession = None
|
|
) -> bool:
|
|
"""Activate a previously deactivated tenant (admin only)"""
|
|
|
|
try:
|
|
# Verify user is owner
|
|
access = await self.verify_user_access(user_id, tenant_id)
|
|
if not access.has_access or access.role != "owner":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only tenant owner can activate tenant"
|
|
)
|
|
|
|
activated_tenant = await self.tenant_repo.activate_tenant(tenant_id)
|
|
|
|
if not activated_tenant:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Tenant not found"
|
|
)
|
|
|
|
# Also reactivate subscription if exists
|
|
subscription = await self.subscription_repo.get_subscription_by_tenant(tenant_id)
|
|
if subscription and subscription.status == "suspended":
|
|
await self.subscription_repo.reactivate_subscription(str(subscription.id))
|
|
|
|
logger.info("Tenant activated",
|
|
tenant_id=tenant_id,
|
|
activated_by=user_id)
|
|
|
|
return True
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Error activating tenant",
|
|
tenant_id=tenant_id,
|
|
error=str(e))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to activate tenant"
|
|
)
|
|
|
|
|
|
# Legacy compatibility alias
|
|
TenantService = EnhancedTenantService
|