Coverage for lynceus/devops/gitlab_devops_analyzer.py: 94%

237 statements  

« prev     ^ index     » next       coverage.py v7.10.0, created at 2025-07-29 08:46 +0000

1from collections import defaultdict 

2from datetime import date, datetime, timedelta, timezone 

3from pathlib import Path 

4from typing import Callable 

5 

6import gitlab 

7from gitlab.const import AccessLevel 

8 

9from lynceus.core.config import DATETIME_FORMAT_SHORT 

10from lynceus.core.exchange.lynceus_exchange import LynceusExchange 

11from lynceus.core.lynceus import LynceusSession 

12from lynceus.devops.devops_analyzer import DevOpsAnalyzer 

13from lynceus.utils import flatten, parse_string_to_datetime 

14from lynceus.utils.lynceus_dict import LynceusDict 

15 

16 

17def gitlab_exception_handler(func): 

18 """ 

19 Decorator to handle GitLab-specific exceptions and convert them to standard exceptions. 

20 

21 Parameters 

22 ---------- 

23 func : callable 

24 Function to wrap with exception handling 

25 

26 Returns 

27 ------- 

28 callable 

29 Wrapped function that handles GitLab exceptions 

30 

31 Raises 

32 ------ 

33 PermissionError 

34 For 401/403 HTTP status codes 

35 NameError 

36 For 404 HTTP status codes 

37 """ 

38 

39 def func_wrapper(*args, **kwargs): 

40 """ 

41 Internal wrapper function that handles GitLab API errors. 

42 

43 Parameters 

44 ---------- 

45 *args : tuple 

46 Positional arguments passed to the wrapped function 

47 **kwargs 

48 Keyword arguments passed to the wrapped function 

49 

50 Returns 

51 ------- 

52 object 

53 Result of the wrapped function call 

54 

55 Raises 

56 ------ 

57 PermissionError 

58 For 401/403 HTTP status codes 

59 NameError 

60 For 404 HTTP status codes 

61 """ 

62 try: 

63 return func(*args, **kwargs) 

64 except gitlab.exceptions.GitlabError as error: 

65 # Intercepts permission error. 

66 if error.response_code in (401, 403): 

67 raise PermissionError( 

68 "You don't have enough permission to perform this operation on Gitlab." 

69 ) from error 

70 if error.response_code == 404: 

71 raise NameError("Unable to find requested Object.") from error 

72 

73 # Raises any other error. 

74 raise 

75 

76 return func_wrapper 

77 

78 

79def get_list_from_paginated_and_count( 

80 plist_func: Callable, count: int | None = None, **kwargs 

81) -> list: 

82 """ 

83 Helper function to get a list from GitLab paginated results with optional count limit. 

84 

85 Parameters 

86 ---------- 

87 plist_func : Callable 

88 Function that returns paginated results 

89 count : int, optional 

90 Maximum number of items to retrieve 

91 **kwargs 

92 Additional arguments to pass to plist_func 

93 

94 Returns 

95 ------- 

96 list 

97 List of items from the paginated result 

98 """ 

99 if count is not None and count: 

100 kwargs = {"per_page": count, "page": 1} | kwargs 

101 else: 

102 kwargs = {"all": True} | kwargs 

103 

104 return list(plist_func(**kwargs)) 

105 

106 

107# See: https://python-gitlab.readthedocs.io/en/stable/api-usage.html 

108# See: https://docs.gitlab.com/ee/api/README.html 

109# See: https://docs.gitlab.com/ee/api/api_resources.html 

110class GitlabDevOpsAnalyzer(DevOpsAnalyzer): 

111 """ 

112 GitLab-specific implementation of the DevOps analyzer. 

113 

114 This class provides concrete implementations for all DevOps operations 

115 specific to GitLab, including authentication, user/group/project management, 

116 repository operations, and statistics gathering. It uses the python-gitlab 

117 library to interact with GitLab's REST API. 

118 

119 Attributes: 

120 IMPORT_EXPORT_STATUS_SUCCESS: Status indicating successful import/export 

121 IMPORT_EXPORT_STATUS_FAILED: Status indicating failed import/export 

122 IMPORT_EXPORT_STATUS_NONE: Status indicating no import/export operation 

123 """ 

124 

125 # See: https://docs.gitlab.com/ce/api/project_import_export.html#export-status 

