# #region ProfileUtils [C:2] [TYPE Module] [SEMANTICS string,normalization,utility,sanitize] # @defgroup Services Module group. # @BRIEF Pure utility helpers for profile data sanitization, normalization, and secret masking. # Also contains shared exception classes to avoid circular imports between profile sub-modules. # @LAYER Domain # @RATIONALE Extracted from ProfileService to satisfy INV_7 — module length under 400 lines, # contract length under 150 lines. These are stateless pure functions that do not # require a class or service container. # @REJECTED Keeping these as private methods on ProfileService was rejected because they # inflate the class beyond 150 lines and mix pure computation with I/O-bound # service logic. > from collections.abc import Iterable, Sequence > from datetime import UTC, datetime > from typing import Any # #region ProfileValidationError [C:2] [TYPE Class] # @defgroup Services Module group. # @RELATION INHERITS -> [EXT:Python:Exception] # @BRIEF Domain validation error for profile preference update requests. > class ProfileValidationError(Exception): > def __init__(self, errors: Sequence[str]): > self.errors = list(errors) > super().__init__("Profile preference validation failed") # #endregion ProfileValidationError # #region EnvironmentNotFoundError [C:2] [TYPE Class] # @defgroup Services Module group. # @RELATION INHERITS -> [EXT:Python:Exception] # @BRIEF Raised when environment_id from lookup request is unknown in app configuration. > class EnvironmentNotFoundError(Exception): > pass # #endregion EnvironmentNotFoundError # #region ProfileAuthorizationError [C:2] [TYPE Class] # @defgroup Services Module group. # @RELATION INHERITS -> [EXT:Python:Exception] # @BRIEF Raised when caller attempts cross-user preference mutation. > class ProfileAuthorizationError(Exception): > pass # #endregion ProfileAuthorizationError > SUPPORTED_START_PAGES: set[str] = {"dashboards", "datasets", "reports"} > SUPPORTED_DENSITIES: set[str] = {"compact", "comfortable"} # #region sanitize_text [C:1] [TYPE Function] # @BRIEF Normalize optional text into trimmed form or None. > def sanitize_text(value: str | None) -> str | None: > normalized = str(value or "").strip() > if not normalized: > return None > return normalized # #endregion sanitize_text # #region sanitize_secret [C:1] [TYPE Function] # @BRIEF Normalize secret input into trimmed form or None. > def sanitize_secret(value: str | None) -> str | None: > if value is None: > return None > normalized = str(value).strip() > if not normalized: > return None > return normalized # #endregion sanitize_secret # #region sanitize_username [C:1] [TYPE Function] # @BRIEF Normalize raw username into trimmed form or None for empty input. > def sanitize_username(value: str | None) -> str | None: > return sanitize_text(value) # #endregion sanitize_username # #region normalize_username [C:1] [TYPE Function] # @BRIEF Apply deterministic trim+lower normalization for actor matching. > def normalize_username(value: str | None) -> str | None: > sanitized = sanitize_username(value) > if sanitized is None: > return None > return sanitized.lower() # #endregion normalize_username # #region normalize_start_page [C:1] [TYPE Function] # @BRIEF Normalize supported start page aliases to canonical values. > def normalize_start_page(value: str | None) -> str: > normalized = str(value or "").strip().lower() > if normalized == "reports-logs": > return "reports" > if normalized in SUPPORTED_START_PAGES: > return normalized > return "dashboards" # #endregion normalize_start_page # #region normalize_density [C:1] [TYPE Function] # @BRIEF Normalize supported density aliases to canonical values. > def normalize_density(value: str | None) -> str: > normalized = str(value or "").strip().lower() > if normalized == "free": > return "comfortable" > if normalized in SUPPORTED_DENSITIES: > return normalized > return "comfortable" # #endregion normalize_density # #region mask_secret_value [C:1] [TYPE Function] # @BRIEF Build a safe display value for sensitive secrets. > def mask_secret_value(secret: str | None) -> str | None: > sanitized = sanitize_secret(secret) > if sanitized is None: > return None > if len(sanitized) <= 4: > return "***" > return f"{sanitized[:2]}***{sanitized[-2:]}" # #endregion mask_secret_value # #region validate_update_payload [C:2] [TYPE Function] # @ingroup Services # @BRIEF Validate username/toggle constraints for preference mutation. # @RELATION CALLS -> [sanitize_username] # @RELATION CALLS -> [sanitize_text] # @RELATION DEPENDS_ON -> [ProfileUtils] # @RELATION DEPENDS_ON -> [ProfileUtils] # @POST Returns validation errors list; empty list means valid. > def validate_update_payload( > superset_username: str | None, > show_only_my_dashboards: bool, > git_email: str | None, > start_page: str, > dashboards_table_density: str, > email_address: str | None = None, > ) -> list[str]: > errors: list[str] = [] > sanitized_username = sanitize_username(superset_username) > if sanitized_username and any(ch.isspace() for ch in sanitized_username): > errors.append( > "Username should not contain spaces. Please enter a valid Apache Superset username." > ) > if show_only_my_dashboards and not sanitized_username: > errors.append( > "Superset username is required when default filter is enabled." > ) > sanitized_git_email = sanitize_text(git_email) > if sanitized_git_email: > if ( > " " in sanitized_git_email > or "@" not in sanitized_git_email > or sanitized_git_email.startswith("@") > or sanitized_git_email.endswith("@") > ): > errors.append("Git email should be a valid email address.") > if start_page not in SUPPORTED_START_PAGES: > errors.append("Start page value is not supported.") > if dashboards_table_density not in SUPPORTED_DENSITIES: > errors.append("Dashboards table density value is not supported.") > sanitized_email = sanitize_text(email_address) > if sanitized_email: > if ( > " " in sanitized_email > or "@" not in sanitized_email > or sanitized_email.startswith("@") > or sanitized_email.endswith("@") > ): > errors.append("Notification email should be a valid email address.") > return errors # #endregion validate_update_payload # #region build_default_preference [C:2] [TYPE Function] [SEMANTICS preference,default,unconfigured] # @ingroup Services # @BRIEF Build non-persisted default preference DTO for unconfigured users. # @RELATION DEPENDS_ON -> [ProfilePreference] > def build_default_preference(user_id: str) -> Any: > """Build default ProfilePreference with disabled toggles and empty usernames.""" > from ..schemas.profile import ProfilePreference > now = datetime.now(UTC) > return ProfilePreference( > user_id=str(user_id), > superset_username=None, > superset_username_normalized=None, > show_only_my_dashboards=False, > show_only_slug_dashboards=True, > git_username=None, > git_email=None, > has_git_personal_access_token=False, > git_personal_access_token_masked=None, > start_page="dashboards", > auto_open_task_drawer=True, > dashboards_table_density="comfortable", > telegram_id=None, > email_address=None, > notify_on_fail=True, > created_at=now, > updated_at=now, > ) # #endregion build_default_preference # #region normalize_owner_tokens [C:2] [TYPE Function] # @ingroup Services # @BRIEF Normalize owners payload into deduplicated lower-cased tokens. # @RELATION CALLS -> [normalize_username] > def normalize_owner_tokens(owners: Iterable[Any] | None) -> list[str]: > if owners is None: > return [] > normalized: list[str] = [] > for owner in owners: > owner_candidates: list[Any] > if isinstance(owner, dict): > first_name = sanitize_username(str(owner.get("first_name") or "")) > last_name = sanitize_username(str(owner.get("last_name") or "")) > full_name = " ".join( > part for part in [first_name, last_name] if part > ).strip() > snake_name = "_".join( > part for part in [first_name, last_name] if part > ).strip("_") > owner_candidates = [ > owner.get("username"), > owner.get("user_name"), > owner.get("name"), > owner.get("full_name"), > first_name, > last_name, > full_name or None, > snake_name or None, > owner.get("email"), > ] > else: > owner_candidates = [owner] > for candidate in owner_candidates: > token = normalize_username(str(candidate or "")) > if token and token not in normalized: > normalized.append(token) > return normalized # #endregion normalize_owner_tokens # #endregion ProfileUtils