Samir Boulahtit
7c43d6f4a2
Eliminate all 103 errors and 96 warnings from the architecture validator: Phase 1 - Validator rules & YAML: - Add NAM-001/NAM-002 exceptions for module-scoped router/service files - Fix API-004 to detect # public comments on decorator lines - Add module-specific exception bases to EXC-004 valid_bases - Exclude storefront files from AUTH-004 store context check - Add SVC-006 exceptions for loyalty service atomic commits - Fix _get_rule() to search naming_rules and auth_rules categories - Use plain # CODE comments instead of # noqa: CODE for custom rules Phase 2 - Billing module (5 route files): - Move _resolve_store_to_merchant to subscription_service - Move tier/feature queries to feature_service, admin_subscription_service - Extract 22 inline Pydantic schemas to billing/schemas/billing.py - Replace all HTTPException with domain exceptions Phase 3 - Loyalty module (4 routes + points_service): - Add 7 domain exceptions (Apple auth, enrollment, device registration) - Add service methods to card_service, program_service, apple_wallet_service - Move all db.query() from routes to service layer - Fix SVC-001: replace HTTPException in points_service with domain exception Phase 4 - Remaining modules: - tenancy: move store stats queries to admin_service - cms: move platform resolution to content_page_service, add NoPlatformSubscriptionException - messaging: move user/customer lookups to messaging_service - Add ConfigDict(from_attributes=True) to ContentPageResponse Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
847 lines
28 KiB
Python
847 lines
28 KiB
Python
# app/modules/tenancy/services/admin_service.py
|
|
"""
|
|
Admin service for managing users and stores.
|
|
|
|
This module provides classes and functions for:
|
|
- User management and status control
|
|
- Store creation with owner user generation
|
|
- Store verification and activation
|
|
- Platform statistics
|
|
|
|
Note: Marketplace import job monitoring has been moved to the marketplace module.
|
|
"""
|
|
|
|
import logging
|
|
import secrets
|
|
import string
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.exceptions import ValidationException
|
|
from app.modules.tenancy.exceptions import (
|
|
AdminOperationException,
|
|
CannotModifySelfException,
|
|
StoreAlreadyExistsException,
|
|
StoreNotFoundException,
|
|
StoreVerificationException,
|
|
UserAlreadyExistsException,
|
|
UserCannotBeDeletedException,
|
|
UserNotFoundException,
|
|
UserRoleChangeException,
|
|
UserStatusChangeException,
|
|
)
|
|
from app.modules.tenancy.models import Merchant, Platform, Role, Store, User
|
|
from app.modules.tenancy.schemas.store import StoreCreate
|
|
from middleware.auth import AuthManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AdminService:
|
|
"""Service class for admin operations following the application's service pattern."""
|
|
|
|
# ============================================================================
|
|
# USER MANAGEMENT
|
|
# ============================================================================
|
|
|
|
def get_all_users(self, db: Session, skip: int = 0, limit: int = 100) -> list[User]:
|
|
"""Get paginated list of all users."""
|
|
try:
|
|
return db.query(User).offset(skip).limit(limit).all()
|
|
except Exception as e:
|
|
logger.error(f"Failed to retrieve users: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="get_all_users", reason="Database query failed"
|
|
)
|
|
|
|
def toggle_user_status(
|
|
self, db: Session, user_id: int, current_admin_id: int
|
|
) -> tuple[User, str]:
|
|
"""Toggle user active status."""
|
|
user = self._get_user_by_id_or_raise(db, user_id)
|
|
|
|
# Prevent self-modification
|
|
if user.id == current_admin_id:
|
|
raise CannotModifySelfException(user_id, "deactivate account")
|
|
|
|
# Check if user is another admin
|
|
if user.role == "admin" and user.id != current_admin_id:
|
|
raise UserStatusChangeException(
|
|
user_id=user_id,
|
|
current_status="admin",
|
|
attempted_action="toggle status",
|
|
reason="Cannot modify another admin user",
|
|
)
|
|
|
|
try:
|
|
original_status = user.is_active
|
|
user.is_active = not user.is_active
|
|
user.updated_at = datetime.now(UTC)
|
|
db.flush()
|
|
db.refresh(user)
|
|
|
|
status_action = "activated" if user.is_active else "deactivated"
|
|
message = f"User {user.username} has been {status_action}"
|
|
|
|
logger.info(f"{message} by admin {current_admin_id}")
|
|
return user, message
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to toggle user {user_id} status: {str(e)}")
|
|
raise UserStatusChangeException(
|
|
user_id=user_id,
|
|
current_status="active" if original_status else "inactive",
|
|
attempted_action="toggle status",
|
|
reason="Database update failed",
|
|
)
|
|
|
|
def list_users(
|
|
self,
|
|
db: Session,
|
|
page: int = 1,
|
|
per_page: int = 10,
|
|
search: str | None = None,
|
|
role: str | None = None,
|
|
scope: str | None = None,
|
|
is_active: bool | None = None,
|
|
) -> tuple[list[User], int, int]:
|
|
"""
|
|
Get paginated list of users with filtering.
|
|
|
|
Args:
|
|
scope: Optional scope filter. 'merchant' returns users who are
|
|
merchant owners or store team members.
|
|
|
|
Returns:
|
|
Tuple of (users, total_count, total_pages)
|
|
"""
|
|
import math
|
|
|
|
from app.modules.tenancy.models import Merchant, StoreUser
|
|
|
|
query = db.query(User)
|
|
|
|
# Apply scope filter
|
|
if scope == "merchant":
|
|
owner_ids = db.query(Merchant.owner_user_id).distinct()
|
|
team_ids = db.query(StoreUser.user_id).distinct()
|
|
query = query.filter(User.id.in_(owner_ids.union(team_ids)))
|
|
|
|
# Apply filters
|
|
if search:
|
|
search_term = f"%{search.lower()}%"
|
|
query = query.filter(
|
|
or_(
|
|
User.username.ilike(search_term),
|
|
User.email.ilike(search_term),
|
|
User.first_name.ilike(search_term),
|
|
User.last_name.ilike(search_term),
|
|
)
|
|
)
|
|
|
|
if role:
|
|
query = query.filter(User.role == role)
|
|
|
|
if is_active is not None:
|
|
query = query.filter(User.is_active == is_active)
|
|
|
|
# Get total count
|
|
total = query.count()
|
|
pages = math.ceil(total / per_page) if total > 0 else 1
|
|
|
|
# Apply pagination
|
|
skip = (page - 1) * per_page
|
|
users = (
|
|
query.order_by(User.created_at.desc()).offset(skip).limit(per_page).all()
|
|
)
|
|
|
|
return users, total, pages
|
|
|
|
def create_user(
|
|
self,
|
|
db: Session,
|
|
email: str,
|
|
username: str,
|
|
password: str,
|
|
first_name: str | None = None,
|
|
last_name: str | None = None,
|
|
role: str = "customer",
|
|
current_admin_id: int | None = None,
|
|
) -> User:
|
|
"""
|
|
Create a new user.
|
|
|
|
Raises:
|
|
UserAlreadyExistsException: If email or username already exists
|
|
"""
|
|
# Check if email exists
|
|
if db.query(User).filter(User.email == email).first():
|
|
raise UserAlreadyExistsException("Email already registered", field="email")
|
|
|
|
# Check if username exists
|
|
if db.query(User).filter(User.username == username).first():
|
|
raise UserAlreadyExistsException("Username already taken", field="username")
|
|
|
|
# Create user
|
|
auth_manager = AuthManager()
|
|
user = User(
|
|
email=email,
|
|
username=username,
|
|
hashed_password=auth_manager.hash_password(password),
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
role=role,
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
db.refresh(user)
|
|
|
|
logger.info(f"Admin {current_admin_id} created user {user.username}")
|
|
return user
|
|
|
|
def get_user_details(self, db: Session, user_id: int) -> User:
|
|
"""
|
|
Get user with relationships loaded.
|
|
|
|
Raises:
|
|
UserNotFoundException: If user not found
|
|
"""
|
|
user = (
|
|
db.query(User)
|
|
.options(
|
|
joinedload(User.owned_merchants), joinedload(User.store_memberships)
|
|
)
|
|
.filter(User.id == user_id)
|
|
.first()
|
|
)
|
|
|
|
if not user:
|
|
raise UserNotFoundException(str(user_id))
|
|
|
|
return user
|
|
|
|
def update_user(
|
|
self,
|
|
db: Session,
|
|
user_id: int,
|
|
current_admin_id: int,
|
|
email: str | None = None,
|
|
username: str | None = None,
|
|
first_name: str | None = None,
|
|
last_name: str | None = None,
|
|
role: str | None = None,
|
|
is_active: bool | None = None,
|
|
) -> User:
|
|
"""
|
|
Update user information.
|
|
|
|
Raises:
|
|
UserNotFoundException: If user not found
|
|
UserAlreadyExistsException: If email/username already taken
|
|
UserRoleChangeException: If trying to change own admin role
|
|
"""
|
|
user = self._get_user_by_id_or_raise(db, user_id)
|
|
|
|
# Prevent changing own admin status
|
|
if user.id == current_admin_id and role and role != "admin":
|
|
raise UserRoleChangeException(
|
|
user_id=user_id,
|
|
current_role=user.role,
|
|
target_role=role,
|
|
reason="Cannot change your own admin role",
|
|
)
|
|
|
|
# Check email uniqueness if changing
|
|
if email and email != user.email:
|
|
if db.query(User).filter(User.email == email).first():
|
|
raise UserAlreadyExistsException(
|
|
"Email already registered", field="email"
|
|
)
|
|
|
|
# Check username uniqueness if changing
|
|
if username and username != user.username:
|
|
if db.query(User).filter(User.username == username).first():
|
|
raise UserAlreadyExistsException(
|
|
"Username already taken", field="username"
|
|
)
|
|
|
|
# Update fields
|
|
if email is not None:
|
|
user.email = email
|
|
if username is not None:
|
|
user.username = username
|
|
if first_name is not None:
|
|
user.first_name = first_name
|
|
if last_name is not None:
|
|
user.last_name = last_name
|
|
if role is not None:
|
|
user.role = role
|
|
if is_active is not None:
|
|
user.is_active = is_active
|
|
|
|
user.updated_at = datetime.now(UTC)
|
|
db.flush()
|
|
db.refresh(user)
|
|
|
|
logger.info(f"Admin {current_admin_id} updated user {user.username}")
|
|
return user
|
|
|
|
def delete_user(self, db: Session, user_id: int, current_admin_id: int) -> str:
|
|
"""
|
|
Delete a user.
|
|
|
|
Raises:
|
|
UserNotFoundException: If user not found
|
|
CannotModifySelfException: If trying to delete yourself
|
|
UserCannotBeDeletedException: If user owns merchants
|
|
"""
|
|
user = (
|
|
db.query(User)
|
|
.options(joinedload(User.owned_merchants))
|
|
.filter(User.id == user_id)
|
|
.first()
|
|
)
|
|
|
|
if not user:
|
|
raise UserNotFoundException(str(user_id))
|
|
|
|
# Prevent deleting yourself
|
|
if user.id == current_admin_id:
|
|
raise CannotModifySelfException(user_id, "delete account")
|
|
|
|
# Prevent deleting users who own merchants
|
|
if user.owned_merchants:
|
|
raise UserCannotBeDeletedException(
|
|
user_id=user_id,
|
|
reason=f"User owns {len(user.owned_merchants)} merchant(ies). Transfer ownership first.",
|
|
owned_count=len(user.owned_merchants),
|
|
)
|
|
|
|
username = user.username
|
|
db.delete(user)
|
|
|
|
logger.info(f"Admin {current_admin_id} deleted user {username}")
|
|
return f"User {username} deleted successfully"
|
|
|
|
def search_users(
|
|
self,
|
|
db: Session,
|
|
query: str,
|
|
limit: int = 10,
|
|
) -> list[dict]:
|
|
"""
|
|
Search users by username or email.
|
|
|
|
Used for autocomplete in ownership transfer.
|
|
"""
|
|
search_term = f"%{query.lower()}%"
|
|
users = (
|
|
db.query(User)
|
|
.filter(
|
|
or_(User.username.ilike(search_term), User.email.ilike(search_term))
|
|
)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
return [
|
|
{
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"is_active": user.is_active,
|
|
}
|
|
for user in users
|
|
]
|
|
|
|
# ============================================================================
|
|
# STORE MANAGEMENT
|
|
# ============================================================================
|
|
|
|
def create_store(self, db: Session, store_data: StoreCreate) -> Store:
|
|
"""
|
|
Create a store (storefront/brand) under an existing merchant.
|
|
|
|
The store inherits owner and contact information from its parent merchant.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_data: Store creation data including merchant_id
|
|
|
|
Returns:
|
|
The created Store object with merchant relationship loaded
|
|
|
|
Raises:
|
|
ValidationException: If merchant not found or store code/subdomain exists
|
|
AdminOperationException: If creation fails
|
|
"""
|
|
try:
|
|
# Validate merchant exists
|
|
merchant = (
|
|
db.query(Merchant).filter(Merchant.id == store_data.merchant_id).first()
|
|
)
|
|
if not merchant:
|
|
raise ValidationException(
|
|
f"Merchant with ID {store_data.merchant_id} not found"
|
|
)
|
|
|
|
# Check if store code already exists
|
|
existing_store = (
|
|
db.query(Store)
|
|
.filter(
|
|
func.upper(Store.store_code) == store_data.store_code.upper()
|
|
)
|
|
.first()
|
|
)
|
|
if existing_store:
|
|
raise StoreAlreadyExistsException(store_data.store_code)
|
|
|
|
# Check if subdomain already exists
|
|
existing_subdomain = (
|
|
db.query(Store)
|
|
.filter(func.lower(Store.subdomain) == store_data.subdomain.lower())
|
|
.first()
|
|
)
|
|
if existing_subdomain:
|
|
raise ValidationException(
|
|
f"Subdomain '{store_data.subdomain}' is already taken"
|
|
)
|
|
|
|
# Create store linked to merchant
|
|
store = Store(
|
|
merchant_id=merchant.id,
|
|
store_code=store_data.store_code.upper(),
|
|
subdomain=store_data.subdomain.lower(),
|
|
name=store_data.name,
|
|
description=store_data.description,
|
|
letzshop_csv_url_fr=store_data.letzshop_csv_url_fr,
|
|
letzshop_csv_url_en=store_data.letzshop_csv_url_en,
|
|
letzshop_csv_url_de=store_data.letzshop_csv_url_de,
|
|
is_active=True,
|
|
is_verified=False, # Needs verification by admin
|
|
)
|
|
db.add(store)
|
|
db.flush() # Get store.id
|
|
|
|
# Create default roles for store
|
|
self._create_default_roles(db, store.id)
|
|
|
|
# Assign store to platforms if provided
|
|
if store_data.platform_ids:
|
|
from app.modules.tenancy.models import StorePlatform
|
|
|
|
for platform_id in store_data.platform_ids:
|
|
# Verify platform exists
|
|
platform = db.query(Platform).filter(Platform.id == platform_id).first()
|
|
if platform:
|
|
store_platform = StorePlatform(
|
|
store_id=store.id,
|
|
platform_id=platform_id,
|
|
is_active=True,
|
|
)
|
|
db.add(store_platform)
|
|
logger.debug(
|
|
f"Assigned store {store.store_code} to platform {platform.code}"
|
|
)
|
|
|
|
db.flush()
|
|
db.refresh(store)
|
|
|
|
logger.info(
|
|
f"Store {store.store_code} created under merchant {merchant.name} (ID: {merchant.id})"
|
|
)
|
|
|
|
return store
|
|
|
|
except (StoreAlreadyExistsException, ValidationException):
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to create store: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="create_store",
|
|
reason=f"Failed to create store: {str(e)}",
|
|
)
|
|
|
|
def get_all_stores(
|
|
self,
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
search: str | None = None,
|
|
is_active: bool | None = None,
|
|
is_verified: bool | None = None,
|
|
) -> tuple[list[Store], int]:
|
|
"""Get paginated list of all stores with filtering."""
|
|
try:
|
|
# Eagerly load merchant relationship to avoid N+1 queries
|
|
query = db.query(Store).options(joinedload(Store.merchant))
|
|
|
|
# Apply search filter
|
|
if search:
|
|
search_term = f"%{search}%"
|
|
query = query.filter(
|
|
or_(
|
|
Store.name.ilike(search_term),
|
|
Store.store_code.ilike(search_term),
|
|
Store.subdomain.ilike(search_term),
|
|
)
|
|
)
|
|
|
|
# Apply status filters
|
|
if is_active is not None:
|
|
query = query.filter(Store.is_active == is_active)
|
|
if is_verified is not None:
|
|
query = query.filter(Store.is_verified == is_verified)
|
|
|
|
# Get total count (without joinedload for performance)
|
|
count_query = db.query(Store)
|
|
if search:
|
|
search_term = f"%{search}%"
|
|
count_query = count_query.filter(
|
|
or_(
|
|
Store.name.ilike(search_term),
|
|
Store.store_code.ilike(search_term),
|
|
Store.subdomain.ilike(search_term),
|
|
)
|
|
)
|
|
if is_active is not None:
|
|
count_query = count_query.filter(Store.is_active == is_active)
|
|
if is_verified is not None:
|
|
count_query = count_query.filter(Store.is_verified == is_verified)
|
|
total = count_query.count()
|
|
|
|
# Get paginated results
|
|
stores = query.offset(skip).limit(limit).all()
|
|
|
|
return stores, total
|
|
except Exception as e:
|
|
logger.error(f"Failed to retrieve stores: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="get_all_stores", reason="Database query failed"
|
|
)
|
|
|
|
def get_store_by_id(self, db: Session, store_id: int) -> Store:
|
|
"""Get store by ID."""
|
|
return self._get_store_by_id_or_raise(db, store_id)
|
|
|
|
def verify_store(self, db: Session, store_id: int) -> tuple[Store, str]:
|
|
"""Toggle store verification status."""
|
|
store = self._get_store_by_id_or_raise(db, store_id)
|
|
|
|
try:
|
|
original_status = store.is_verified
|
|
store.is_verified = not store.is_verified
|
|
store.updated_at = datetime.now(UTC)
|
|
|
|
if store.is_verified:
|
|
store.verified_at = datetime.now(UTC)
|
|
|
|
db.flush()
|
|
db.refresh(store)
|
|
|
|
status_action = "verified" if store.is_verified else "unverified"
|
|
message = f"Store {store.store_code} has been {status_action}"
|
|
|
|
logger.info(message)
|
|
return store, message
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to verify store {store_id}: {str(e)}")
|
|
raise StoreVerificationException(
|
|
store_id=store_id,
|
|
reason="Database update failed",
|
|
current_verification_status=original_status,
|
|
)
|
|
|
|
def toggle_store_status(self, db: Session, store_id: int) -> tuple[Store, str]:
|
|
"""Toggle store active status."""
|
|
store = self._get_store_by_id_or_raise(db, store_id)
|
|
|
|
try:
|
|
store.is_active = not store.is_active
|
|
store.updated_at = datetime.now(UTC)
|
|
db.flush()
|
|
db.refresh(store)
|
|
|
|
status_action = "activated" if store.is_active else "deactivated"
|
|
message = f"Store {store.store_code} has been {status_action}"
|
|
|
|
logger.info(message)
|
|
return store, message
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to toggle store {store_id} status: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="toggle_store_status",
|
|
reason="Database update failed",
|
|
target_type="store",
|
|
target_id=str(store_id),
|
|
)
|
|
|
|
def delete_store(self, db: Session, store_id: int) -> str:
|
|
"""Delete store and all associated data."""
|
|
store = self._get_store_by_id_or_raise(db, store_id)
|
|
|
|
try:
|
|
store_code = store.store_code
|
|
|
|
# TODO: Delete associated data in correct order
|
|
# - Delete orders
|
|
# - Delete customers
|
|
# - Delete products
|
|
# - Delete team members
|
|
# - Delete roles
|
|
# - Delete import jobs
|
|
|
|
db.delete(store)
|
|
|
|
logger.warning(f"Store {store_code} and all associated data deleted")
|
|
return f"Store {store_code} successfully deleted"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete store {store_id}: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="delete_store", reason="Database deletion failed"
|
|
)
|
|
|
|
def update_store(
|
|
self,
|
|
db: Session,
|
|
store_id: int,
|
|
store_update, # StoreUpdate schema
|
|
) -> Store:
|
|
"""
|
|
Update store information (Admin only).
|
|
|
|
Can update:
|
|
- Store details (name, description, subdomain)
|
|
- Business contact info (contact_email, phone, etc.)
|
|
- Status (is_active, is_verified)
|
|
|
|
Cannot update:
|
|
- store_code (immutable)
|
|
- merchant_id (store cannot be moved between merchants)
|
|
|
|
Note: Ownership is managed at the Merchant level.
|
|
Use merchant_service.transfer_ownership() for ownership changes.
|
|
|
|
Args:
|
|
db: Database session
|
|
store_id: ID of store to update
|
|
store_update: StoreUpdate schema with updated data
|
|
|
|
Returns:
|
|
Updated store object
|
|
|
|
Raises:
|
|
StoreNotFoundException: If store not found
|
|
ValidationException: If subdomain already taken
|
|
"""
|
|
store = self._get_store_by_id_or_raise(db, store_id)
|
|
|
|
try:
|
|
# Get update data
|
|
update_data = store_update.model_dump(exclude_unset=True)
|
|
|
|
# Handle reset_contact_to_merchant flag
|
|
if update_data.pop("reset_contact_to_merchant", False):
|
|
# Reset all contact fields to None (inherit from merchant)
|
|
update_data["contact_email"] = None
|
|
update_data["contact_phone"] = None
|
|
update_data["website"] = None
|
|
update_data["business_address"] = None
|
|
update_data["tax_number"] = None
|
|
|
|
# Convert empty strings to None for contact fields (empty = inherit)
|
|
contact_fields = [
|
|
"contact_email",
|
|
"contact_phone",
|
|
"website",
|
|
"business_address",
|
|
"tax_number",
|
|
]
|
|
for field in contact_fields:
|
|
if field in update_data and update_data[field] == "":
|
|
update_data[field] = None
|
|
|
|
# Check subdomain uniqueness if changing
|
|
if (
|
|
"subdomain" in update_data
|
|
and update_data["subdomain"] != store.subdomain
|
|
):
|
|
existing = (
|
|
db.query(Store)
|
|
.filter(
|
|
Store.subdomain == update_data["subdomain"],
|
|
Store.id != store_id,
|
|
)
|
|
.first()
|
|
)
|
|
if existing:
|
|
raise ValidationException(
|
|
f"Subdomain '{update_data['subdomain']}' is already taken"
|
|
)
|
|
|
|
# Update store fields
|
|
for field, value in update_data.items():
|
|
setattr(store, field, value)
|
|
|
|
store.updated_at = datetime.now(UTC)
|
|
|
|
db.flush()
|
|
db.refresh(store)
|
|
|
|
logger.info(
|
|
f"Store {store_id} ({store.store_code}) updated by admin. "
|
|
f"Fields updated: {', '.join(update_data.keys())}"
|
|
)
|
|
return store
|
|
|
|
except ValidationException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to update store {store_id}: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="update_store", reason=f"Database update failed: {str(e)}"
|
|
)
|
|
|
|
# NOTE: Store ownership transfer is now handled at the Merchant level.
|
|
# Use merchant_service.transfer_ownership() instead.
|
|
|
|
# NOTE: Marketplace import job operations have been moved to the marketplace module.
|
|
# Use app.modules.marketplace routes for import job management.
|
|
|
|
# ============================================================================
|
|
# STATISTICS
|
|
# ============================================================================
|
|
|
|
def get_store_statistics(self, db: Session) -> dict:
|
|
"""
|
|
Get store statistics for admin dashboard.
|
|
|
|
Returns:
|
|
Dict with total, verified, pending, and inactive counts.
|
|
"""
|
|
try:
|
|
total = db.query(Store).count()
|
|
verified = db.query(Store).filter(Store.is_verified == True).count() # noqa: E712
|
|
active = db.query(Store).filter(Store.is_active == True).count() # noqa: E712
|
|
inactive = total - active
|
|
pending = db.query(Store).filter(
|
|
Store.is_active == True, Store.is_verified == False # noqa: E712
|
|
).count()
|
|
|
|
return {
|
|
"total": total,
|
|
"verified": verified,
|
|
"pending": pending,
|
|
"inactive": inactive,
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Failed to get store statistics: {str(e)}")
|
|
raise AdminOperationException(
|
|
operation="get_store_statistics", reason="Database query failed"
|
|
)
|
|
|
|
def get_recent_stores(self, db: Session, limit: int = 5) -> list[dict]:
|
|
"""Get recently created stores."""
|
|
try:
|
|
stores = (
|
|
db.query(Store).order_by(Store.created_at.desc()).limit(limit).all()
|
|
)
|
|
|
|
return [
|
|
{
|
|
"id": v.id,
|
|
"store_code": v.store_code,
|
|
"name": v.name,
|
|
"subdomain": v.subdomain,
|
|
"is_active": v.is_active,
|
|
"is_verified": v.is_verified,
|
|
"created_at": v.created_at,
|
|
}
|
|
for v in stores
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"Failed to get recent stores: {str(e)}")
|
|
return []
|
|
|
|
# NOTE: get_recent_import_jobs has been moved to the marketplace module
|
|
|
|
# ============================================================================
|
|
# PRIVATE HELPER METHODS
|
|
# ============================================================================
|
|
|
|
def _get_user_by_id_or_raise(self, db: Session, user_id: int) -> User:
|
|
"""Get user by ID or raise UserNotFoundException."""
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise UserNotFoundException(str(user_id))
|
|
return user
|
|
|
|
def _get_store_by_id_or_raise(self, db: Session, store_id: int) -> Store:
|
|
"""Get store by ID or raise StoreNotFoundException."""
|
|
store = (
|
|
db.query(Store)
|
|
.options(joinedload(Store.merchant).joinedload(Merchant.owner))
|
|
.filter(Store.id == store_id)
|
|
.first()
|
|
)
|
|
if not store:
|
|
raise StoreNotFoundException(str(store_id), identifier_type="id")
|
|
return store
|
|
|
|
def _generate_temp_password(self, length: int = 12) -> str:
|
|
"""Generate secure temporary password."""
|
|
alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
|
|
return "".join(secrets.choice(alphabet) for _ in range(length))
|
|
|
|
def _create_default_roles(self, db: Session, store_id: int):
|
|
"""Create default roles for a new store."""
|
|
default_roles = [
|
|
{"name": "Owner", "permissions": ["*"]}, # Full access
|
|
{
|
|
"name": "Manager",
|
|
"permissions": [
|
|
"products.*",
|
|
"orders.*",
|
|
"customers.view",
|
|
"inventory.*",
|
|
"team.view",
|
|
],
|
|
},
|
|
{
|
|
"name": "Editor",
|
|
"permissions": [
|
|
"products.view",
|
|
"products.edit",
|
|
"orders.view",
|
|
"inventory.view",
|
|
],
|
|
},
|
|
{
|
|
"name": "Viewer",
|
|
"permissions": [
|
|
"products.view",
|
|
"orders.view",
|
|
"customers.view",
|
|
"inventory.view",
|
|
],
|
|
},
|
|
]
|
|
|
|
for role_data in default_roles:
|
|
role = Role(
|
|
store_id=store_id,
|
|
name=role_data["name"],
|
|
permissions=role_data["permissions"],
|
|
)
|
|
db.add(role)
|
|
|
|
|
|
# Create service instance
|
|
admin_service = AdminService()
|