1  """JSON-based user role resolver implementation."""
  2  
  3  from __future__ import annotations
  4  
  5  import logging
  6  from pathlib import Path
  7  from typing import Any
  8  
  9  from src.utils import load_json_file
 10  
 11  logger = logging.getLogger(__name__)
 12  
 13  
 14  class JsonUserRoleResolver:
 15      """Resolves user roles from a JSON mapping file.
 16  
 17      The JSON file should have the following structure:
 18      {
 19          "users": {
 20              "user@example.com": "Finance",
 21              "admin@example.com": "Admin"
 22          },
 23          "default_role": "Public"
 24      }
 25  
 26      Users are looked up by email address. If the user is not found,
 27      the default_role from the file is used (or the constructor default).
 28      """
 29  
 30      def __init__(
 31          self,
 32          mapping_file: Path | str | None = None,
 33          default_role: str = "Public",
 34      ) -> None:
 35          """Initialize the JSON user role resolver.
 36  
 37          Parameters
 38          ----------
 39          mapping_file
 40              Path to the JSON file containing user-to-role mappings.
 41              If None or file doesn't exist, all users get the default role.
 42          default_role
 43              Default role to use when user is not found in the mapping.
 44              Can be overridden by 'default_role' in the JSON file.
 45          """
 46          self._default_role = default_role
 47          self._users: dict[str, str] = {}
 48  
 49          if mapping_file:
 50              self._load_mapping(Path(mapping_file))
 51  
 52      def _load_mapping(self, mapping_file: Path) -> None:
 53          """Load the user mapping from a JSON file.
 54  
 55          Parameters
 56          ----------
 57          mapping_file
 58              Path to the JSON mapping file.
 59          """
 60          logger.info(f"Loading user mapping from: {mapping_file}")
 61          data = load_json_file(mapping_file)
 62  
 63          if data is None:
 64              return
 65  
 66          self._users = data.get("users", {})
 67  
 68          logger.info(
 69              f"Loaded {len(self._users)} user mappings, "
 70              f"default_role='{self._default_role}'"
 71          )
 72  
 73      def resolve_role(
 74          self,
 75          user_email: str | None = None,
 76          user_id: str | None = None,
 77      ) -> str:
 78          """Resolve the role for a user based on email.
 79  
 80          Parameters
 81          ----------
 82          user_email
 83              User's email address to look up.
 84          user_id
 85              User's ID (not used in this implementation, but part of protocol).
 86  
 87          Returns
 88          -------
 89          str
 90              The user's role if found, otherwise the default role.
 91          """
 92          if user_email and user_email in self._users:
 93              role = self._users[user_email]
 94              logger.debug(f"Resolved role for '{user_email}': {role}")
 95              return role
 96  
 97          logger.debug(
 98              f"User '{user_email or user_id}' not found, using default role: {self._default_role}"
 99          )
100          return self._default_role
101  
102      @property
103      def default_role(self) -> str:
104          """Get the default role used when user is not found.
105  
106          Returns
107          -------
108          str
109              The default role name.
110          """
111          return self._default_role
112  
113  
114  def create_json_resolver(config: dict[str, Any]) -> JsonUserRoleResolver:
115      """Factory function to create a JsonUserRoleResolver.
116  
117      Parameters
118      ----------
119      config
120          Configuration dictionary with optional keys:
121          - mapping_file: Path to the JSON mapping file
122          - default_role: Default role when user not found
123  
124      Returns
125      -------
126      JsonUserRoleResolver
127          Configured resolver instance.
128      """
129      return JsonUserRoleResolver(
130          mapping_file=config.get("mapping_file"),
131          default_role=config.get("default_role", "Public"),
132      )