Samir Boulahtit
86e85a98b8
Some checks failed
Enforce MOD-025/MOD-026 rules: zero top-level cross-module model imports remain in any service file. All 66 files migrated using deferred import patterns (method-body, _get_model() helpers, instance-cached self._Model) and new cross-module service methods in tenancy. Documentation updated with Pattern 6 (deferred imports), migration plan marked complete, and violations status reflects 84→0 service-layer violations. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
305 lines
8.5 KiB
Python
305 lines
8.5 KiB
Python
# app/modules/tenancy/services/team_service.py
|
|
"""
|
|
Team service for store team management.
|
|
|
|
This module provides:
|
|
- Team member invitation
|
|
- Role management
|
|
- Team member CRUD operations
|
|
"""
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.tenancy.exceptions import (
|
|
TeamMemberNotFoundException,
|
|
TeamValidationException,
|
|
)
|
|
from app.modules.tenancy.models import Role, StoreUser, User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TeamService:
|
|
"""Service for team management operations."""
|
|
|
|
def get_team_members(
|
|
self, db: Session, store_id: int, current_user: User
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Get all team members for store.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
current_user: Current user
|
|
|
|
Returns:
|
|
List of team members
|
|
"""
|
|
try:
|
|
store_users = (
|
|
db.query(StoreUser)
|
|
.filter(StoreUser.store_id == store_id, StoreUser.is_active == True)
|
|
.all()
|
|
)
|
|
|
|
members = []
|
|
for vu in store_users:
|
|
members.append(
|
|
{
|
|
"id": vu.user_id,
|
|
"email": vu.user.email,
|
|
"first_name": vu.user.first_name,
|
|
"last_name": vu.user.last_name,
|
|
"role": vu.role.name,
|
|
"role_id": vu.role_id,
|
|
"is_active": vu.is_active,
|
|
"joined_at": vu.created_at,
|
|
}
|
|
)
|
|
|
|
return members
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error getting team members: {str(e)}")
|
|
raise TeamValidationException("Failed to retrieve team members")
|
|
|
|
def invite_team_member(
|
|
self, db: Session, store_id: int, invitation_data: dict, current_user: User
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Invite a new team member.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
invitation_data: Invitation details
|
|
current_user: Current user
|
|
|
|
Returns:
|
|
Invitation result
|
|
"""
|
|
try:
|
|
# TODO: Implement full invitation flow with email
|
|
# For now, return placeholder
|
|
return {
|
|
"message": "Team invitation feature coming soon",
|
|
"email": invitation_data.get("email"),
|
|
"role": invitation_data.get("role"),
|
|
}
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error inviting team member: {str(e)}")
|
|
raise TeamValidationException("Failed to invite team member")
|
|
|
|
def update_team_member(
|
|
self,
|
|
db: Session,
|
|
store_id: int,
|
|
user_id: int,
|
|
update_data: dict,
|
|
current_user: User,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Update team member role or status.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
user_id: User ID to update
|
|
update_data: Update data
|
|
current_user: Current user
|
|
|
|
Returns:
|
|
Updated member info
|
|
"""
|
|
try:
|
|
store_user = (
|
|
db.query(StoreUser)
|
|
.filter(
|
|
StoreUser.store_id == store_id, StoreUser.user_id == user_id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not store_user:
|
|
raise TeamMemberNotFoundException(user_id=user_id, store_id=store_id)
|
|
|
|
# Update fields
|
|
if "role_id" in update_data:
|
|
store_user.role_id = update_data["role_id"]
|
|
|
|
if "is_active" in update_data:
|
|
store_user.is_active = update_data["is_active"]
|
|
|
|
store_user.updated_at = datetime.now(UTC)
|
|
db.flush()
|
|
db.refresh(store_user)
|
|
|
|
return {
|
|
"message": "Team member updated successfully",
|
|
"user_id": user_id,
|
|
}
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error updating team member: {str(e)}")
|
|
raise TeamValidationException("Failed to update team member")
|
|
|
|
def remove_team_member(
|
|
self, db: Session, store_id: int, user_id: int, current_user: User
|
|
) -> bool:
|
|
"""
|
|
Remove team member from store.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
user_id: User ID to remove
|
|
current_user: Current user
|
|
|
|
Returns:
|
|
True if removed
|
|
"""
|
|
try:
|
|
store_user = (
|
|
db.query(StoreUser)
|
|
.filter(
|
|
StoreUser.store_id == store_id, StoreUser.user_id == user_id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not store_user:
|
|
raise TeamMemberNotFoundException(user_id=user_id, store_id=store_id)
|
|
|
|
# Soft delete
|
|
store_user.is_active = False
|
|
store_user.updated_at = datetime.now(UTC)
|
|
|
|
logger.info(f"Removed user {user_id} from store {store_id}")
|
|
return True
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error removing team member: {str(e)}")
|
|
raise TeamValidationException("Failed to remove team member")
|
|
|
|
# ========================================================================
|
|
# Cross-module public API methods
|
|
# ========================================================================
|
|
|
|
def get_store_owner(self, db: Session, store_id: int) -> StoreUser | None:
|
|
"""
|
|
Get the owner StoreUser for a store.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
|
|
Returns:
|
|
StoreUser with is_owner=True, or None
|
|
"""
|
|
return (
|
|
db.query(StoreUser)
|
|
.filter(
|
|
StoreUser.store_id == store_id,
|
|
StoreUser.is_owner == True, # noqa: E712
|
|
)
|
|
.first()
|
|
)
|
|
|
|
def get_active_team_member_count(self, db: Session, store_id: int) -> int:
|
|
"""
|
|
Count active team members for a store.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
|
|
Returns:
|
|
Number of active team members
|
|
"""
|
|
return (
|
|
db.query(func.count(StoreUser.id))
|
|
.filter(
|
|
StoreUser.store_id == store_id,
|
|
StoreUser.is_active == True, # noqa: E712
|
|
)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
def get_store_users_with_user(
|
|
self, db: Session, store_id: int, active_only: bool = True
|
|
) -> list[tuple[User, StoreUser]]:
|
|
"""
|
|
Get User and StoreUser pairs for a store.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
active_only: Only active users
|
|
|
|
Returns:
|
|
List of (User, StoreUser) tuples
|
|
"""
|
|
query = (
|
|
db.query(User, StoreUser)
|
|
.join(StoreUser, User.id == StoreUser.user_id)
|
|
.filter(StoreUser.store_id == store_id)
|
|
)
|
|
if active_only:
|
|
query = query.filter(User.is_active == True) # noqa: E712
|
|
return query.all()
|
|
|
|
def get_store_roles(self, db: Session, store_id: int) -> list[dict[str, Any]]:
|
|
"""
|
|
Get available roles for store.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: Store ID
|
|
|
|
Returns:
|
|
List of roles
|
|
"""
|
|
try:
|
|
roles = db.query(Role).filter(Role.store_id == store_id).all()
|
|
|
|
return [
|
|
{
|
|
"id": role.id,
|
|
"name": role.name,
|
|
"permissions": role.permissions,
|
|
}
|
|
for role in roles
|
|
]
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error getting store roles: {str(e)}")
|
|
raise TeamValidationException("Failed to retrieve roles")
|
|
|
|
|
|
def get_total_active_team_member_count(self, db: Session) -> int:
|
|
"""
|
|
Count active team members across all stores.
|
|
|
|
Returns:
|
|
Total number of active team members platform-wide
|
|
"""
|
|
return (
|
|
db.query(func.count(StoreUser.id))
|
|
.filter(StoreUser.is_active == True) # noqa: E712
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
|
|
# Create service instance
|
|
team_service = TeamService()
|