126 # See: https://docs.gitlab.com/ce/api/project_import_export.html#import-status 

127 IMPORT_EXPORT_STATUS_SUCCESS: str = "finished" 

128 IMPORT_EXPORT_STATUS_FAILED: str = "failed" 

129 IMPORT_EXPORT_STATUS_NONE: str = "none" 

130 

131 def __init__( 

132 self, 

133 lynceus_session: LynceusSession, 

134 uri: str, 

135 token: str, 

136 lynceus_exchange: LynceusExchange, 

137 ): 

138 """ 

139 Initialize the GitLab DevOps analyzer. 

140 

141 Parameters 

142 ---------- 

143 lynceus_session : LynceusSession 

144 The Lynceus session instance 

145 uri : str 

146 The GitLab instance URI 

147 token : str 

148 Personal access token for GitLab authentication 

149 lynceus_exchange : LynceusExchange 

150 Exchange instance for data communication 

151 """ 

152 super().__init__(lynceus_session, uri, "gitlab", lynceus_exchange) 

153 self._manager = gitlab.Gitlab(uri, private_token=token) 

154 self.__current_token: dict | None = None 

155 

156 # The following methods are only for uniformity and coherence. 

157 def _extract_user_info(self, user) -> LynceusDict: 

158 # Extracts information which are always available. 

159 user_info = { 

160 "id": user.id, 

161 "name": user.name, 

162 "login": user.username, 

163 "username": user.username, 

164 } 

165 

166 # Extracts extra information if available. 

167 for extra_info_key, extra_info_attr in ( 

168 ("e-mail", "public_email"), 

169 ("avatar_url", "avatar_url"), 

170 ("bio", "bio"), 

171 ): 

172 value = ( 

173 user.attributes[extra_info_attr] 

174 if extra_info_attr in user.attributes 

175 else self.INFO_UNDEFINED 

176 ) 

177 user_info.update({extra_info_key: value}) 

178 

179 return LynceusDict(user_info) 

180 

181 def _extract_group_info(self, group) -> LynceusDict: 

182 return LynceusDict( 

183 {"id": group.id, "name": group.name, "path": group.full_path} 

184 ) 

185 

186 def _extract_project_info(self, project) -> LynceusDict: 

187 return LynceusDict( 

188 { 

189 "id": project.id, 

190 "name": project.name, 

191 "path": project.path_with_namespace, 

192 "web_url": project.web_url, 

193 } 

194 ) 

195 

196 def _extract_member_info(self, member) -> LynceusDict: 

197 member_info = { 

198 "id": member.id, 

199 "name": member.name, 

200 "login": member.username, 

201 "username": member.username, 

202 "state": member.state, 

203 } 

204 

205 # Extracts extra information if available. 

206 for extra_info_key, extra_info_attr in ("parent_id", "group_id"), ( 

207 "parent_id", 

208 "project_id", 

209 ): 

210 value = ( 

211 member.attributes[extra_info_attr] 

212 if extra_info_attr in member.attributes 

213 else None 

214 ) 

215 if value: 

216 member_info.update({extra_info_key: value}) 

217 

218 return LynceusDict(member_info) 

219 

220 def _extract_issue_event_info(self, issue_event, **kwargs) -> LynceusDict: 

221 return LynceusDict( 

222 { 

223 "id": issue_event.id, 

224 "issue_id": issue_event.target_iid, 

225 "action": issue_event.action_name, 

226 "target_type": issue_event.target_type, 

227 "created_at": parse_string_to_datetime( 

228 datetime_str=issue_event.created_at 

229 ), 

230 "author": issue_event.author["name"], 

231 "title": issue_event.target_title, 

232 "issue_web_url": kwargs["project_web_url"] + f"/-/issues/{issue_event.target_iid}", 

233 # N.B.: project information is unable, and must be added by caller via kwargs. 

234 } 

235 | kwargs 

236 ) 

237 

238 def _extract_commit_info(self, commit) -> LynceusDict: 

239 return LynceusDict( 

240 { 

241 "id": commit.id, 

242 "short_id": commit.short_id, 

243 "parent_ids": commit.parent_ids, 

244 "message": commit.message, 

245 "created_at": commit.created_at, 

246 "author_name": commit.author_name, 

247 "author_email": commit.author_email, 

248 "committer_name": commit.committer_name, 

249 "committer_email": commit.committer_email, 

250 } 

251 ) 

