Coverage for .nox/test-3-12/lib/python3.12/site-packages/nskit/vcs/providers/github.py: 56%

1"""Github provider using ghapi.""" 

2from enum import Enum 

3from typing import List, Optional 

4 

5try: 

6 from fastcore.net import HTTP404NotFoundError 

7 from ghapi.all import GhApi, GhDeviceAuth, paged, Scope 

8 from ghapi.auth import _def_clientid 

9except ImportError: 

10 raise ImportError('Github Provider requires installing extra dependencies (ghapi), use pip install nskit[github]') 

11from pydantic import Field, field_validator, HttpUrl, SecretStr, ValidationInfo 

12from pydantic_settings import SettingsConfigDict 

13 

14from nskit.common.configuration import BaseConfiguration 

15from nskit.vcs.providers.abstract import RepoClient, VCSProviderSettings 

16 

17 

18class GithubRepoSettings(BaseConfiguration): 

19 """Github Repo settings.""" 

20 private: bool = True 

21 has_issue: Optional[bool] = None 

22 has_wiki: Optional[bool] = None 

23 has_downloads: Optional[bool] = None 

24 has_projects: Optional[bool] = None 

25 allow_squash_merge: Optional[bool] = None 

26 allow_merge_commit: Optional[bool] = None 

27 allow_rebase_merge: Optional[bool] = None 

28 delete_branch_on_merge: Optional[bool] = None 

29 auto_init: bool = False 

30 

31 

32class GithubSettings(VCSProviderSettings): 

33 """Github settings. 

34 

35 Uses PAT token for auth (set in environment variables as GITHUB_TOKEN) 

36 """ 

37 model_config = SettingsConfigDict(env_prefix='GITHUB_', env_file='.env') 

38 interactive: bool = Field(False, description='Use Interactive Validation for token') 

39 url: HttpUrl = "https://github.com" 

40 organisation: Optional[str] = Field(None, description='Organisation to work in, otherwise uses the user for the token') 

41 token: SecretStr = Field(None, validate_default=True, description='Token to use for authentication, falls back to interactive device authentication if not provided') 

42 repo: GithubRepoSettings = GithubRepoSettings() 

43 

44 @property 

45 def repo_client(self) -> 'GithubRepoClient': 

46 """Get the instantiated repo client.""" 

47 return GithubRepoClient(self) 

48 

49 @field_validator('token', mode='before') 

50 @classmethod 

51 def _validate_token(cls, value, info: ValidationInfo): 

52 if value is None and info.data.get('interactive', False): 

53 ghauth = GhDeviceAuth(_def_clientid, Scope.repo, Scope.delete_repo) 

54 print(ghauth.url_docs()) 

55 ghauth.open_browser() 

56 value = ghauth.wait() 

57 return value 

58 

59 

60class GithubOrgType(Enum): 

61 """Org type, user or org.""" 

62 user = 'User' 

63 org = 'Org' 

64 

65 

66class GithubRepoClient(RepoClient): 

67 """Client for managing github repos.""" 

68 

69 def __init__(self, config: GithubSettings): 

70 """Initialise the client.""" 

71 self._config = config 

72 self._github = GhApi(token=self._config.token.get_secret_value(), gh_host=self._config.url) 

73 # If the organisation is set, we get it, and assume that the token is valid 

74 # Otherwise default to the user 

75 if self._config.organisation: 

76 try: 

77 self._github.orgs.get(self._config.organisation) 

78 self._org_type = GithubOrgType.org 

79 except HTTP404NotFoundError: 

80 self._github.user.get_by_username(self._config.organisation) 

81 self._org_type = GithubOrgType.user 

82 else: 

83 self._config.organisation = self._github.get_authenticated()['login'] 

84 self._org_type = GithubOrgType.user 

85 

86 def create(self, repo_name: str): 

87 """Create the repo in the user/organisation.""" 

88 return self._github.repos.create_in_org( 

89 self._config.organisation, 

90 name=repo_name, 

91 private=self._config.repo.private, 

92 has_issues=self._config.repo.has_issues, 

93 has_wiki=self._config.repo.has_wiki, 

94 has_downloads=self._config.repo.has_downloads, 

95 has_projects=self._config.repo.has_projects, 

96 allow_squash_merge=self._config.repo.allow_squash_merge, 

97 allow_merge_commit=self._config.repo.allow_merge_commit, 

98 allow_rebase_merge=self._config.repo.allow_rebase_merge, 

99 auto_init=self._config.repo.auto_init, 

100 delete_branch_on_merge=self._config.repo.delete_branch_on_merge 

101 ) 

102 

103 def get_remote_url(self, repo_name: str) -> HttpUrl: 

104 """Get the remote url for the repo.""" 

105 if self.check_exists(repo_name): 

106 return self._github.repos.get(self._config.organisation, repo_name)['clone_url'] 

107 

108 def delete(self, repo_name: str): 

109 """Delete the repo if it exists in the organisation/user.""" 

110 if self.check_exists(repo_name): 

111 return self._github.repos.delete(self._config.organisation, repo_name) 

112 

113 def check_exists(self, repo_name: str) -> bool: 

114 """Check if the repo exists in the organisation/user.""" 

115 try: 

116 self._github.repos.get(self._config.organisation, repo_name) 

117 return True 

118 except HTTP404NotFoundError: 

119 return False 

120 

121 def list(self) -> List[str]: 

122 """List the repos in the project.""" 

123 repos = [] 

124 if self._org_type == GithubOrgType.org: 

125 get_method = self._github.repos.list_for_org 

126 else: 

127 get_method = self._github.repos.list_for_user 

128 for u in paged(get_method, self._config.organisation, per_page=100): 

129 repos += [x['name'] for x in u] 

130 return repos