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

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 

12 

13from nskit.common.configuration import BaseConfiguration, SettingsConfigDict 

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

15 

16 

17class GithubRepoSettings(BaseConfiguration): 

18 """Github Repo settings.""" 

19 model_config = SettingsConfigDict(env_prefix='GITHUB_REPO_', env_file='.env', dotenv_extra='ignore') 

20 

21 private: bool = True 

22 has_issues: Optional[bool] = None 

23 has_wiki: Optional[bool] = None 

24 has_downloads: Optional[bool] = None 

25 has_projects: Optional[bool] = None 

26 allow_squash_merge: Optional[bool] = None 

27 allow_merge_commit: Optional[bool] = None 

28 allow_rebase_merge: Optional[bool] = None 

29 delete_branch_on_merge: Optional[bool] = None 

30 auto_init: bool = False 

31 

32 

33class GithubSettings(VCSProviderSettings): 

34 """Github settings. 

35 

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

37 """ 

38 model_config = SettingsConfigDict(env_prefix='GITHUB_', env_file='.env', dotenv_extra='ignore') 

39 

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

41 url: HttpUrl = "https://api.github.com" 

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

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

44 repo: GithubRepoSettings = Field(default_factory=GithubRepoSettings) 

45 

46 @property 

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

48 """Get the instantiated repo client.""" 

49 return GithubRepoClient(self) 

50 

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

52 @classmethod 

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

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

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

56 print(ghauth.url_docs()) 

57 ghauth.open_browser() 

58 value = ghauth.wait() 

59 return value 

60 

61 

62class GithubOrgType(Enum): 

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

64 user = 'User' 

65 org = 'Org' 

66 

67 

68class GithubRepoClient(RepoClient): 

69 """Client for managing github repos.""" 

70 

71 def __init__(self, config: GithubSettings): 

72 """Initialise the client.""" 

73 self._config = config 

74 self._github = GhApi(token=self._config.token.get_secret_value(), gh_host=str(self._config.url).rstrip('/')) 

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

76 # Otherwise default to the user 

77 if self._config.organisation: 

78 try: 

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

80 self._org_type = GithubOrgType.org 

81 except HTTP404NotFoundError: 

82 self._github.users.get_by_username(self._config.organisation) 

83 self._org_type = GithubOrgType.user 

84 else: 

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

86 self._org_type = GithubOrgType.user 

87 

88 def create(self, repo_name: str): 

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

90 kwargs = { 

91 'name': repo_name, 

92 'private': self._config.repo.private, 

93 'has_issues': self._config.repo.has_issues, 

94 'has_wiki': self._config.repo.has_wiki, 

95 'has_downloads': self._config.repo.has_downloads, 

96 'has_projects': self._config.repo.has_projects, 

97 'allow_squash_merge': self._config.repo.allow_squash_merge, 

98 'allow_merge_commit': self._config.repo.allow_merge_commit, 

99 'allow_rebase_merge': self._config.repo.allow_rebase_merge, 

100 'auto_init': self._config.repo.auto_init, 

101 'delete_branch_on_merge': self._config.repo.delete_branch_on_merge 

102 } 

103 if self._org_type == GithubOrgType.org: 

104 self._github.repos.create_in_org(self._config.organisation, **kwargs) 

105 else: 

106 self._github.repos.create_for_authenticated_user(**kwargs) 

107 

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

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

110 if self.check_exists(repo_name): 

111 return self._github.repos.get(self._config.organisation, repo_name)['html_url'] 

112 

113 def get_clone_url(self, repo_name: str) -> HttpUrl: 

114 """Get the clone url for the repo.""" 

115 if self.check_exists(repo_name): 

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

117 

118 def delete(self, repo_name: str): 

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

120 if self.check_exists(repo_name): 

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

122 

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

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

125 try: 

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

127 return True 

128 except HTTP404NotFoundError: 

129 return False 

130 

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

132 """List the repos in the project.""" 

133 repos = [] 

134 if self._org_type == GithubOrgType.org: 

135 get_method = self._github.repos.list_for_org 

136 else: 

137 get_method = self._github.repos.list_for_user 

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

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

140 return repos