252 

253 def _extract_branch_info(self, branch) -> LynceusDict: 

254 return LynceusDict( 

255 { 

256 "name": branch.name, 

257 "merged": branch.merged, 

258 "commit_id": branch.commit["id"], 

259 "commit_short_id": branch.commit["short_id"], 

260 "created_at": parse_string_to_datetime( 

261 datetime_str=branch.commit["created_at"] 

262 ), 

263 # Not available for other DevOps: 'project_id': branch.project_id, 

264 } 

265 ) 

266 

267 def _extract_tag_info(self, tag) -> LynceusDict: 

268 return LynceusDict( 

269 { 

270 "name": tag.name, 

271 "commit_id": tag.commit["id"], 

272 "commit_short_id": tag.commit["short_id"], 

273 "created_at": parse_string_to_datetime( 

274 datetime_str=tag.commit["created_at"] 

275 ), 

276 # Not available for other DevOps: 'project_id': branch.project_id, 

277 } 

278 ) 

279 

280 # The following methods are only performing read access on DevOps backend. 

281 @gitlab_exception_handler 

282 def authenticate(self): 

283 """ 

284 Authenticate with the GitLab instance using the configured credentials. 

285 

286 Perform authentication with the GitLab API using the access token 

287 or other credentials provided during initialization. 

288 """ 

289 self._manager.auth() 

290 

291 @gitlab_exception_handler 

292 def _do_get_current_user(self): 

293 return self._manager.user 

294 

295 @gitlab_exception_handler 

296 def _do_get_user_without_cache( 

297 self, *, username: str = None, email: str = None, **kwargs 

298 ): 

299 if "user_id" in kwargs: 

300 return self._manager.users.get(id=kwargs["user_id"]) 

301 

302 # Note: there is still no API route to search an user directly 

303 # from username or email (but only from only known id) 

304 # - https://python-gitlab.readthedocs.io/en/stable/gl_objects/users.html#users-examples 

305 # - https://docs.gitlab.com/api/users/ 

306 users = self._manager.users.list( 

307 iterator=True, username=username, email=email, **kwargs 

308 ) 

309 try: 

310 user = next(users) 

311 self._logger.debug( 

312 f'Successfully lookup user with parameters: "{username=}"/"{email=}".' 

313 ) 

314 return user 

315 except StopIteration: 

316 # pylint: disable=raise-missing-from 

317 raise NameError( 

318 f'User "{username=}"/"{email=}" has not been found with specified parameters ({kwargs if kwargs else "none"}).' 

319 ) 

320 

321 @gitlab_exception_handler 

322 def _do_get_groups(self, *, count: int | None = None, **kwargs): 

323 return get_list_from_paginated_and_count( 

324 self._manager.groups.list, count, **kwargs 

325 ) 

326 

327 @gitlab_exception_handler 

328 def _do_get_group_without_cache(self, *, full_path: str, **kwargs): 

329 groups = self._manager.groups.list( 

330 iterator=True, search=full_path, search_namespaces=True, **kwargs 

331 ) 

332 if not groups: 

333 raise NameError( 

334 f'Group "{full_path}" has not been found with specified parameters ({kwargs if kwargs else "none"}).' 

335 ) 

336 

337 # Checks if match must be exact (which is now the case by default ... check if has been defined to False manually). 

338 if not bool(kwargs.get("exact_match", True)): 

339 if len(groups) > 1: 

340 self._logger.warning( 

341 f'There are more than one group matching requested path "{full_path}"' 

342 + f'(add "exact_match" parameter to ensure getting only one group): {groups} ' 

343 ) 

344 return next(groups) 

345 

346 try: 

347 group = next( 

348 filter( 

349 lambda grp: str.lower(grp.full_path) == str.lower(full_path), groups 

350 ) 

351 ) 

352 self._logger.debug( 

353 f'Successfully lookup group with full path "{full_path}".' 

354 ) 

355 return group 

356 except StopIteration: 

357 # pylint: disable=raise-missing-from 

358 raise NameError( 

359 f'Group "{full_path}" has not been found with specified parameters ({kwargs if kwargs else "none"}).' 

360 + f" Maybe you are looking for one of the following paths: {[group.full_path for group in groups]}" 

361 ) 

