| 104 | |
| 105 | |
| 106 | class ProviderHandler: |
| 107 | # Class variable for provider domains |
| 108 | PROVIDER_DOMAINS: dict[ProviderType, str] = { |
| 109 | ProviderType.GITHUB: 'github.com', |
| 110 | ProviderType.GITLAB: GITLAB_HOST, |
| 111 | ProviderType.BITBUCKET: 'bitbucket.org', |
| 112 | ProviderType.FORGEJO: 'codeberg.org', |
| 113 | ProviderType.AZURE_DEVOPS: 'dev.azure.com', |
| 114 | } |
| 115 | |
| 116 | def __init__( |
| 117 | self, |
| 118 | provider_tokens: PROVIDER_TOKEN_TYPE, |
| 119 | external_auth_id: str | None = None, |
| 120 | external_auth_token: SecretStr | None = None, |
| 121 | external_token_manager: bool = False, |
| 122 | session_api_key: str | None = None, |
| 123 | sid: str | None = None, |
| 124 | ): |
| 125 | if not isinstance(provider_tokens, MappingProxyType): |
| 126 | raise TypeError( |
| 127 | f'provider_tokens must be a MappingProxyType, got {type(provider_tokens).__name__}' |
| 128 | ) |
| 129 | |
| 130 | self.service_class_map: dict[ProviderType, type[GitService]] = { |
| 131 | ProviderType.GITHUB: GithubServiceImpl, |
| 132 | ProviderType.GITLAB: GitLabServiceImpl, |
| 133 | ProviderType.BITBUCKET: BitBucketServiceImpl, |
| 134 | ProviderType.BITBUCKET_DATA_CENTER: BitbucketDCServiceImpl, |
| 135 | ProviderType.FORGEJO: ForgejoServiceImpl, |
| 136 | ProviderType.AZURE_DEVOPS: AzureDevOpsServiceImpl, |
| 137 | } |
| 138 | |
| 139 | self.external_auth_id = external_auth_id |
| 140 | self.external_auth_token = external_auth_token |
| 141 | self.external_token_manager = external_token_manager |
| 142 | self.session_api_key = session_api_key |
| 143 | self.sid = sid |
| 144 | self._provider_tokens = provider_tokens |
| 145 | WEB_HOST = os.getenv('WEB_HOST', '').strip() |
| 146 | self.REFRESH_TOKEN_URL = ( |
| 147 | f'https://{WEB_HOST}/api/refresh-tokens' if WEB_HOST else None |
| 148 | ) |
| 149 | |
| 150 | @property |
| 151 | def provider_tokens(self) -> PROVIDER_TOKEN_TYPE: |
| 152 | """Read-only access to provider tokens.""" |
| 153 | return self._provider_tokens |
| 154 | |
| 155 | def get_service(self, provider: ProviderType) -> GitService: |
| 156 | """Helper method to instantiate a service for a given provider""" |
| 157 | token = self.provider_tokens[provider] |
| 158 | service_class = self.service_class_map[provider] |
| 159 | return service_class( |
| 160 | user_id=token.user_id, |
| 161 | external_auth_id=self.external_auth_id, |
| 162 | external_auth_token=self.external_auth_token, |
| 163 | token=token.token, |
no outgoing calls