Coverage for /Users/antonigmitruk/golf/src/golf/auth/registry.py: 0%

1"""Provider registry system for extensible authentication providers. 

2 

3This module provides a registry-based dispatch system that allows custom 

4authentication providers to be added without modifying the core factory code. 

5""" 

6 

7from typing import Protocol, TYPE_CHECKING 

8from abc import ABC, abstractmethod 

9 

10if TYPE_CHECKING: 

11 from fastmcp.server.auth.auth import AuthProvider 

12 

13from .providers import AuthConfig 

14 

15 

16class AuthProviderFactory(Protocol): 

17 """Protocol for auth provider factory functions. 

18 

19 Custom provider factories must implement this interface to be compatible 

20 with the registry system. 

21 """ 

22 

23 def __call__(self, config: AuthConfig) -> "AuthProvider": 

24 """Create an AuthProvider from configuration. 

25 

26 Args: 

27 config: Authentication configuration object 

28 

29 Returns: 

30 Configured FastMCP AuthProvider instance 

31 

32 Raises: 

33 ValueError: If configuration is invalid 

34 ImportError: If required dependencies are missing 

35 """ 

36 ... 

37 

38 

39class BaseProviderPlugin(ABC): 

40 """Base class for auth provider plugins. 

41 

42 Provider plugins can extend this class to provide both configuration 

43 and factory logic in a single cohesive unit. 

44 """ 

45 

46 @property 

47 @abstractmethod 

48 def provider_type(self) -> str: 

49 """Return the provider type identifier.""" 

50 ... 

51 

52 @property 

53 @abstractmethod 

54 def config_class(self) -> type[AuthConfig]: 

55 """Return the configuration class for this provider.""" 

56 ... 

57 

58 @abstractmethod 

59 def create_provider(self, config: AuthConfig) -> "AuthProvider": 

60 """Create the auth provider from configuration. 

61 

62 Args: 

63 config: Authentication configuration (must be instance of config_class) 

64 

65 Returns: 

66 Configured FastMCP AuthProvider instance 

67 """ 

68 ... 

69 

70 def validate_config(self, config: AuthConfig) -> None: 

71 """Validate the configuration before creating provider. 

72 

73 Override this method to add custom validation logic. 

74 Default implementation checks config is correct type. 

75 

76 Args: 

77 config: Configuration to validate 

78 

79 Raises: 

80 ValueError: If configuration is invalid 

81 """ 

82 if not isinstance(config, self.config_class): 

83 raise ValueError( 

84 f"Expected {self.config_class.__name__} for {self.provider_type} provider, got {type(config).__name__}" 

85 ) 

86 

87 

88class AuthProviderRegistry: 

89 """Registry for authentication provider factories and plugins. 

90 

91 This registry allows custom authentication providers to be registered 

92 without modifying the core factory code. Providers can be registered 

93 either as simple factory functions or as full plugin classes. 

94 """ 

95 

96 def __init__(self) -> None: 

97 self._factories: dict[str, AuthProviderFactory] = {} 

98 self._plugins: dict[str, BaseProviderPlugin] = {} 

99 

100 def register_factory(self, provider_type: str, factory: AuthProviderFactory) -> None: 

101 """Register a factory function for a provider type. 

102 

103 Args: 

104 provider_type: Unique identifier for the provider type 

105 factory: Factory function that creates providers 

106 

107 Raises: 

108 ValueError: If provider_type is already registered 

109 """ 

110 if provider_type in self._factories or provider_type in self._plugins: 

111 raise ValueError(f"Provider type '{provider_type}' is already registered") 

112 

113 self._factories[provider_type] = factory 

114 

115 def register_plugin(self, plugin: BaseProviderPlugin) -> None: 

116 """Register a provider plugin. 

117 

118 Args: 

119 plugin: Provider plugin instance 

120 

121 Raises: 

122 ValueError: If provider type is already registered 

123 """ 

124 provider_type = plugin.provider_type 

125 if provider_type in self._factories or provider_type in self._plugins: 