362 

363 @gitlab_exception_handler 

364 def _do_get_projects(self, *, count: int | None = None, **kwargs): 

365 return get_list_from_paginated_and_count( 

366 self._manager.projects.list, count, **kwargs 

367 ) 

368 

369 @gitlab_exception_handler 

370 def _do_get_project_without_cache(self, *, full_path: str, **kwargs): 

371 try: 

372 project = self._manager.projects.get( 

373 full_path, search_namespaces=True, **kwargs 

374 ) 

375 self._logger.debug( 

376 f'Successfully lookup project with full path "{full_path}".' 

377 ) 

378 return project 

379 except gitlab.exceptions.GitlabError as error: 

380 if error.response_code != 404: 

381 # Lets the @gitlab_exception_handler manages the exception. 

382 raise 

383 

384 # pylint: disable=raise-missing-from 

385 raise NameError( 

386 f'Project "{full_path}" has not been found with specified parameters ({kwargs if kwargs else "none"}).' 

387 ) 

388 

389 @gitlab_exception_handler 

390 def __get_current_token(self): 

391 """ 

392 Get information about the current personal access token. 

393 

394 Returns 

395 ------- 

396 dict 

397 Token information including ID, name, scopes, and expiration 

398 """ 

399 if not self.__current_token: 

400 # Sample: {'id': 60, 'name': 'Lynceus CI/CD', 'revoked': False, 'created_at': '2020-10-22T15:50:13.516Z', 

401 # 'scopes': ['api', 'admin_mode'], 'user_id': 5, 'last_used_at': '2023-04-12T09:22:19.495Z', 

402 # 'active': True, 'expires_at': None} 

403 self.__current_token = self._manager.personal_access_tokens.get("self") 

404 

405 return self.__current_token 

406 

407 @gitlab_exception_handler 

408 # pylint: disable=too-many-return-statements 

409 def check_permissions_on_project( 

410 self, 

411 *, 

412 full_path: str, 

413 get_metadata: bool, 

414 pull: bool, 

415 push: bool = False, 

416 maintain: bool = False, 

417 admin: bool = False, 

418 **kwargs, 

419 ): 

420 """ 

421 Check user permissions on a specific GitLab project. 

422 

423 Verify that the authenticated user has the requested permissions 

424 on the specified project by checking their access level and membership. 

425 

426 Parameters 

427 ---------- 

428 full_path : str 

429 Full path to the GitLab project (e.g., 'group/project') 

430 get_metadata : bool 

431 Whether metadata access is required 

432 pull : bool 

433 Whether pull/read access is required 

434 push : bool, optional 

435 Whether push/write access is required (default: False) 

436 maintain : bool, optional 

437 Whether maintainer access is required (default: False) 

438 admin : bool, optional 

439 Whether admin access is required (default: False) 

440 **kwargs 

441 Additional project lookup arguments 

442 

443 Returns 

444 ------- 

445 bool 

446 Permission check results with granted access levels 

447 

448 Raises 

449 ------ 

450 PermissionError 

451 If required permissions are not granted 

452 """ 

453 try: 

454 # First of all, checks get_metadata permission which is required anyway. 

455 _ = self._do_get_project(full_path=full_path, **kwargs) 

456 

457 # From here, we consider get_metadata permission is OK. 

458 

459 # See: https://docs.gitlab.com/ee/user/permissions.html 

460 # See: https://gitlab.com/gitlab-org/gitlab/-/blob/e97357824bedf007e75f8782259fe07435b64fbb/lib/gitlab/access.rb#L12-18 

461 

462 # Retrieves current use, and corresponding membership information if any, for specified project. 

463 current_user = self._do_get_current_user() 

464 members = self._do_get_project_members(full_path=full_path, recursive=True) 

465 

466 member = next(filter(lambda grp: grp.id == current_user.id, members)) 

467 access_level = member.access_level 

468 

469 # According to my tests, required role (aka access_level): 

470 # - at least REPORTER role to get access to repository metadata (name, tags, branches ...) 

471 # - at least DEVELOPER role to be able to pull the repository (the read_repository scope is NOT enough here) 

472 # - at least MAINTAINER role to get access to project statistics 

473 

474 # Checks scopes of the token. 

475 token = self.__get_current_token() 

