Coverage for repo_ctx / providers / gitlab.py: 14%

1"""GitLab provider implementation.""" 

2import gitlab 

3import base64 

4import json 

5from typing import Optional, List 

6from .base import GitProvider, ProviderProject, ProviderFile 

7from .exceptions import ( 

8 ProviderNotFoundError, 

9 ProviderAuthError, 

10 ProviderFileNotFoundError, 

11 ProviderError, 

12 ProviderRateLimitError 

13) 

14 

15 

16class GitLabProvider(GitProvider): 

17 """GitLab repository provider using python-gitlab.""" 

18 

19 def __init__(self, url: str, token: str): 

20 """ 

21 Initialize GitLab provider. 

22 

23 Args: 

24 url: GitLab instance URL 

25 token: Personal access token with read_api scope 

26 

27 Raises: 

28 ProviderAuthError: Authentication failed 

29 """ 

30 self.url = url 

31 self.token = token 

32 try: 

33 self.client = gitlab.Gitlab(url, private_token=token) 

34 self.client.auth() 

35 except gitlab.exceptions.GitlabAuthenticationError as e: 

36 raise ProviderAuthError(f"GitLab authentication failed: {e}") 

37 except Exception as e: 

38 raise ProviderError(f"Failed to initialize GitLab client: {e}") 

39 

40 async def get_project(self, path: str) -> ProviderProject: 

41 """ 

42 Get project metadata from GitLab. 

43 

44 Args: 

45 path: Project path (format: group/project or group/subgroup/project) 

46 

47 Returns: 

48 ProviderProject with normalized metadata 

49 

50 Raises: 

51 ProviderNotFoundError: Project doesn't exist 

52 ProviderAuthError: Authentication failed 

53 """ 

54 try: 

55 project = self.client.projects.get(path) 

56 

57 return ProviderProject( 

58 id=str(project.id), 

59 name=project.name, 

60 path=project.path_with_namespace, 

61 description=project.description, 

62 default_branch=project.default_branch, 

63 web_url=project.web_url 

64 ) 

65 except gitlab.exceptions.GitlabGetError as e: 

66 if e.response_code == 404: 

67 raise ProviderNotFoundError(f"GitLab project not found: {path}") 

68 elif e.response_code == 401: 

69 raise ProviderAuthError(f"Authentication failed for project: {path}") 

70 else: 

71 raise ProviderError(f"Error getting GitLab project {path}: {e}") 

72 except Exception as e: 

73 raise ProviderError(f"Unexpected error getting project {path}: {e}") 

74 

75 async def get_default_branch(self, project: ProviderProject) -> str: 

76 """ 

77 Get default branch name. 

78 

79 Args: 

80 project: Project to query 

81 

82 Returns: 

83 Default branch name (e.g., "main", "master") 

84 """ 

85 # Already in ProviderProject, but fetch fresh if needed 

86 if project.default_branch: 

87 return project.default_branch 

88 

89 # Fallback: fetch from API 

90 try: 

91 gitlab_project = self.client.projects.get(project.id) 

92 return gitlab_project.default_branch 

93 except Exception as e: 

94 raise ProviderError(f"Error getting default branch: {e}") 

95 

96 async def get_file_tree( 

97 self, 

98 project: ProviderProject, 

99 ref: str, 

100 recursive: bool = True 

101 ) -> List[str]: 

102 """ 

103 Get list of all file paths in repository. 

104 

105 Args: 

106 project: Project to query 

107 ref: Branch, tag, or commit SHA 

108 recursive: Include subdirectories 

109 

110 Returns: 

111 List of file paths relative to repo root 

112 

113 Raises: 

114 ProviderError: Error accessing file tree 

115 """ 

116 try: 

117 gitlab_project = self.client.projects.get(project.id) 

118 tree = gitlab_project.repository_tree( 

119 ref=ref, 

120 recursive=recursive, 

121 all=True 

122 ) 

123 

124 # Extract file paths from tree objects 

125 file_paths = [] 

126 for item in tree: 

127 if item['type'] == 'blob': # Files only, not directories 

128 file_paths.append(item['path']) 

129 

130 return file_paths 

131 

132 except gitlab.exceptions.GitlabGetError as e: 

133 # If ref not found, try with default branch 

134 if e.response_code == 404: 

135 raise ProviderError( 

136 f"Branch/tag '{ref}' not found in project {project.path}" 

137 ) 

138 else: 

139 raise ProviderError(f"Error getting file tree: {e}") 

140 except Exception as e: 

141 raise ProviderError(f"Unexpected error getting file tree: {e}") 

142 

143 async def read_file( 

144 self, 

145 project: ProviderProject, 

146 path: str, 

147 ref: str 

148 ) -> ProviderFile: 

149 """ 

150 Read file contents from GitLab. 

151 

152 Args: 

153 project: Project to query 

154 path: File path relative to repo root 

155 ref: Branch, tag, or commit SHA 

156 

157 Returns: 

158 ProviderFile with content and metadata 

159 

160 Raises: 

161 ProviderFileNotFoundError: File doesn't exist at ref 

162 """ 

163 try: 

164 gitlab_project = self.client.projects.get(project.id) 

165 file_data = gitlab_project.files.get(file_path=path, ref=ref) 