126 raise ValueError(f"Provider type '{provider_type}' is already registered") 

127 

128 self._plugins[provider_type] = plugin 

129 

130 def unregister(self, provider_type: str) -> None: 

131 """Unregister a provider type. 

132 

133 Args: 

134 provider_type: Provider type to remove 

135 

136 Raises: 

137 KeyError: If provider type is not registered 

138 """ 

139 if provider_type in self._factories: 

140 del self._factories[provider_type] 

141 elif provider_type in self._plugins: 

142 del self._plugins[provider_type] 

143 else: 

144 raise KeyError(f"Provider type '{provider_type}' is not registered") 

145 

146 def get_factory(self, provider_type: str) -> AuthProviderFactory: 

147 """Get factory function for a provider type. 

148 

149 Args: 

150 provider_type: Provider type to look up 

151 

152 Returns: 

153 Factory function for the provider type 

154 

155 Raises: 

156 KeyError: If provider type is not registered 

157 """ 

158 # Check factories first 

159 if provider_type in self._factories: 

160 return self._factories[provider_type] 

161 

162 # Check plugins 

163 if provider_type in self._plugins: 

164 plugin = self._plugins[provider_type] 

165 

166 # Wrap plugin method to match factory signature 

167 def plugin_factory(config: AuthConfig) -> "AuthProvider": 

168 plugin.validate_config(config) 

169 return plugin.create_provider(config) 

170 

171 return plugin_factory 

172 

173 raise KeyError(f"No provider registered for type '{provider_type}'") 

174 

175 def create_provider(self, config: AuthConfig) -> "AuthProvider": 

176 """Create a provider from configuration using the registry. 

177 

178 Args: 

179 config: Authentication configuration 

180 

181 Returns: 

182 Configured AuthProvider instance 

183 

184 Raises: 

185 KeyError: If provider type is not registered 

186 ValueError: If configuration is invalid 

187 """ 

188 provider_type = getattr(config, "provider_type", None) 

189 if not provider_type: 

190 raise ValueError(f"Configuration {type(config).__name__} missing provider_type attribute") 

191 

192 factory = self.get_factory(provider_type) 

193 return factory(config) 

194 

195 def list_providers(self) -> list[str]: 

196 """List all registered provider types. 

197 

198 Returns: 

199 List of provider type identifiers 

200 """ 

201 return sorted(list(self._factories.keys()) + list(self._plugins.keys())) 

202 

203 def is_registered(self, provider_type: str) -> bool: 

204 """Check if a provider type is registered. 

205 

206 Args: 

207 provider_type: Provider type to check 

208 

209 Returns: 

210 True if provider type is registered 

211 """ 

212 return provider_type in self._factories or provider_type in self._plugins 

213 

214 

215# Global registry instance 

216_default_registry = AuthProviderRegistry() 

217 

218 

219def get_provider_registry() -> AuthProviderRegistry: 

220 """Get the default provider registry. 

221 

222 Returns: 

223 Default AuthProviderRegistry instance 

224 """ 

225 return _default_registry 

226 

227 

228def register_provider_factory(provider_type: str, factory: AuthProviderFactory) -> None: 

229 """Register a factory function in the default registry. 

230 

231 Args: 

232 provider_type: Unique identifier for the provider type 

233 factory: Factory function that creates providers 

234 """ 

235 _default_registry.register_factory(provider_type, factory) 

236 

237 

238def register_provider_plugin(plugin: BaseProviderPlugin) -> None: 

239 """Register a provider plugin in the default registry. 

240 

241 Args: 

242 plugin: Provider plugin instance 

243 """ 

244 _default_registry.register_plugin(plugin) 

245 

246 

247def create_auth_provider_from_registry(config: AuthConfig) -> "AuthProvider": 

248 """Create an auth provider using the default registry. 

249 

250 Args: 

251 config: Authentication configuration 

252 

253 Returns: 

254 Configured AuthProvider instance 

255 """ 

256 return _default_registry.create_provider(config)