476 token_scopes = token.scopes 

477 api_scope = "api" in token_scopes 

478 admin_scope = "admin" in token_scopes 

479 read_repository_scope = "read_repository" in token_scopes 

480 write_repository_scope = "write_repository" in token_scopes 

481 

482 if pull and ( 

483 (not api_scope and not read_repository_scope) 

484 or access_level < AccessLevel.DEVELOPER 

485 ): 

486 return False 

487 

488 if push and ( 

489 (not api_scope and not write_repository_scope) 

490 or access_level < AccessLevel.DEVELOPER 

491 ): 

492 return False 

493 

494 if maintain and ( 

495 (not api_scope and not write_repository_scope) 

496 or access_level < AccessLevel.MAINTAINER 

497 ): 

498 return False 

499 

500 if admin and ( 

501 (not api_scope and not admin_scope) or access_level < AccessLevel.OWNER 

502 ): 

503 return False 

504 

505 # All permission checks OK. 

506 return True 

507 except NameError: 

508 # Returns True if there were NO permission at all to check ... 

509 return ( 

510 not get_metadata 

511 and not pull 

512 and not push 

513 and not maintain 

514 and not admin 

515 ) 

516 except StopIteration: 

517 # Returns True if there were NO more permission to check ... 

518 return not pull and not push and not maintain and not admin 

519 

520 @gitlab_exception_handler 

521 def _do_get_project_commits( 

522 self, *, full_path: str, git_ref_name: str, count: int | None = None, **kwargs 

523 ): 

524 project = self._do_get_project(full_path=full_path, **kwargs) 

525 return get_list_from_paginated_and_count( 

526 project.commits.list, count, ref_name=git_ref_name, **kwargs 

527 ) 

528 

529 @gitlab_exception_handler 

530 def _do_get_project_branches( 

531 self, *, full_path: str, count: int | None = None, **kwargs 

532 ): 

533 project = self._do_get_project(full_path=full_path, **kwargs) 

534 return get_list_from_paginated_and_count(project.branches.list, count, **kwargs) 

535 

536 @gitlab_exception_handler 

537 def _do_get_project_tags( 

538 self, *, full_path: str, count: int | None = None, **kwargs 

539 ): 

540 project = self._do_get_project(full_path=full_path, **kwargs) 

541 return get_list_from_paginated_and_count(project.tags.list, count, **kwargs) 

542 

543 @gitlab_exception_handler 

544 def _do_get_project_members( 

545 self, *, full_path: str, count: int | None = None, **kwargs 

546 ): 

547 project = self._do_get_project(full_path=full_path, **kwargs) 

548 

549 if not bool(kwargs.get("recursive", False)): 

550 return get_list_from_paginated_and_count( 

551 project.members.list, count, **kwargs 

552 ) 

553 

554 return get_list_from_paginated_and_count( 

555 project.members_all.list, count, **kwargs 

556 ) 

557 

558 @gitlab_exception_handler 

559 def _do_get_group_members( 

560 self, *, full_path: str, count: int | None = None, **kwargs 

561 ): 

562 group = self._do_get_group(full_path=full_path, **kwargs) 

563 

564 if not bool(kwargs.get("recursive", False)): 

565 return get_list_from_paginated_and_count( 

566 group.members.list, count, **kwargs 

567 ) 

568 

569 return get_list_from_paginated_and_count( 

570 group.members_all.list, count, **kwargs 

571 ) 

572 

573 @gitlab_exception_handler 

574 def _do_get_project_issue_events( 

575 self, 

576 *, 

577 full_path: str, 

578 action: str | None = None, 

579 from_date: datetime | None = None, 

580 to_date: datetime | None = None, 

581 count: int | None = None, 

582 **kwargs, 

583 ): 

584 # See: https://python-gitlab.readthedocs.io/en/stable/api/gitlab.v4.html?highlight=events#gitlab.v4.objects.ProjectEventManager 

585 # Object listing filters 

586 # action => https://docs.gitlab.com/ee/user/profile/index.html#user-contribution-events 

587 # target_type => https://docs.gitlab.com/ee/api/events.html#target-types 

588 # sort 

589 project = self._do_get_project(full_path=full_path, **kwargs) 

590 

591 # Adds filters if needed. 

592 if action: 

593 kwargs["action"] = action 

594 