166 

167 # Decode content if base64 encoded 

168 content = file_data.content 

169 if file_data.encoding == 'base64': 

170 content = base64.b64decode(content).decode('utf-8') 

171 

172 return ProviderFile( 

173 path=path, 

174 content=content, 

175 size=file_data.size if hasattr(file_data, 'size') else len(content) 

176 ) 

177 

178 except gitlab.exceptions.GitlabGetError as e: 

179 if e.response_code == 404: 

180 raise ProviderFileNotFoundError( 

181 f"File '{path}' not found in {project.path} at ref '{ref}'" 

182 ) 

183 else: 

184 raise ProviderError(f"Error reading file {path}: {e}") 

185 except UnicodeDecodeError: 

186 raise ProviderError( 

187 f"File '{path}' is not valid UTF-8 (binary file?)" 

188 ) 

189 except Exception as e: 

190 raise ProviderError(f"Unexpected error reading file {path}: {e}") 

191 

192 async def read_config( 

193 self, 

194 project: ProviderProject, 

195 ref: str 

196 ) -> Optional[dict]: 

197 """ 

198 Read repo-ctx configuration file if it exists. 

199 

200 Searches for configuration files in this order: 

201 1. git_context.json (current name) 

202 2. .git_context.json 

203 3. repo_context.json 

204 4. .repo-ctx.json 

205 

206 Args: 

207 project: Project to query 

208 ref: Branch, tag, or commit SHA 

209 

210 Returns: 

211 Parsed JSON config or None if not found 

212 """ 

213 config_filenames = [ 

214 "git_context.json", 

215 ".git_context.json", 

216 "repo_context.json", 

217 ".repo-ctx.json" 

218 ] 

219 

220 for filename in config_filenames: 

221 try: 

222 file = await self.read_file(project, filename, ref) 

223 return json.loads(file.content) 

224 except ProviderFileNotFoundError: 

225 # Try next filename 

226 continue 

227 except json.JSONDecodeError as e: 

228 raise ProviderError( 

229 f"Invalid JSON in config file {filename}: {e}" 

230 ) 

231 except Exception: 

232 # Try next filename 

233 continue 

234 

235 # No config file found 

236 return None 

237 

238 async def get_tags( 

239 self, 

240 project: ProviderProject, 

241 limit: int = 5 

242 ) -> List[str]: 

243 """ 

244 Get repository tags (most recent first). 

245 

246 Args: 

247 project: Project to query 

248 limit: Maximum number of tags to return 

249 

250 Returns: 

251 List of tag names 

252 

253 Raises: 

254 ProviderError: Error accessing tags 

255 """ 

256 try: 

257 gitlab_project = self.client.projects.get(project.id) 

258 tags = gitlab_project.tags.list(all=True) 

259 

260 # GitLab returns tags in reverse chronological order by default 

261 tag_names = [tag.name for tag in tags[:limit]] 

262 

263 return tag_names 

264 

265 except gitlab.exceptions.GitlabError as e: 

266 raise ProviderError(f"Error getting tags: {e}") 

267 except Exception as e: 

268 raise ProviderError(f"Unexpected error getting tags: {e}") 

269 

270 async def list_projects_in_group( 

271 self, 

272 group_path: str, 

273 include_subgroups: bool = True 

274 ) -> List[ProviderProject]: 

275 """ 

276 List all projects in a GitLab group. 

277 

278 Args: 

279 group_path: Group path (e.g., "mygroup" or "mygroup/subgroup") 

280 include_subgroups: Include projects from nested subgroups 

281 

282 Returns: 

283 List of projects in the group 

284 

285 Raises: 

286 ProviderNotFoundError: Group not found 

287 ProviderError: Error accessing group 

288 """ 

289 try: 

290 group = self.client.groups.get(group_path) 

291 projects = group.projects.list( 

292 all=True, 

293 include_subgroups=include_subgroups 

294 ) 

295 

296 result = [] 

297 for proj in projects: 

298 result.append(ProviderProject( 

299 id=str(proj.id), 

300 name=proj.name, 

301 path=proj.path_with_namespace, 

302 description=getattr(proj, 'description', None), 

303 default_branch=getattr(proj, 'default_branch', 'main'), 

304 web_url=getattr(proj, 'web_url', None) 

305 )) 

306 

307 return result 

308 

309 except gitlab.exceptions.GitlabGetError as e: 

310 if e.response_code == 404: 

311 raise ProviderNotFoundError(f"GitLab group not found: {group_path}") 

312 elif e.response_code == 401: 

313 raise ProviderAuthError( 

314 f"Authentication failed for group: {group_path}" 

315 ) 

316 else: 

317 raise ProviderError(f"Error getting group {group_path}: {e}") 

318 except gitlab.exceptions.GitlabHttpError as e: 

319 if e.response_code == 429: 

320 raise ProviderRateLimitError( 

321 f"GitLab rate limit exceeded for group: {group_path}" 

322 ) 

323 else: 

324 raise ProviderError(f"HTTP error accessing group {group_path}: {e}") 

325 except Exception as e: 

326 raise ProviderError( 

327 f"Unexpected error listing projects in group {group_path}: {e}" 

328 )