595 # before & after => https://docs.gitlab.com/ee/api/events.html#date-formatting 

596 if from_date: 

597 kwargs["after"] = from_date.strftime("%Y-%m-%d") 

598 

599 if to_date: 

600 kwargs["before"] = to_date.strftime("%Y-%m-%d") 

601 

602 return get_list_from_paginated_and_count( 

603 project.events.list, count, target_type="issue", **kwargs 

604 ) 

605 

606 @gitlab_exception_handler 

607 def _do_get_project_issues( 

608 self, *, full_path: str, count: int | None = None, **kwargs 

609 ): 

610 project = self._do_get_project(full_path=full_path, **kwargs) 

611 return get_list_from_paginated_and_count(project.issues.list, count, **kwargs) 

612 

613 @gitlab_exception_handler 

614 def _do_get_project_merge_requests( 

615 self, *, full_path: str, count: int | None = None, **kwargs 

616 ): 

617 project = self._do_get_project(full_path=full_path, **kwargs) 

618 return get_list_from_paginated_and_count( 

619 project.mergerequests.list, count, **kwargs 

620 ) 

621 

622 @gitlab_exception_handler 

623 def _do_get_project_milestones( 

624 self, *, full_path: str, count: int | None = None, **kwargs 

625 ): 

626 project = self._do_get_project(full_path=full_path, **kwargs) 

627 return get_list_from_paginated_and_count( 

628 project.milestones.list, count, **kwargs 

629 ) 

630 

631 @gitlab_exception_handler 

632 def __get_recursive_groups( 

633 self, *, full_path: str, count: int | None = None, **kwargs 

634 ): 

635 """ 

636 Recursively get all subgroups for a given group. 

637 

638 Parameters 

639 ---------- 

640 full_path : str 

641 Full path of the parent group 

642 count : int, optional 

643 Maximum number of groups to retrieve 

644 **kwargs 

645 Additional filtering parameters 

646 

647 Returns 

648 ------- 

649 list 

650 List of all groups including the parent and all recursive subgroups 

651 """ 

652 # TODO: find a way to implement count properly. 

653 group = self._do_get_group(full_path=full_path, **kwargs) 

654 subgroups = get_list_from_paginated_and_count( 

655 group.subgroups.list, count, **kwargs 

656 ) 

657 if not subgroups: 

658 return [group] 

659 return flatten( 

660 self.__get_recursive_groups(full_path=subgroup.full_path, **kwargs) 

661 for subgroup in subgroups 

662 ) 

663 

664 @gitlab_exception_handler 

665 def _do_get_group_projects( 

666 self, *, full_path: str, count: int | None = None, **kwargs 

667 ): 

668 group = self._do_get_group(full_path=full_path, **kwargs) 

669 

670 # Checks if recursive is requested. 

671 if not bool(kwargs.get("recursive", False)): 

672 project_list = get_list_from_paginated_and_count( 

673 group.projects.list, count, **kwargs 

674 ) 

675 else: 

676 # TODO: find a way to implement count 

677 all_groups = {group} | set( 

678 self.__get_recursive_groups(full_path=full_path, **kwargs) 

679 ) 

680 project_list = flatten( 

681 group.projects.list(all=True) for group in all_groups 

682 ) 

683 

684 return list( 

685 map( 

686 lambda gp: self._do_get_project(full_path=gp.path_with_namespace), 

687 project_list, 

688 ) 

689 ) 

690 

691 @gitlab_exception_handler 

692 def _do_get_group_milestones( 

693 self, *, full_path: str, count: int | None = None, **kwargs 

694 ): 

695 group = self._do_get_group(full_path=full_path, **kwargs) 

696 return get_list_from_paginated_and_count(group.milestones.list, count, **kwargs) 

697 

698 @gitlab_exception_handler 

699 def get_user_stats_commit_activity( 

700 self, 

701 *, 

702 group_full_path: str = None, 

703 project_full_path: str = None, 

704 since: datetime = None, 

705 keep_empty_stats: bool = False, 

706 count: int | None = None, 

707 ): 

708 """ 

709 Get commit activity statistics for the authenticated user in GitLab. 

710 

711 Parameters 

712 ---------- 

713 group_full_path : str, optional 

714 Group path to limit statistics to 

715 project_full_path : str, optional 

716 Project path to include in statistics 

717 since : datetime, optional 

718 Start date for statistics (defaults to 365 days ago) 

719 keep_empty_stats : bool, optional 

720 Whether to include days with zero commits (default: False) 

721 count : int, optional 

722 Maximum number of projects to analyze 

723 

724 Returns 

725 ------- 

726 dict 

727 Mapping of dates to commit counts 

728 """ 

729 # TODO: find a way to implement count 

730 # See: https://docs.gitlab.com/ee/api/project_statistics.html#get-the-statistics-of-the-last-30-days 

731 

732 # Defines projects on which to perform statistics. 

733 if group_full_path is None: 

734 projects = self._do_get_projects(count=count) 

735 else: 

736 projects = self._do_get_group_projects( 

737 full_path=group_full_path, count=count 

738 ) 

739 if project_full_path is not None: 

740 projects.append(self._do_get_project(full_path=project_full_path)) 

741 

742 # Defines threshold date. 

743 contributions_since: datetime = ( 

744 since if since else datetime.now(tz=timezone.utc) - timedelta(days=365) 

745 ) 

746 stats_user_commit_activity: dict[date, int] = defaultdict(int) 

747 for _project in projects: 

748 for commit_activity in _project.additionalstatistics.get().fetches["days"]: 

749 day_date = parse_string_to_datetime( 

750 datetime_str=commit_activity["date"], 

751 datetime_format=DATETIME_FORMAT_SHORT, 

752 ) 

753 

754 # Ignores oldest statistics. 

755 if day_date < contributions_since: 

756 continue 

757 

758 # Ignores 0 stats but if wanted. 

759 if not keep_empty_stats and not commit_activity["count"]: 

760 continue 

761 

762 stats_user_commit_activity[day_date.date()] += commit_activity["count"] 

763 

764 return stats_user_commit_activity 

765 

766 @gitlab_exception_handler 

767 # pylint: disable=unused-argument 

768 def get_user_contributions( 

769 self, 

770 *, 

771 since: datetime = None, 

772 keep_empty_stats: bool = False, 

773 count: int | None = None, 

774 ): 

775 """ 

776 Get user contribution statistics (not implemented for GitLab). 

777 

778 This method is not implemented for GitLab as the equivalent functionality 

779 is not readily available through the GitLab API. 

780 

781 Parameters 

782 ---------- 

783 since : datetime, optional 

784 Start date for contributions (unused) 

785 keep_empty_stats : bool, optional 

786 Whether to include empty statistics (unused) (default: False) 

787 count : int, optional 

788 Maximum number of projects to analyze (unused) 

789 

790 Returns 

791 ------- 

792 dict 

793 Empty dictionary 

794 """ 

795 self._logger.warning( 

796 "get_user_contributions is not implemented yet for GitLab DevOps Analyzer" 

797 ) 

798 return {} 

799 

800 @gitlab_exception_handler 

801 def get_user_stats_code_frequency(self, *, count: int | None = None): 

802 """ 

803 Get code frequency statistics (not implemented for GitLab). 

804 

805 This method is not implemented for GitLab as the equivalent functionality 

806 is not readily available through the GitLab API. 

807 

808 Parameters 

809 ---------- 

810 count : int, optional 

811 Maximum number of projects to analyze (unused) 

812 

813 Returns 

814 ------- 

815 dict 

816 Empty dictionary 

817 """ 

818 self._logger.warning( 

819 "get_user_stats_commit_activity is not implemented yet for GitLab DevOps Analyzer" 

820 ) 

821 return {} 

822 

823 @gitlab_exception_handler 

824 def _do_download_repository( 

825 self, 

826 *, 

827 project_full_path: str, 

828 dest_path: Path, 

829 reference: str = None, 

830 chunk_size: int = 1024, 

831 **kwargs, 

832 ): 

833 # See: https://docs.gitlab.com/ee/api/repositories.html#get-file-archive 

834 # See: https://github.com/python-gitlab/python-gitlab/blob/main/gitlab/v4/objects.py#L4465 

835 project = self._do_get_project(full_path=project_full_path, **kwargs) 

836 with open(dest_path, "wb") as dest_file: 

837 project.repository_archive( 

838 sha=reference, 

839 streamed=True, 

840 action=dest_file.write, 

841 chunk_size=chunk_size, 

842 **kwargs, 

843 )