allspice.apiobject
1# cSpell:words ECAD 2 3from __future__ import annotations 4 5import logging 6import re 7from dataclasses import asdict, dataclass 8from datetime import datetime, timezone 9from enum import Enum 10from functools import cached_property 11from typing import ( 12 IO, 13 Any, 14 ClassVar, 15 Dict, 16 FrozenSet, 17 List, 18 Literal, 19 Optional, 20 Sequence, 21 Set, 22 Tuple, 23 Union, 24) 25 26try: 27 from typing_extensions import Self 28except ImportError: 29 from typing import Self 30 31from .baseapiobject import ApiObject, ReadonlyApiObject 32from .exceptions import ConflictException, NotFoundException 33 34 35class Organization(ApiObject): 36 """see https://hub.allspice.io/api/swagger#/organization/orgGetAll""" 37 38 active: Optional[bool] 39 avatar_url: str 40 created: Optional[str] 41 description: str 42 email: str 43 followers_count: Optional[int] 44 following_count: Optional[int] 45 full_name: str 46 html_url: Optional[str] 47 id: int 48 is_admin: Optional[bool] 49 language: Optional[str] 50 last_login: Optional[str] 51 location: str 52 login: Optional[str] 53 login_name: Optional[str] 54 name: Optional[str] 55 prohibit_login: Optional[bool] 56 repo_admin_change_team_access: Optional[bool] 57 restricted: Optional[bool] 58 source_id: Optional[int] 59 starred_repos_count: Optional[int] 60 username: str 61 visibility: str 62 website: str 63 64 API_OBJECT = """/orgs/{name}""" # <org> 65 ORG_REPOS_REQUEST = """/orgs/%s/repos""" # <org> 66 ORG_TEAMS_REQUEST = """/orgs/%s/teams""" # <org> 67 ORG_TEAMS_CREATE = """/orgs/%s/teams""" # <org> 68 ORG_GET_MEMBERS = """/orgs/%s/members""" # <org> 69 ORG_IS_MEMBER = """/orgs/%s/members/%s""" # <org>, <username> 70 ORG_HEATMAP = """/users/%s/heatmap""" # <username> 71 72 def __init__(self, allspice_client): 73 super().__init__(allspice_client) 74 75 def __eq__(self, other): 76 if not isinstance(other, Organization): 77 return False 78 return self.allspice_client == other.allspice_client and self.name == other.name 79 80 def __hash__(self): 81 return hash(self.allspice_client) ^ hash(self.name) 82 83 @classmethod 84 def request(cls, allspice_client, name: str) -> Self: 85 return cls._request(allspice_client, {"name": name}) 86 87 @classmethod 88 def parse_response(cls, allspice_client, result) -> "Organization": 89 api_object = super().parse_response(allspice_client, result) 90 # add "name" field to make this behave similar to users for gitea < 1.18 91 # also necessary for repository-owner when org is repo owner 92 if not hasattr(api_object, "name"): 93 Organization._add_read_property("name", result["username"], api_object) 94 return api_object 95 96 _patchable_fields: ClassVar[set[str]] = { 97 "description", 98 "full_name", 99 "location", 100 "visibility", 101 "website", 102 } 103 104 def commit(self): 105 args = {"name": self.name} 106 self._commit(args) 107 108 def create_repo( 109 self, 110 repoName: str, 111 description: str = "", 112 private: bool = False, 113 autoInit=True, 114 gitignores: Optional[str] = None, 115 license: Optional[str] = None, 116 readme: str = "Default", 117 issue_labels: Optional[str] = None, 118 default_branch="master", 119 ): 120 """Create an organization Repository 121 122 Throws: 123 AlreadyExistsException: If the Repository exists already. 124 Exception: If something else went wrong. 125 """ 126 result = self.allspice_client.requests_post( 127 f"/orgs/{self.name}/repos", 128 data={ 129 "name": repoName, 130 "description": description, 131 "private": private, 132 "auto_init": autoInit, 133 "gitignores": gitignores, 134 "license": license, 135 "issue_labels": issue_labels, 136 "readme": readme, 137 "default_branch": default_branch, 138 }, 139 ) 140 if "id" in result: 141 self.allspice_client.logger.info("Successfully created Repository %s " % result["name"]) 142 else: 143 self.allspice_client.logger.error(result["message"]) 144 raise Exception("Repository not created... (gitea: %s)" % result["message"]) 145 return Repository.parse_response(self.allspice_client, result) 146 147 def get_repositories(self) -> List["Repository"]: 148 results = self.allspice_client.requests_get_paginated( 149 Organization.ORG_REPOS_REQUEST % self.username 150 ) 151 return [Repository.parse_response(self.allspice_client, result) for result in results] 152 153 def get_repository(self, name) -> "Repository": 154 repos = self.get_repositories() 155 for repo in repos: 156 if repo.name == name: 157 return repo 158 raise NotFoundException("Repository %s not existent in organization." % name) 159 160 def get_teams(self) -> List["Team"]: 161 results = self.allspice_client.requests_get(Organization.ORG_TEAMS_REQUEST % self.username) 162 teams = [Team.parse_response(self.allspice_client, result) for result in results] 163 # organisation seems to be missing using this request, so we add org manually 164 for t in teams: 165 setattr(t, "_organization", self) 166 return teams 167 168 def get_team(self, name) -> "Team": 169 teams = self.get_teams() 170 for team in teams: 171 if team.name == name: 172 return team 173 raise NotFoundException("Team not existent in organization.") 174 175 def create_team( 176 self, 177 name: str, 178 description: str = "", 179 permission: str = "read", 180 can_create_org_repo: bool = False, 181 includes_all_repositories: bool = False, 182 units=( 183 "repo.code", 184 "repo.issues", 185 "repo.ext_issues", 186 "repo.wiki", 187 "repo.pulls", 188 "repo.releases", 189 "repo.ext_wiki", 190 ), 191 units_map={}, 192 ) -> "Team": 193 """Alias for AllSpice#create_team""" 194 # TODO: Move AllSpice#create_team to Organization#create_team and 195 # deprecate AllSpice#create_team. 196 return self.allspice_client.create_team( 197 org=self, 198 name=name, 199 description=description, 200 permission=permission, 201 can_create_org_repo=can_create_org_repo, 202 includes_all_repositories=includes_all_repositories, 203 units=units, 204 units_map=units_map, 205 ) 206 207 def get_members(self) -> List["User"]: 208 results = self.allspice_client.requests_get(Organization.ORG_GET_MEMBERS % self.username) 209 return [User.parse_response(self.allspice_client, result) for result in results] 210 211 def is_member(self, username) -> bool: 212 if isinstance(username, User): 213 username = username.username 214 try: 215 # returns 204 if its ok, 404 if its not 216 self.allspice_client.requests_get( 217 Organization.ORG_IS_MEMBER % (self.username, username) 218 ) 219 return True 220 except Exception: 221 return False 222 223 def remove_member(self, user: "User"): 224 path = f"/orgs/{self.username}/members/{user.username}" 225 self.allspice_client.requests_delete(path) 226 227 def delete(self): 228 """Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" 229 for repo in self.get_repositories(): 230 repo.delete() 231 self.allspice_client.requests_delete(Organization.API_OBJECT.format(name=self.username)) 232 self.deleted = True 233 234 def get_heatmap(self) -> List[Tuple[datetime, int]]: 235 results = self.allspice_client.requests_get(User.USER_HEATMAP % self.username) 236 results = [ 237 (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) 238 for result in results 239 ] 240 return results 241 242 243class User(ApiObject): 244 active: bool 245 admin: Any 246 allow_create_organization: Any 247 allow_git_hook: Any 248 allow_import_local: Any 249 avatar_url: str 250 created: str 251 description: str 252 email: str 253 emails: List[Any] 254 followers_count: int 255 following_count: int 256 full_name: str 257 html_url: str 258 id: int 259 is_admin: bool 260 language: str 261 last_login: str 262 location: str 263 login: str 264 login_name: str 265 max_repo_creation: Any 266 must_change_password: Any 267 password: Any 268 prohibit_login: bool 269 restricted: bool 270 source_id: int 271 starred_repos_count: int 272 username: str 273 visibility: str 274 website: str 275 276 API_OBJECT = """/users/{name}""" # <org> 277 USER_MAIL = """/user/emails?sudo=%s""" # <name> 278 USER_PATCH = """/admin/users/%s""" # <username> 279 ADMIN_DELETE_USER = """/admin/users/%s""" # <username> 280 ADMIN_EDIT_USER = """/admin/users/{username}""" # <username> 281 USER_HEATMAP = """/users/%s/heatmap""" # <username> 282 283 def __init__(self, allspice_client): 284 super().__init__(allspice_client) 285 self._emails = [] 286 287 def __eq__(self, other): 288 if not isinstance(other, User): 289 return False 290 return self.allspice_client == other.allspice_client and self.id == other.id 291 292 def __hash__(self): 293 return hash(self.allspice_client) ^ hash(self.id) 294 295 @property 296 def emails(self): 297 self.__request_emails() 298 return self._emails 299 300 @classmethod 301 def request(cls, allspice_client, name: str) -> "User": 302 api_object = cls._request(allspice_client, {"name": name}) 303 return api_object 304 305 _patchable_fields: ClassVar[set[str]] = { 306 "active", 307 "admin", 308 "allow_create_organization", 309 "allow_git_hook", 310 "allow_import_local", 311 "email", 312 "full_name", 313 "location", 314 "login_name", 315 "max_repo_creation", 316 "must_change_password", 317 "password", 318 "prohibit_login", 319 "website", 320 } 321 322 def commit(self, login_name: str, source_id: int = 0): 323 """ 324 Unfortunately it is necessary to require the login name 325 as well as the login source (that is not supplied when getting a user) for 326 changing a user. 327 Usually source_id is 0 and the login_name is equal to the username. 328 """ 329 values = self.get_dirty_fields() 330 values.update( 331 # api-doc says that the "source_id" is necessary; works without though 332 {"login_name": login_name, "source_id": source_id} 333 ) 334 args = {"username": self.username} 335 self.allspice_client.requests_patch(User.ADMIN_EDIT_USER.format(**args), data=values) 336 self._dirty_fields = {} 337 338 def create_repo( 339 self, 340 repoName: str, 341 description: str = "", 342 private: bool = False, 343 autoInit=True, 344 gitignores: Optional[str] = None, 345 license: Optional[str] = None, 346 readme: str = "Default", 347 issue_labels: Optional[str] = None, 348 default_branch="master", 349 ): 350 """Create a user Repository 351 352 Throws: 353 AlreadyExistsException: If the Repository exists already. 354 Exception: If something else went wrong. 355 """ 356 result = self.allspice_client.requests_post( 357 "/user/repos", 358 data={ 359 "name": repoName, 360 "description": description, 361 "private": private, 362 "auto_init": autoInit, 363 "gitignores": gitignores, 364 "license": license, 365 "issue_labels": issue_labels, 366 "readme": readme, 367 "default_branch": default_branch, 368 }, 369 ) 370 if "id" in result: 371 self.allspice_client.logger.info("Successfully created Repository %s " % result["name"]) 372 else: 373 self.allspice_client.logger.error(result["message"]) 374 raise Exception("Repository not created... (gitea: %s)" % result["message"]) 375 return Repository.parse_response(self.allspice_client, result) 376 377 def get_repositories(self) -> List["Repository"]: 378 """Get all Repositories owned by this User.""" 379 url = f"/users/{self.username}/repos" 380 results = self.allspice_client.requests_get_paginated(url) 381 return [Repository.parse_response(self.allspice_client, result) for result in results] 382 383 def get_orgs(self) -> List[Organization]: 384 """Get all Organizations this user is a member of.""" 385 url = f"/users/{self.username}/orgs" 386 results = self.allspice_client.requests_get_paginated(url) 387 return [Organization.parse_response(self.allspice_client, result) for result in results] 388 389 def get_teams(self) -> List["Team"]: 390 url = "/user/teams" 391 results = self.allspice_client.requests_get_paginated(url, sudo=self) 392 return [Team.parse_response(self.allspice_client, result) for result in results] 393 394 def get_accessible_repos(self) -> List["Repository"]: 395 """Get all Repositories accessible by the logged in User.""" 396 results = self.allspice_client.requests_get("/user/repos", sudo=self) 397 return [Repository.parse_response(self.allspice_client, result) for result in results] 398 399 def __request_emails(self): 400 result = self.allspice_client.requests_get(User.USER_MAIL % self.login) 401 # report if the adress changed by this 402 for mail in result: 403 self._emails.append(mail["email"]) 404 if mail["primary"]: 405 self._email = mail["email"] 406 407 def delete(self): 408 """Deletes this User. Also deletes all Repositories he owns.""" 409 self.allspice_client.requests_delete(User.ADMIN_DELETE_USER % self.username) 410 self.deleted = True 411 412 def get_heatmap(self) -> List[Tuple[datetime, int]]: 413 results = self.allspice_client.requests_get(User.USER_HEATMAP % self.username) 414 results = [ 415 (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) 416 for result in results 417 ] 418 return results 419 420 421class Branch(ReadonlyApiObject): 422 commit: Dict[str, Optional[Union[str, Dict[str, str], Dict[str, Optional[Union[bool, str]]]]]] 423 effective_branch_protection_name: str 424 enable_status_check: bool 425 name: str 426 protected: bool 427 required_approvals: int 428 status_check_contexts: List[Any] 429 user_can_merge: bool 430 user_can_push: bool 431 432 API_OBJECT = """/repos/{owner}/{repo}/branches/{branch}""" 433 434 def __init__(self, allspice_client): 435 super().__init__(allspice_client) 436 437 def __eq__(self, other): 438 if not isinstance(other, Branch): 439 return False 440 return self.commit == other.commit and self.name == other.name 441 442 def __hash__(self): 443 return hash(self.commit["id"]) ^ hash(self.name) 444 445 _fields_to_parsers: ClassVar[dict] = { 446 # This is not a commit object 447 # "commit": lambda allspice_client, c: Commit.parse_response(allspice_client, c) 448 } 449 450 @classmethod 451 def request(cls, allspice_client, owner: str, repo: str, branch: str): 452 return cls._request(allspice_client, {"owner": owner, "repo": repo, "branch": branch}) 453 454 455class GitEntry(ReadonlyApiObject): 456 """ 457 An object representing a file or directory in the Git tree. 458 """ 459 460 mode: str 461 path: str 462 sha: str 463 size: int 464 type: str 465 url: str 466 467 def __init__(self, allspice_client): 468 super().__init__(allspice_client) 469 470 def __eq__(self, other) -> bool: 471 if not isinstance(other, GitEntry): 472 return False 473 return self.sha == other.sha 474 475 def __hash__(self) -> int: 476 return hash(self.sha) 477 478 479class Repository(ApiObject): 480 allow_fast_forward_only_merge: bool 481 allow_manual_merge: Any 482 allow_merge_commits: bool 483 allow_rebase: bool 484 allow_rebase_explicit: bool 485 allow_rebase_update: bool 486 allow_squash_merge: bool 487 archived: bool 488 archived_at: str 489 autodetect_manual_merge: Any 490 avatar_url: str 491 clone_url: str 492 created_at: str 493 default_allow_maintainer_edit: bool 494 default_branch: str 495 default_delete_branch_after_merge: bool 496 default_merge_style: str 497 description: str 498 empty: bool 499 enable_prune: Any 500 external_tracker: Any 501 external_wiki: Any 502 fork: bool 503 forks_count: int 504 full_name: str 505 has_actions: bool 506 has_issues: bool 507 has_packages: bool 508 has_projects: bool 509 has_pull_requests: bool 510 has_releases: bool 511 has_wiki: bool 512 html_url: str 513 id: int 514 ignore_whitespace_conflicts: bool 515 internal: bool 516 internal_tracker: Dict[str, bool] 517 language: str 518 languages_url: str 519 licenses: Any 520 link: str 521 mirror: bool 522 mirror_interval: str 523 mirror_updated: str 524 name: str 525 object_format_name: str 526 open_issues_count: int 527 open_pr_counter: int 528 original_url: str 529 owner: Union["User", "Organization"] 530 parent: Any 531 permissions: Dict[str, bool] 532 private: bool 533 projects_mode: str 534 release_counter: int 535 repo_transfer: Any 536 size: int 537 ssh_url: str 538 stars_count: int 539 template: bool 540 topics: List[Union[Any, str]] 541 updated_at: datetime 542 url: str 543 watchers_count: int 544 website: str 545 546 API_OBJECT = """/repos/{owner}/{name}""" # <owner>, <reponame> 547 REPO_IS_COLLABORATOR = """/repos/%s/%s/collaborators/%s""" # <owner>, <reponame>, <username> 548 REPO_SEARCH = """/repos/search/""" 549 REPO_BRANCHES = """/repos/%s/%s/branches""" # <owner>, <reponame> 550 REPO_BRANCH = """/repos/{owner}/{repo}/branches/{branch}""" 551 REPO_ISSUES = """/repos/{owner}/{repo}/issues""" # <owner, reponame> 552 REPO_DESIGN_REVIEWS = """/repos/{owner}/{repo}/pulls""" 553 REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame> 554 REPO_TIMES = """/repos/%s/%s/times""" # <owner>, <reponame> 555 REPO_USER_TIME = """/repos/%s/%s/times/%s""" # <owner>, <reponame>, <username> 556 REPO_COMMITS = "/repos/%s/%s/commits" # <owner>, <reponame> 557 REPO_TRANSFER = "/repos/{owner}/{repo}/transfer" 558 REPO_MILESTONES = """/repos/{owner}/{repo}/milestones""" 559 REPO_GET_ARCHIVE = "/repos/{owner}/{repo}/archive/{ref}.{format}" 560 REPO_GET_ALLSPICE_JSON = "/repos/{owner}/{repo}/allspice_generated/json/{content}" 561 REPO_GET_ALLSPICE_SVG = "/repos/{owner}/{repo}/allspice_generated/svg/{content}" 562 REPO_GET_ALLSPICE_PROJECT = "/repos/{owner}/{repo}/allspice_generated/project/{content}" 563 REPO_GET_TOPICS = "/repos/{owner}/{repo}/topics" 564 REPO_ADD_TOPIC = "/repos/{owner}/{repo}/topics/{topic}" 565 REPO_GET_RELEASES = "/repos/{owner}/{repo}/releases" 566 REPO_GET_LATEST_RELEASE = "/repos/{owner}/{repo}/releases/latest" 567 REPO_GET_RELEASE_BY_TAG = "/repos/{owner}/{repo}/releases/tags/{tag}" 568 REPO_GET_COMMIT_STATUS = "/repos/{owner}/{repo}/statuses/{sha}" 569 REPO_GET_MEDIA = "/repos/{owner}/{repo}/media/{path}" 570 REPO_GET_TREE = "/repos/{owner}/{repo}/git/trees/{ref}" 571 572 class ArchiveFormat(Enum): 573 """ 574 Archive formats for Repository.get_archive 575 """ 576 577 TAR = "tar.gz" 578 ZIP = "zip" 579 580 class CommitStatusSort(Enum): 581 """ 582 Sort order for Repository.get_commit_status 583 """ 584 585 OLDEST = "oldest" 586 RECENT_UPDATE = "recentupdate" 587 LEAST_UPDATE = "leastupdate" 588 LEAST_INDEX = "leastindex" 589 HIGHEST_INDEX = "highestindex" 590 591 def __init__(self, allspice_client): 592 super().__init__(allspice_client) 593 594 def __eq__(self, other): 595 if not isinstance(other, Repository): 596 return False 597 return self.owner == other.owner and self.name == other.name 598 599 def __hash__(self): 600 return hash(self.owner) ^ hash(self.name) 601 602 _fields_to_parsers: ClassVar[dict] = { 603 # dont know how to tell apart user and org as owner except form email being empty. 604 "owner": lambda allspice_client, r: ( 605 Organization.parse_response(allspice_client, r) 606 if r["email"] == "" 607 else User.parse_response(allspice_client, r) 608 ), 609 "updated_at": lambda _, t: Util.convert_time(t), 610 } 611 612 @classmethod 613 def request( 614 cls, 615 allspice_client, 616 owner: str, 617 name: str, 618 ) -> Repository: 619 return cls._request(allspice_client, {"owner": owner, "name": name}) 620 621 @classmethod 622 def search( 623 cls, 624 allspice_client, 625 query: Optional[str] = None, 626 topic: bool = False, 627 include_description: bool = False, 628 user: Optional[User] = None, 629 owner_to_prioritize: Union[User, Organization, None] = None, 630 ) -> list[Repository]: 631 """ 632 Search for repositories. 633 634 See https://hub.allspice.io/api/swagger#/repository/repoSearch 635 636 :param query: The query string to search for 637 :param topic: If true, the query string will only be matched against the 638 repository's topic. 639 :param include_description: If true, the query string will be matched 640 against the repository's description as well. 641 :param user: If specified, only repositories that this user owns or 642 contributes to will be searched. 643 :param owner_to_prioritize: If specified, repositories owned by the 644 given entity will be prioritized in the search. 645 :returns: All repositories matching the query. If there are many 646 repositories matching this query, this may take some time. 647 """ 648 649 params = {} 650 651 if query is not None: 652 params["q"] = query 653 if topic: 654 params["topic"] = topic 655 if include_description: 656 params["include_description"] = include_description 657 if user is not None: 658 params["user"] = user.id 659 if owner_to_prioritize is not None: 660 params["owner_to_prioritize"] = owner_to_prioritize.id 661 662 responses = allspice_client.requests_get_paginated(cls.REPO_SEARCH, params=params) 663 664 return [Repository.parse_response(allspice_client, response) for response in responses] 665 666 _patchable_fields: ClassVar[set[str]] = { 667 "allow_manual_merge", 668 "allow_merge_commits", 669 "allow_rebase", 670 "allow_rebase_explicit", 671 "allow_rebase_update", 672 "allow_squash_merge", 673 "archived", 674 "autodetect_manual_merge", 675 "default_branch", 676 "default_delete_branch_after_merge", 677 "default_merge_style", 678 "description", 679 "enable_prune", 680 "external_tracker", 681 "external_wiki", 682 "has_actions", 683 "has_issues", 684 "has_projects", 685 "has_pull_requests", 686 "has_wiki", 687 "ignore_whitespace_conflicts", 688 "internal_tracker", 689 "mirror_interval", 690 "name", 691 "private", 692 "template", 693 "website", 694 } 695 696 def commit(self): 697 args = {"owner": self.owner.username, "name": self.name} 698 self._commit(args) 699 700 def get_branches(self) -> List["Branch"]: 701 """Get all the Branches of this Repository.""" 702 703 results = self.allspice_client.requests_get_paginated( 704 Repository.REPO_BRANCHES % (self.owner.username, self.name) 705 ) 706 return [Branch.parse_response(self.allspice_client, result) for result in results] 707 708 def get_branch(self, name: str) -> "Branch": 709 """Get a specific Branch of this Repository.""" 710 result = self.allspice_client.requests_get( 711 Repository.REPO_BRANCH.format(owner=self.owner.username, repo=self.name, branch=name) 712 ) 713 return Branch.parse_response(self.allspice_client, result) 714 715 def add_branch(self, create_from: Ref, newname: str) -> "Branch": 716 """Add a branch to the repository""" 717 # Note: will only work with gitea 1.13 or higher! 718 719 ref_name = Util.data_params_for_ref(create_from) 720 if "ref" not in ref_name: 721 raise ValueError("create_from must be a Branch, Commit or string") 722 ref_name = ref_name["ref"] 723 724 data = {"new_branch_name": newname, "old_ref_name": ref_name} 725 result = self.allspice_client.requests_post( 726 Repository.REPO_BRANCHES % (self.owner.username, self.name), data=data 727 ) 728 return Branch.parse_response(self.allspice_client, result) 729 730 def get_issues( 731 self, 732 state: Literal["open", "closed", "all"] = "all", 733 search_query: Optional[str] = None, 734 labels: Optional[List[str]] = None, 735 milestones: Optional[List[Union[Milestone, str]]] = None, 736 assignee: Optional[Union[User, str]] = None, 737 since: Optional[datetime] = None, 738 before: Optional[datetime] = None, 739 ) -> List["Issue"]: 740 """ 741 Get all Issues of this Repository (open and closed) 742 743 https://hub.allspice.io/api/swagger#/repository/repoListIssues 744 745 All params of this method are optional filters. If you don't specify a filter, it 746 will not be applied. 747 748 :param state: The state of the Issues to get. If None, all Issues are returned. 749 :param search_query: Filter issues by text. This is equivalent to searching for 750 `search_query` in the Issues on the web interface. 751 :param labels: Filter issues by labels. 752 :param milestones: Filter issues by milestones. 753 :param assignee: Filter issues by the assigned user. 754 :param since: Filter issues by the date they were created. 755 :param before: Filter issues by the date they were created. 756 :return: A list of Issues. 757 """ 758 759 data = { 760 "state": state, 761 } 762 if search_query: 763 data["q"] = search_query 764 if labels: 765 data["labels"] = ",".join(labels) 766 if milestones: 767 data["milestone"] = ",".join( 768 [ 769 milestone.name if isinstance(milestone, Milestone) else milestone 770 for milestone in milestones 771 ] 772 ) 773 if assignee: 774 if isinstance(assignee, User): 775 data["assignee"] = assignee.username 776 else: 777 data["assignee"] = assignee 778 if since: 779 data["since"] = Util.format_time(since) 780 if before: 781 data["before"] = Util.format_time(before) 782 783 results = self.allspice_client.requests_get_paginated( 784 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 785 params=data, 786 ) 787 788 issues = [] 789 for result in results: 790 issue = Issue.parse_response(self.allspice_client, result) 791 # See Issue.request 792 setattr(issue, "_repository", self) 793 # This is mostly for compatibility with an older implementation 794 Issue._add_read_property("repo", self, issue) 795 issues.append(issue) 796 797 return issues 798 799 def get_design_reviews( 800 self, 801 state: Literal["open", "closed", "all"] = "all", 802 milestone: Optional[Union[Milestone, str]] = None, 803 labels: Optional[List[str]] = None, 804 ) -> List["DesignReview"]: 805 """ 806 Get all Design Reviews of this Repository. 807 808 https://hub.allspice.io/api/swagger#/repository/repoListPullRequests 809 810 :param state: The state of the Design Reviews to get. If None, all Design Reviews 811 are returned. 812 :param milestone: The milestone of the Design Reviews to get. 813 :param labels: A list of label IDs to filter DRs by. 814 :return: A list of Design Reviews. 815 """ 816 817 params = { 818 "state": state, 819 } 820 if milestone: 821 if isinstance(milestone, Milestone): 822 params["milestone"] = milestone.name 823 else: 824 params["milestone"] = milestone 825 if labels: 826 params["labels"] = ",".join(labels) 827 828 results = self.allspice_client.requests_get_paginated( 829 self.REPO_DESIGN_REVIEWS.format(owner=self.owner.username, repo=self.name), 830 params=params, 831 ) 832 return [DesignReview.parse_response(self.allspice_client, result) for result in results] 833 834 def get_commits( 835 self, 836 sha: Optional[str] = None, 837 path: Optional[str] = None, 838 stat: bool = True, 839 ) -> List["Commit"]: 840 """ 841 Get all the Commits of this Repository. 842 843 https://hub.allspice.io/api/swagger#/repository/repoGetAllCommits 844 845 :param sha: The SHA of the commit to start listing commits from. 846 :param path: filepath of a file/dir. 847 :param stat: Include the number of additions and deletions in the response. 848 Disable for speedup. 849 :return: A list of Commits. 850 """ 851 852 data = {} 853 if sha: 854 data["sha"] = sha 855 if path: 856 data["path"] = path 857 if not stat: 858 data["stat"] = False 859 860 try: 861 results = self.allspice_client.requests_get_paginated( 862 Repository.REPO_COMMITS % (self.owner.username, self.name), 863 params=data, 864 ) 865 except ConflictException as err: 866 logging.warning(err) 867 logging.warning("Repository %s/%s is Empty" % (self.owner.username, self.name)) 868 results = [] 869 return [Commit.parse_response(self.allspice_client, result) for result in results] 870 871 def get_issues_state(self, state) -> List["Issue"]: 872 """ 873 DEPRECATED: Use get_issues() instead. 874 875 Get issues of state Issue.open or Issue.closed of a repository. 876 """ 877 878 assert state in [Issue.OPENED, Issue.CLOSED] 879 issues = [] 880 data = {"state": state} 881 results = self.allspice_client.requests_get_paginated( 882 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 883 params=data, 884 ) 885 for result in results: 886 issue = Issue.parse_response(self.allspice_client, result) 887 # adding data not contained in the issue response 888 # See Issue.request() 889 setattr(issue, "_repository", self) 890 Issue._add_read_property("repo", self, issue) 891 Issue._add_read_property("owner", self.owner, issue) 892 issues.append(issue) 893 return issues 894 895 def get_times(self): 896 results = self.allspice_client.requests_get( 897 Repository.REPO_TIMES % (self.owner.username, self.name) 898 ) 899 return results 900 901 def get_user_time(self, username) -> float: 902 if isinstance(username, User): 903 username = username.username 904 results = self.allspice_client.requests_get( 905 Repository.REPO_USER_TIME % (self.owner.username, self.name, username) 906 ) 907 time = sum(r["time"] for r in results) 908 return time 909 910 def get_full_name(self) -> str: 911 return self.owner.username + "/" + self.name 912 913 def create_issue(self, title, assignees=frozenset(), description="") -> ApiObject: 914 data = { 915 "assignees": assignees, 916 "body": description, 917 "closed": False, 918 "title": title, 919 } 920 result = self.allspice_client.requests_post( 921 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 922 data=data, 923 ) 924 925 issue = Issue.parse_response(self.allspice_client, result) 926 setattr(issue, "_repository", self) 927 Issue._add_read_property("repo", self, issue) 928 return issue 929 930 def create_design_review( 931 self, 932 title: str, 933 head: Union[Branch, str], 934 base: Union[Branch, str], 935 assignees: Optional[Set[Union[User, str]]] = None, 936 body: Optional[str] = None, 937 due_date: Optional[datetime] = None, 938 milestone: Optional["Milestone"] = None, 939 ) -> "DesignReview": 940 """ 941 Create a new Design Review. 942 943 See https://hub.allspice.io/api/swagger#/repository/repoCreatePullRequest 944 945 :param title: Title of the Design Review 946 :param head: Branch or name of the branch to merge into the base branch 947 :param base: Branch or name of the branch to merge into 948 :param assignees: Optional. A list of users to assign this review. List can be of 949 User objects or of usernames. 950 :param body: An Optional Description for the Design Review. 951 :param due_date: An Optional Due date for the Design Review. 952 :param milestone: An Optional Milestone for the Design Review 953 :return: The created Design Review 954 """ 955 956 data: dict[str, Any] = { 957 "title": title, 958 } 959 960 if isinstance(head, Branch): 961 data["head"] = head.name 962 else: 963 data["head"] = head 964 if isinstance(base, Branch): 965 data["base"] = base.name 966 else: 967 data["base"] = base 968 if assignees: 969 data["assignees"] = [a.username if isinstance(a, User) else a for a in assignees] 970 if body: 971 data["body"] = body 972 if due_date: 973 data["due_date"] = Util.format_time(due_date) 974 if milestone: 975 data["milestone"] = milestone.id 976 977 result = self.allspice_client.requests_post( 978 self.REPO_DESIGN_REVIEWS.format(owner=self.owner.username, repo=self.name), 979 data=data, 980 ) 981 982 return DesignReview.parse_response(self.allspice_client, result) 983 984 def create_milestone( 985 self, 986 title: str, 987 description: str, 988 due_date: Optional[str] = None, 989 state: str = "open", 990 ) -> "Milestone": 991 url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) 992 data = {"title": title, "description": description, "state": state} 993 if due_date: 994 data["due_date"] = due_date 995 result = self.allspice_client.requests_post(url, data=data) 996 return Milestone.parse_response(self.allspice_client, result) 997 998 def create_gitea_hook(self, hook_url: str, events: List[str]): 999 url = f"/repos/{self.owner.username}/{self.name}/hooks" 1000 data = { 1001 "type": "gitea", 1002 "config": {"content_type": "json", "url": hook_url}, 1003 "events": events, 1004 "active": True, 1005 } 1006 return self.allspice_client.requests_post(url, data=data) 1007 1008 def list_hooks(self): 1009 url = f"/repos/{self.owner.username}/{self.name}/hooks" 1010 return self.allspice_client.requests_get(url) 1011 1012 def delete_hook(self, id: str): 1013 url = f"/repos/{self.owner.username}/{self.name}/hooks/{id}" 1014 self.allspice_client.requests_delete(url) 1015 1016 def is_collaborator(self, username) -> bool: 1017 if isinstance(username, User): 1018 username = username.username 1019 try: 1020 # returns 204 if its ok, 404 if its not 1021 self.allspice_client.requests_get( 1022 Repository.REPO_IS_COLLABORATOR % (self.owner.username, self.name, username) 1023 ) 1024 return True 1025 except Exception: 1026 return False 1027 1028 def get_users_with_access(self) -> Sequence[User]: 1029 url = f"/repos/{self.owner.username}/{self.name}/collaborators" 1030 response = self.allspice_client.requests_get(url) 1031 collabs = [User.parse_response(self.allspice_client, user) for user in response] 1032 if isinstance(self.owner, User): 1033 return [*collabs, self.owner] 1034 else: 1035 # owner must be org 1036 teams = self.owner.get_teams() 1037 for team in teams: 1038 team_repos = team.get_repos() 1039 if self.name in [n.name for n in team_repos]: 1040 collabs += team.get_members() 1041 return collabs 1042 1043 def remove_collaborator(self, user_name: str): 1044 url = f"/repos/{self.owner.username}/{self.name}/collaborators/{user_name}" 1045 self.allspice_client.requests_delete(url) 1046 1047 def transfer_ownership( 1048 self, 1049 new_owner: Union[User, Organization], 1050 new_teams: Set[Team] | FrozenSet[Team] = frozenset(), 1051 ): 1052 url = Repository.REPO_TRANSFER.format(owner=self.owner.username, repo=self.name) 1053 data: dict[str, Any] = {"new_owner": new_owner.username} 1054 if isinstance(new_owner, Organization): 1055 new_team_ids = [team.id for team in new_teams if team in new_owner.get_teams()] 1056 data["team_ids"] = new_team_ids 1057 self.allspice_client.requests_post(url, data=data) 1058 # TODO: make sure this instance is either updated or discarded 1059 1060 def get_git_content( 1061 self, 1062 ref: Optional["Ref"] = None, 1063 commit: "Optional[Commit]" = None, 1064 ) -> List[Content]: 1065 """ 1066 Get the metadata for all files in the root directory. 1067 1068 https://hub.allspice.io/api/swagger#/repository/repoGetContentsList 1069 1070 :param ref: branch or commit to get content from 1071 :param commit: commit to get content from (deprecated) 1072 """ 1073 url = f"/repos/{self.owner.username}/{self.name}/contents" 1074 data = Util.data_params_for_ref(ref or commit) 1075 1076 result = [ 1077 Content.parse_response(self.allspice_client, f) 1078 for f in self.allspice_client.requests_get(url, data) 1079 ] 1080 return result 1081 1082 def get_tree(self, ref: Optional[Ref] = None, recursive: bool = False) -> List[GitEntry]: 1083 """ 1084 Get the repository's tree on a given ref. 1085 1086 By default, this will only return the top-level entries in the tree. If you want 1087 to get the entire tree, set `recursive` to True. 1088 1089 :param ref: The ref to get the tree from. If not provided, the default branch is used. 1090 :param recursive: Whether to get the entire tree or just the top-level entries. 1091 """ 1092 1093 ref = Util.data_params_for_ref(ref).get("ref", self.default_branch) 1094 url = self.REPO_GET_TREE.format(owner=self.owner.username, repo=self.name, ref=ref) 1095 params = {"recursive": recursive} 1096 results = self.allspice_client.requests_get_paginated(url, params=params) 1097 return [GitEntry.parse_response(self.allspice_client, result) for result in results] 1098 1099 def get_file_content( 1100 self, 1101 content: Content, 1102 ref: Optional[Ref] = None, 1103 commit: Optional[Commit] = None, 1104 ) -> Union[str, List["Content"]]: 1105 """https://hub.allspice.io/api/swagger#/repository/repoGetContents""" 1106 url = f"/repos/{self.owner.username}/{self.name}/contents/{content.path}" 1107 data = Util.data_params_for_ref(ref or commit) 1108 1109 if content.type == Content.FILE: 1110 return self.allspice_client.requests_get(url, data)["content"] 1111 else: 1112 return [ 1113 Content.parse_response(self.allspice_client, f) 1114 for f in self.allspice_client.requests_get(url, data) 1115 ] 1116 1117 def get_raw_file( 1118 self, 1119 file_path: str, 1120 ref: Optional[Ref] = None, 1121 ) -> bytes: 1122 """ 1123 Get the raw, binary data of a single file. 1124 1125 Note 1: if the file you are requesting is a text file, you might want to 1126 use .decode() on the result to get a string. For example: 1127 1128 content = repo.get_raw_file("file.txt").decode("utf-8") 1129 1130 Note 2: this method will store the entire file in memory. If you want 1131 to download a large file, you might want to use `download_to_file` 1132 instead. 1133 1134 See https://hub.allspice.io/api/swagger#/repository/repoGetRawFileOrLFS 1135 1136 :param file_path: The path to the file to get. 1137 :param ref: The branch or commit to get the file from. If not provided, 1138 the default branch is used. 1139 """ 1140 1141 url = self.REPO_GET_MEDIA.format( 1142 owner=self.owner.username, 1143 repo=self.name, 1144 path=file_path, 1145 ) 1146 params = Util.data_params_for_ref(ref) 1147 return self.allspice_client.requests_get_raw(url, params=params) 1148 1149 def download_to_file( 1150 self, 1151 file_path: str, 1152 io: IO, 1153 ref: Optional[Ref] = None, 1154 ) -> None: 1155 """ 1156 Download the binary data of a file to a file-like object. 1157 1158 Example: 1159 1160 with open("schematic.DSN", "wb") as f: 1161 Repository.download_to_file("Schematics/my_schematic.DSN", f) 1162 1163 :param file_path: The path to the file in the repository from the root 1164 of the repository. 1165 :param io: The file-like object to write the data to. 1166 """ 1167 1168 url = self.allspice_client._AllSpice__get_url( 1169 self.REPO_GET_MEDIA.format( 1170 owner=self.owner.username, 1171 repo=self.name, 1172 path=file_path, 1173 ) 1174 ) 1175 params = Util.data_params_for_ref(ref) 1176 response = self.allspice_client.requests.get( 1177 url, 1178 params=params, 1179 headers=self.allspice_client.headers, 1180 stream=True, 1181 ) 1182 1183 for chunk in response.iter_content(chunk_size=4096): 1184 if chunk: 1185 io.write(chunk) 1186 1187 def get_generated_json(self, content: Union[Content, str], ref: Optional[Ref] = None) -> dict: 1188 """ 1189 Get the json blob for a cad file if it exists, otherwise enqueue 1190 a new job and return a 503 status. 1191 1192 WARNING: This is still experimental and not recommended for critical 1193 applications. The structure and content of the returned dictionary can 1194 change at any time. 1195 1196 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceJSON 1197 """ 1198 1199 if isinstance(content, Content): 1200 content = content.path 1201 1202 url = self.REPO_GET_ALLSPICE_JSON.format( 1203 owner=self.owner.username, 1204 repo=self.name, 1205 content=content, 1206 ) 1207 data = Util.data_params_for_ref(ref) 1208 return self.allspice_client.requests_get(url, data) 1209 1210 def get_generated_svg(self, content: Union[Content, str], ref: Optional[Ref] = None) -> bytes: 1211 """ 1212 Get the svg blob for a cad file if it exists, otherwise enqueue 1213 a new job and return a 503 status. 1214 1215 WARNING: This is still experimental and not yet recommended for 1216 critical applications. The content of the returned svg can change 1217 at any time. 1218 1219 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceSVG 1220 """ 1221 1222 if isinstance(content, Content): 1223 content = content.path 1224 1225 url = self.REPO_GET_ALLSPICE_SVG.format( 1226 owner=self.owner.username, 1227 repo=self.name, 1228 content=content, 1229 ) 1230 data = Util.data_params_for_ref(ref) 1231 return self.allspice_client.requests_get_raw(url, data) 1232 1233 def get_generated_projectdata( 1234 self, content: Union[Content, str], ref: Optional[Ref] = None 1235 ) -> dict: 1236 """ 1237 Get the json project data based on the cad file provided 1238 1239 WARNING: This is still experimental and not yet recommended for 1240 critical applications. The content of the returned dictionary can change 1241 at any time. 1242 1243 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceProject 1244 """ 1245 if isinstance(content, Content): 1246 content = content.path 1247 1248 url = self.REPO_GET_ALLSPICE_PROJECT.format( 1249 owner=self.owner.username, 1250 repo=self.name, 1251 content=content, 1252 ) 1253 data = Util.data_params_for_ref(ref) 1254 return self.allspice_client.requests_get(url, data) 1255 1256 def create_file(self, file_path: str, content: str, data: Optional[dict] = None): 1257 """https://hub.allspice.io/api/swagger#/repository/repoCreateFile""" 1258 if not data: 1259 data = {} 1260 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1261 data.update({"content": content}) 1262 return self.allspice_client.requests_post(url, data) 1263 1264 def change_file(self, file_path: str, file_sha: str, content: str, data: Optional[dict] = None): 1265 """https://hub.allspice.io/api/swagger#/repository/repoCreateFile""" 1266 if not data: 1267 data = {} 1268 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1269 data.update({"sha": file_sha, "content": content}) 1270 return self.allspice_client.requests_put(url, data) 1271 1272 def delete_file(self, file_path: str, file_sha: str, data: Optional[dict] = None): 1273 """https://hub.allspice.io/api/swagger#/repository/repoDeleteFile""" 1274 if not data: 1275 data = {} 1276 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1277 data.update({"sha": file_sha}) 1278 return self.allspice_client.requests_delete(url, data) 1279 1280 def get_archive( 1281 self, 1282 ref: Ref = "main", 1283 archive_format: ArchiveFormat = ArchiveFormat.ZIP, 1284 ) -> bytes: 1285 """ 1286 Download all the files in a specific ref of a repository as a zip or tarball 1287 archive. 1288 1289 https://hub.allspice.io/api/swagger#/repository/repoGetArchive 1290 1291 :param ref: branch or commit to get content from, defaults to the "main" branch 1292 :param archive_format: zip or tar, defaults to zip 1293 """ 1294 1295 ref_string = Util.data_params_for_ref(ref)["ref"] 1296 url = self.REPO_GET_ARCHIVE.format( 1297 owner=self.owner.username, 1298 repo=self.name, 1299 ref=ref_string, 1300 format=archive_format.value, 1301 ) 1302 return self.allspice_client.requests_get_raw(url) 1303 1304 def get_topics(self) -> list[str]: 1305 """ 1306 Gets the list of topics on this repository. 1307 1308 See http://localhost:3000/api/swagger#/repository/repoListTopics 1309 """ 1310 1311 url = self.REPO_GET_TOPICS.format( 1312 owner=self.owner.username, 1313 repo=self.name, 1314 ) 1315 return self.allspice_client.requests_get(url)["topics"] 1316 1317 def add_topic(self, topic: str): 1318 """ 1319 Adds a topic to the repository. 1320 1321 See https://hub.allspice.io/api/swagger#/repository/repoAddTopic 1322 1323 :param topic: The topic to add. Topic names must consist only of 1324 lowercase letters, numnbers and dashes (-), and cannot start with 1325 dashes. Topic names also must be under 35 characters long. 1326 """ 1327 1328 url = self.REPO_ADD_TOPIC.format(owner=self.owner.username, repo=self.name, topic=topic) 1329 self.allspice_client.requests_put(url) 1330 1331 def create_release( 1332 self, 1333 tag_name: str, 1334 name: Optional[str] = None, 1335 body: Optional[str] = None, 1336 draft: bool = False, 1337 ): 1338 """ 1339 Create a release for this repository. The release will be created for 1340 the tag with the given name. If there is no tag with this name, create 1341 the tag first. 1342 1343 See https://hub.allspice.io/api/swagger#/repository/repoCreateRelease 1344 """ 1345 1346 url = self.REPO_GET_RELEASES.format(owner=self.owner.username, repo=self.name) 1347 data = { 1348 "tag_name": tag_name, 1349 "draft": draft, 1350 } 1351 if name is not None: 1352 data["name"] = name 1353 if body is not None: 1354 data["body"] = body 1355 response = self.allspice_client.requests_post(url, data) 1356 return Release.parse_response(self.allspice_client, response, self) 1357 1358 def get_releases( 1359 self, draft: Optional[bool] = None, pre_release: Optional[bool] = None 1360 ) -> List[Release]: 1361 """ 1362 Get the list of releases for this repository. 1363 1364 See https://hub.allspice.io/api/swagger#/repository/repoListReleases 1365 """ 1366 1367 data = {} 1368 1369 if draft is not None: 1370 data["draft"] = draft 1371 if pre_release is not None: 1372 data["pre-release"] = pre_release 1373 1374 url = self.REPO_GET_RELEASES.format(owner=self.owner.username, repo=self.name) 1375 responses = self.allspice_client.requests_get_paginated(url, params=data) 1376 1377 return [ 1378 Release.parse_response(self.allspice_client, response, self) for response in responses 1379 ] 1380 1381 def get_latest_release(self) -> Release: 1382 """ 1383 Get the latest release for this repository. 1384 1385 See https://hub.allspice.io/api/swagger#/repository/repoGetLatestRelease 1386 """ 1387 1388 url = self.REPO_GET_LATEST_RELEASE.format(owner=self.owner.username, repo=self.name) 1389 response = self.allspice_client.requests_get(url) 1390 release = Release.parse_response(self.allspice_client, response, self) 1391 return release 1392 1393 def get_release_by_tag(self, tag: str) -> Release: 1394 """ 1395 Get a release by its tag. 1396 1397 See https://hub.allspice.io/api/swagger#/repository/repoGetReleaseByTag 1398 """ 1399 1400 url = self.REPO_GET_RELEASE_BY_TAG.format( 1401 owner=self.owner.username, repo=self.name, tag=tag 1402 ) 1403 response = self.allspice_client.requests_get(url) 1404 release = Release.parse_response(self.allspice_client, response, self) 1405 return release 1406 1407 def get_commit_statuses( 1408 self, 1409 commit: Union[str, Commit], 1410 sort: Optional[CommitStatusSort] = None, 1411 state: Optional[CommitStatusState] = None, 1412 ) -> List[CommitStatus]: 1413 """ 1414 Get a list of statuses for a commit. 1415 1416 This is roughly equivalent to the Commit.get_statuses method, but this 1417 method allows you to sort and filter commits and is more convenient if 1418 you have a commit SHA and don't need to get the commit itself. 1419 1420 See https://hub.allspice.io/api/swagger#/repository/repoListStatuses 1421 """ 1422 1423 if isinstance(commit, Commit): 1424 commit = commit.sha 1425 1426 params = {} 1427 if sort is not None: 1428 params["sort"] = sort.value 1429 if state is not None: 1430 params["state"] = state.value 1431 1432 url = self.REPO_GET_COMMIT_STATUS.format( 1433 owner=self.owner.username, repo=self.name, sha=commit 1434 ) 1435 response = self.allspice_client.requests_get_paginated(url, params=params) 1436 return [CommitStatus.parse_response(self.allspice_client, status) for status in response] 1437 1438 def create_commit_status( 1439 self, 1440 commit: Union[str, Commit], 1441 context: Optional[str] = None, 1442 description: Optional[str] = None, 1443 state: Optional[CommitStatusState] = None, 1444 target_url: Optional[str] = None, 1445 ) -> CommitStatus: 1446 """ 1447 Create a status on a commit. 1448 1449 See https://hub.allspice.io/api/swagger#/repository/repoCreateStatus 1450 """ 1451 1452 if isinstance(commit, Commit): 1453 commit = commit.sha 1454 1455 data = {} 1456 if context is not None: 1457 data["context"] = context 1458 if description is not None: 1459 data["description"] = description 1460 if state is not None: 1461 data["state"] = state.value 1462 if target_url is not None: 1463 data["target_url"] = target_url 1464 1465 url = self.REPO_GET_COMMIT_STATUS.format( 1466 owner=self.owner.username, repo=self.name, sha=commit 1467 ) 1468 response = self.allspice_client.requests_post(url, data=data) 1469 return CommitStatus.parse_response(self.allspice_client, response) 1470 1471 def delete(self): 1472 self.allspice_client.requests_delete( 1473 Repository.REPO_DELETE % (self.owner.username, self.name) 1474 ) 1475 self.deleted = True 1476 1477 1478class Milestone(ApiObject): 1479 allow_merge_commits: Any 1480 allow_rebase: Any 1481 allow_rebase_explicit: Any 1482 allow_squash_merge: Any 1483 archived: Any 1484 closed_at: Any 1485 closed_issues: int 1486 created_at: str 1487 default_branch: Any 1488 description: str 1489 due_on: Any 1490 has_issues: Any 1491 has_pull_requests: Any 1492 has_wiki: Any 1493 id: int 1494 ignore_whitespace_conflicts: Any 1495 name: Any 1496 open_issues: int 1497 private: Any 1498 state: str 1499 title: str 1500 updated_at: str 1501 website: Any 1502 1503 API_OBJECT = """/repos/{owner}/{repo}/milestones/{number}""" # <owner, repo> 1504 1505 def __init__(self, allspice_client): 1506 super().__init__(allspice_client) 1507 1508 def __eq__(self, other): 1509 if not isinstance(other, Milestone): 1510 return False 1511 return self.allspice_client == other.allspice_client and self.id == other.id 1512 1513 def __hash__(self): 1514 return hash(self.allspice_client) ^ hash(self.id) 1515 1516 _fields_to_parsers: ClassVar[dict] = { 1517 "closed_at": lambda _, t: Util.convert_time(t), 1518 "due_on": lambda _, t: Util.convert_time(t), 1519 } 1520 1521 _patchable_fields: ClassVar[set[str]] = { 1522 "allow_merge_commits", 1523 "allow_rebase", 1524 "allow_rebase_explicit", 1525 "allow_squash_merge", 1526 "archived", 1527 "default_branch", 1528 "description", 1529 "has_issues", 1530 "has_pull_requests", 1531 "has_wiki", 1532 "ignore_whitespace_conflicts", 1533 "name", 1534 "private", 1535 "website", 1536 } 1537 1538 @classmethod 1539 def request(cls, allspice_client, owner: str, repo: str, number: str): 1540 return cls._request(allspice_client, {"owner": owner, "repo": repo, "number": number}) 1541 1542 1543class Attachment(ReadonlyApiObject): 1544 """ 1545 An asset attached to a comment. 1546 1547 You cannot edit or delete the attachment from this object - see the instance methods 1548 Comment.edit_attachment and delete_attachment for that. 1549 """ 1550 1551 browser_download_url: str 1552 created_at: str 1553 download_count: int 1554 id: int 1555 name: str 1556 size: int 1557 uuid: str 1558 1559 def __init__(self, allspice_client): 1560 super().__init__(allspice_client) 1561 1562 def __eq__(self, other): 1563 if not isinstance(other, Attachment): 1564 return False 1565 1566 return self.uuid == other.uuid 1567 1568 def __hash__(self): 1569 return hash(self.uuid) 1570 1571 def download_to_file(self, io: IO): 1572 """ 1573 Download the raw, binary data of this Attachment to a file-like object. 1574 1575 Example: 1576 1577 with open("my_file.zip", "wb") as f: 1578 attachment.download_to_file(f) 1579 1580 :param io: The file-like object to write the data to. 1581 """ 1582 1583 response = self.allspice_client.requests.get( 1584 self.browser_download_url, 1585 headers=self.allspice_client.headers, 1586 stream=True, 1587 ) 1588 # 4kb chunks 1589 for chunk in response.iter_content(chunk_size=4096): 1590 if chunk: 1591 io.write(chunk) 1592 1593 1594class Comment(ApiObject): 1595 assets: List[Union[Any, Dict[str, Union[int, str]]]] 1596 body: str 1597 created_at: datetime 1598 html_url: str 1599 id: int 1600 issue_url: str 1601 original_author: str 1602 original_author_id: int 1603 pull_request_url: str 1604 updated_at: datetime 1605 user: User 1606 1607 API_OBJECT = """/repos/{owner}/{repo}/issues/comments/{id}""" 1608 GET_ATTACHMENTS_PATH = """/repos/{owner}/{repo}/issues/comments/{id}/assets""" 1609 ATTACHMENT_PATH = """/repos/{owner}/{repo}/issues/comments/{id}/assets/{attachment_id}""" 1610 1611 def __init__(self, allspice_client): 1612 super().__init__(allspice_client) 1613 1614 def __eq__(self, other): 1615 if not isinstance(other, Comment): 1616 return False 1617 return self.repository == other.repository and self.id == other.id 1618 1619 def __hash__(self): 1620 return hash(self.repository) ^ hash(self.id) 1621 1622 @classmethod 1623 def request(cls, allspice_client, owner: str, repo: str, id: str) -> "Comment": 1624 return cls._request(allspice_client, {"owner": owner, "repo": repo, "id": id}) 1625 1626 _fields_to_parsers: ClassVar[dict] = { 1627 "user": lambda allspice_client, r: User.parse_response(allspice_client, r), 1628 "created_at": lambda _, t: Util.convert_time(t), 1629 "updated_at": lambda _, t: Util.convert_time(t), 1630 } 1631 1632 _patchable_fields: ClassVar[set[str]] = {"body"} 1633 1634 @property 1635 def parent_url(self) -> str: 1636 """URL of the parent of this comment (the issue or the pull request)""" 1637 1638 if self.issue_url is not None and self.issue_url != "": 1639 return self.issue_url 1640 else: 1641 return self.pull_request_url 1642 1643 @cached_property 1644 def repository(self) -> Repository: 1645 """The repository this comment was posted on.""" 1646 1647 owner_name, repo_name = self.parent_url.split("/")[-4:-2] 1648 return Repository.request(self.allspice_client, owner_name, repo_name) 1649 1650 def __fields_for_path(self): 1651 return { 1652 "owner": self.repository.owner.username, 1653 "repo": self.repository.name, 1654 "id": self.id, 1655 } 1656 1657 def commit(self): 1658 self._commit(self.__fields_for_path()) 1659 1660 def delete(self): 1661 self.allspice_client.requests_delete(self.API_OBJECT.format(**self.__fields_for_path())) 1662 self.deleted = True 1663 1664 def get_attachments(self) -> List[Attachment]: 1665 """ 1666 Get all attachments on this comment. This returns Attachment objects, which 1667 contain a link to download the attachment. 1668 1669 https://hub.allspice.io/api/swagger#/issue/issueListIssueCommentAttachments 1670 """ 1671 1672 results = self.allspice_client.requests_get( 1673 self.GET_ATTACHMENTS_PATH.format(**self.__fields_for_path()) 1674 ) 1675 return [Attachment.parse_response(self.allspice_client, result) for result in results] 1676 1677 def create_attachment(self, file: IO, name: Optional[str] = None) -> Attachment: 1678 """ 1679 Create an attachment on this comment. 1680 1681 https://hub.allspice.io/api/swagger#/issue/issueCreateIssueCommentAttachment 1682 1683 :param file: The file to attach. This should be a file-like object. 1684 :param name: The name of the file. If not provided, the name of the file will be 1685 used. 1686 :return: The created attachment. 1687 """ 1688 1689 args: dict[str, Any] = { 1690 "files": {"attachment": file}, 1691 } 1692 if name is not None: 1693 args["params"] = {"name": name} 1694 1695 result = self.allspice_client.requests_post( 1696 self.GET_ATTACHMENTS_PATH.format(**self.__fields_for_path()), 1697 **args, 1698 ) 1699 return Attachment.parse_response(self.allspice_client, result) 1700 1701 def edit_attachment(self, attachment: Attachment, data: dict) -> Attachment: 1702 """ 1703 Edit an attachment. 1704 1705 The list of params that can be edited is available at 1706 https://hub.allspice.io/api/swagger#/issue/issueEditIssueCommentAttachment 1707 1708 :param attachment: The attachment to be edited 1709 :param data: The data parameter should be a dictionary of the fields to edit. 1710 :return: The edited attachment 1711 """ 1712 1713 args = { 1714 **self.__fields_for_path(), 1715 "attachment_id": attachment.id, 1716 } 1717 result = self.allspice_client.requests_patch( 1718 self.ATTACHMENT_PATH.format(**args), 1719 data=data, 1720 ) 1721 return Attachment.parse_response(self.allspice_client, result) 1722 1723 def delete_attachment(self, attachment: Attachment): 1724 """https://hub.allspice.io/api/swagger#/issue/issueDeleteIssueCommentAttachment""" 1725 1726 args = { 1727 **self.__fields_for_path(), 1728 "attachment_id": attachment.id, 1729 } 1730 self.allspice_client.requests_delete(self.ATTACHMENT_PATH.format(**args)) 1731 attachment.deleted = True 1732 1733 1734class Commit(ReadonlyApiObject): 1735 author: User 1736 commit: Dict[str, Union[str, Dict[str, str], Dict[str, Optional[Union[bool, str]]]]] 1737 committer: Dict[str, Union[int, str, bool]] 1738 created: str 1739 files: List[Dict[str, str]] 1740 html_url: str 1741 inner_commit: Dict[str, Union[str, Dict[str, str], Dict[str, Optional[Union[bool, str]]]]] 1742 parents: List[Union[Dict[str, str], Any]] 1743 sha: str 1744 stats: Dict[str, int] 1745 url: str 1746 1747 API_OBJECT = """/repos/{owner}/{repo}/commits/{sha}""" 1748 COMMIT_GET_STATUS = """/repos/{owner}/{repo}/commits/{sha}/status""" 1749 COMMIT_GET_STATUSES = """/repos/{owner}/{repo}/commits/{sha}/statuses""" 1750 1751 # Regex to extract owner and repo names from the url property 1752 URL_REGEXP = re.compile(r"/repos/([^/]+)/([^/]+)/git/commits") 1753 1754 def __init__(self, allspice_client): 1755 super().__init__(allspice_client) 1756 1757 _fields_to_parsers: ClassVar[dict] = { 1758 # NOTE: api may return None for commiters that are no allspice users 1759 "author": lambda allspice_client, u: User.parse_response(allspice_client, u) if u else None 1760 } 1761 1762 def __eq__(self, other): 1763 if not isinstance(other, Commit): 1764 return False 1765 return self.sha == other.sha 1766 1767 def __hash__(self): 1768 return hash(self.sha) 1769 1770 @classmethod 1771 def parse_response(cls, allspice_client, result) -> "Commit": 1772 commit_cache = result["commit"] 1773 api_object = cls(allspice_client) 1774 cls._initialize(allspice_client, api_object, result) 1775 # inner_commit for legacy reasons 1776 Commit._add_read_property("inner_commit", commit_cache, api_object) 1777 return api_object 1778 1779 def get_status(self) -> CommitCombinedStatus: 1780 """ 1781 Get a combined status consisting of all statues on this commit. 1782 1783 Note that the returned object is a CommitCombinedStatus object, which 1784 also contains a list of all statuses on the commit. 1785 1786 https://hub.allspice.io/api/swagger#/repository/repoGetCommitStatus 1787 """ 1788 1789 result = self.allspice_client.requests_get( 1790 self.COMMIT_GET_STATUS.format(**self._fields_for_path) 1791 ) 1792 return CommitCombinedStatus.parse_response(self.allspice_client, result) 1793 1794 def get_statuses(self) -> List[CommitStatus]: 1795 """ 1796 Get a list of all statuses on this commit. 1797 1798 https://hub.allspice.io/api/swagger#/repository/repoListCommitStatuses 1799 """ 1800 1801 results = self.allspice_client.requests_get( 1802 self.COMMIT_GET_STATUSES.format(**self._fields_for_path) 1803 ) 1804 return [CommitStatus.parse_response(self.allspice_client, result) for result in results] 1805 1806 @cached_property 1807 def _fields_for_path(self) -> dict[str, str]: 1808 matches = self.URL_REGEXP.search(self.url) 1809 if not matches: 1810 raise ValueError(f"Invalid commit URL: {self.url}") 1811 1812 return { 1813 "owner": matches.group(1), 1814 "repo": matches.group(2), 1815 "sha": self.sha, 1816 } 1817 1818 1819class CommitStatusState(Enum): 1820 PENDING = "pending" 1821 SUCCESS = "success" 1822 ERROR = "error" 1823 FAILURE = "failure" 1824 WARNING = "warning" 1825 1826 @classmethod 1827 def try_init(cls, value: str) -> CommitStatusState | str: 1828 """ 1829 Try converting a string to the enum, and if that fails, return the 1830 string itself. 1831 """ 1832 1833 try: 1834 return cls(value) 1835 except ValueError: 1836 return value 1837 1838 1839class CommitStatus(ReadonlyApiObject): 1840 context: str 1841 created_at: str 1842 creator: User 1843 description: str 1844 id: int 1845 status: CommitStatusState 1846 target_url: str 1847 updated_at: str 1848 url: str 1849 1850 def __init__(self, allspice_client): 1851 super().__init__(allspice_client) 1852 1853 _fields_to_parsers: ClassVar[dict] = { 1854 # Gitea/ASH doesn't actually validate that the status is a "valid" 1855 # status, so we can expect empty or unknown strings in the status field. 1856 "status": lambda _, s: CommitStatusState.try_init(s), 1857 "creator": lambda allspice_client, u: ( 1858 User.parse_response(allspice_client, u) if u else None 1859 ), 1860 } 1861 1862 def __eq__(self, other): 1863 if not isinstance(other, CommitStatus): 1864 return False 1865 return self.id == other.id 1866 1867 def __hash__(self): 1868 return hash(self.id) 1869 1870 1871class CommitCombinedStatus(ReadonlyApiObject): 1872 commit_url: str 1873 repository: Repository 1874 sha: str 1875 state: CommitStatusState 1876 statuses: List["CommitStatus"] 1877 total_count: int 1878 url: str 1879 1880 def __init__(self, allspice_client): 1881 super().__init__(allspice_client) 1882 1883 _fields_to_parsers: ClassVar[dict] = { 1884 # See CommitStatus 1885 "state": lambda _, s: CommitStatusState.try_init(s), 1886 "statuses": lambda allspice_client, statuses: [ 1887 CommitStatus.parse_response(allspice_client, status) for status in statuses 1888 ], 1889 "repository": lambda allspice_client, r: Repository.parse_response(allspice_client, r), 1890 } 1891 1892 def __eq__(self, other): 1893 if not isinstance(other, CommitCombinedStatus): 1894 return False 1895 return self.sha == other.sha 1896 1897 def __hash__(self): 1898 return hash(self.sha) 1899 1900 @classmethod 1901 def parse_response(cls, allspice_client, result) -> "CommitCombinedStatus": 1902 api_object = cls(allspice_client) 1903 cls._initialize(allspice_client, api_object, result) 1904 return api_object 1905 1906 1907class Issue(ApiObject): 1908 """ 1909 An issue on a repository. 1910 1911 Note: `Issue.assets` may not have any entries even if the issue has 1912 attachments. This happens when an issue is fetched via a bulk method like 1913 `Repository.get_issues`. In most cases, prefer using 1914 `Issue.get_attachments` to get the attachments on an issue. 1915 """ 1916 1917 assets: List[Union[Any, "Attachment"]] 1918 assignee: Any 1919 assignees: Any 1920 body: str 1921 closed_at: Any 1922 comments: int 1923 created_at: str 1924 due_date: Any 1925 html_url: str 1926 id: int 1927 is_locked: bool 1928 labels: List[Any] 1929 milestone: Optional["Milestone"] 1930 number: int 1931 original_author: str 1932 original_author_id: int 1933 pin_order: int 1934 pull_request: Any 1935 ref: str 1936 repository: Dict[str, Union[int, str]] 1937 state: str 1938 title: str 1939 updated_at: str 1940 url: str 1941 user: User 1942 1943 API_OBJECT = """/repos/{owner}/{repo}/issues/{index}""" # <owner, repo, index> 1944 GET_TIME = """/repos/%s/%s/issues/%s/times""" # <owner, repo, index> 1945 GET_COMMENTS = """/repos/{owner}/{repo}/issues/{index}/comments""" 1946 CREATE_ISSUE = """/repos/{owner}/{repo}/issues""" 1947 GET_ATTACHMENTS = """/repos/{owner}/{repo}/issues/{index}/assets""" 1948 1949 OPENED = "open" 1950 CLOSED = "closed" 1951 1952 def __init__(self, allspice_client): 1953 super().__init__(allspice_client) 1954 1955 def __eq__(self, other): 1956 if not isinstance(other, Issue): 1957 return False 1958 return self.repository == other.repository and self.id == other.id 1959 1960 def __hash__(self): 1961 return hash(self.repository) ^ hash(self.id) 1962 1963 _fields_to_parsers: ClassVar[dict] = { 1964 "milestone": lambda allspice_client, m: Milestone.parse_response(allspice_client, m), 1965 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 1966 "assets": lambda allspice_client, assets: [ 1967 Attachment.parse_response(allspice_client, a) for a in assets 1968 ], 1969 "assignee": lambda allspice_client, u: User.parse_response(allspice_client, u), 1970 "assignees": lambda allspice_client, us: [ 1971 User.parse_response(allspice_client, u) for u in us 1972 ], 1973 "state": lambda _, s: Issue.CLOSED if s == "closed" else Issue.OPENED, 1974 } 1975 1976 _parsers_to_fields: ClassVar[dict] = { 1977 "milestone": lambda m: m.id, 1978 } 1979 1980 _patchable_fields: ClassVar[set[str]] = { 1981 "assignee", 1982 "assignees", 1983 "body", 1984 "due_date", 1985 "milestone", 1986 "state", 1987 "title", 1988 } 1989 1990 def commit(self): 1991 args = { 1992 "owner": self.repository.owner.username, 1993 "repo": self.repository.name, 1994 "index": self.number, 1995 } 1996 self._commit(args) 1997 1998 @classmethod 1999 def request(cls, allspice_client, owner: str, repo: str, number: str): 2000 api_object = cls._request(allspice_client, {"owner": owner, "repo": repo, "index": number}) 2001 # The repository in the response is a RepositoryMeta object, so request 2002 # the full repository object and add it to the issue object. 2003 repository = Repository.request(allspice_client, owner, repo) 2004 setattr(api_object, "_repository", repository) 2005 # For legacy reasons 2006 cls._add_read_property("repo", repository, api_object) 2007 return api_object 2008 2009 @classmethod 2010 def create_issue(cls, allspice_client, repo: Repository, title: str, body: str = ""): 2011 args = {"owner": repo.owner.username, "repo": repo.name} 2012 data = {"title": title, "body": body} 2013 result = allspice_client.requests_post(Issue.CREATE_ISSUE.format(**args), data=data) 2014 issue = Issue.parse_response(allspice_client, result) 2015 setattr(issue, "_repository", repo) 2016 cls._add_read_property("repo", repo, issue) 2017 return issue 2018 2019 @property 2020 def owner(self) -> Organization | User: 2021 return self.repository.owner 2022 2023 def get_time_sum(self, user: User) -> int: 2024 results = self.allspice_client.requests_get( 2025 Issue.GET_TIME % (self.owner.username, self.repository.name, self.number) 2026 ) 2027 return sum(result["time"] for result in results if result and result["user_id"] == user.id) 2028 2029 def get_times(self) -> Optional[Dict]: 2030 return self.allspice_client.requests_get( 2031 Issue.GET_TIME % (self.owner.username, self.repository.name, self.number) 2032 ) 2033 2034 def delete_time(self, time_id: str): 2035 path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times/{time_id}" 2036 self.allspice_client.requests_delete(path) 2037 2038 def add_time(self, time: int, created: Optional[str] = None, user_name: Optional[User] = None): 2039 path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times" 2040 self.allspice_client.requests_post( 2041 path, data={"created": created, "time": int(time), "user_name": user_name} 2042 ) 2043 2044 def get_comments(self) -> List[Comment]: 2045 """https://hub.allspice.io/api/swagger#/issue/issueGetComments""" 2046 2047 results = self.allspice_client.requests_get( 2048 self.GET_COMMENTS.format( 2049 owner=self.owner.username, repo=self.repository.name, index=self.number 2050 ) 2051 ) 2052 2053 return [Comment.parse_response(self.allspice_client, result) for result in results] 2054 2055 def create_comment(self, body: str) -> Comment: 2056 """https://hub.allspice.io/api/swagger#/issue/issueCreateComment""" 2057 2058 path = self.GET_COMMENTS.format( 2059 owner=self.owner.username, repo=self.repository.name, index=self.number 2060 ) 2061 2062 response = self.allspice_client.requests_post(path, data={"body": body}) 2063 return Comment.parse_response(self.allspice_client, response) 2064 2065 def get_attachments(self) -> List[Attachment]: 2066 """ 2067 Fetch all attachments on this issue. 2068 2069 Unlike the assets field, this will always fetch all attachments from the 2070 API. 2071 2072 See https://hub.allspice.io/api/swagger#/issue/issueListIssueAttachments 2073 """ 2074 2075 path = self.GET_ATTACHMENTS.format( 2076 owner=self.owner.username, repo=self.repository.name, index=self.number 2077 ) 2078 response = self.allspice_client.requests_get(path) 2079 2080 return [Attachment.parse_response(self.allspice_client, result) for result in response] 2081 2082 def create_attachment(self, file: IO, name: Optional[str] = None) -> Attachment: 2083 """ 2084 Create an attachment on this issue. 2085 2086 https://hub.allspice.io/api/swagger#/issue/issueCreateIssueAttachment 2087 2088 :param file: The file to attach. This should be a file-like object. 2089 :param name: The name of the file. If not provided, the name of the file will be 2090 used. 2091 :return: The created attachment. 2092 """ 2093 2094 args: dict[str, Any] = { 2095 "files": {"attachment": file}, 2096 } 2097 if name is not None: 2098 args["params"] = {"name": name} 2099 2100 result = self.allspice_client.requests_post( 2101 self.GET_ATTACHMENTS.format( 2102 owner=self.owner.username, repo=self.repository.name, index=self.number 2103 ), 2104 **args, 2105 ) 2106 2107 return Attachment.parse_response(self.allspice_client, result) 2108 2109 2110class DesignReviewReviewComment(ApiObject): 2111 """ 2112 A comment on a Design Review Review. 2113 """ 2114 2115 body: str 2116 commit_id: str 2117 created_at: str 2118 diff_hunk: str 2119 html_url: str 2120 id: int 2121 original_commit_id: str 2122 original_position: int 2123 path: str 2124 position: int 2125 pull_request_review_id: int 2126 pull_request_url: str 2127 resolver: Any 2128 sub_path: str 2129 updated_at: str 2130 user: User 2131 2132 def __init__(self, allspice_client): 2133 super().__init__(allspice_client) 2134 2135 _fields_to_parsers: ClassVar[dict] = { 2136 "resolver": lambda allspice_client, r: User.parse_response(allspice_client, r), 2137 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 2138 } 2139 2140 2141class DesignReviewReview(ReadonlyApiObject): 2142 """ 2143 A review on a Design Review. 2144 """ 2145 2146 body: str 2147 comments_count: int 2148 commit_id: str 2149 dismissed: bool 2150 html_url: str 2151 id: int 2152 official: bool 2153 pull_request_url: str 2154 stale: bool 2155 state: ReviewEvent 2156 submitted_at: str 2157 team: Any 2158 updated_at: str 2159 user: User 2160 2161 API_OBJECT = "/repos/{owner}/{repo}/pulls/{index}/reviews/{id}" 2162 GET_COMMENTS = "/repos/{owner}/{repo}/pulls/{index}/reviews/{id}/comments" 2163 2164 class ReviewEvent(Enum): 2165 APPROVED = "APPROVED" 2166 PENDING = "PENDING" 2167 COMMENT = "COMMENT" 2168 REQUEST_CHANGES = "REQUEST_CHANGES" 2169 REQUEST_REVIEW = "REQUEST_REVIEW" 2170 UNKNOWN = "" 2171 2172 @dataclass 2173 class ReviewComment: 2174 """ 2175 Data required to create a review comment on a design review. 2176 2177 :param body: The body of the comment. 2178 :param path: The path of the file to comment on. If you have a 2179 `Content` object, get the path using the `path` property. 2180 :param sub_path: The sub-path of the file to comment on. This is 2181 usually the page ID of the page in the multi-page document. 2182 :param new_position: The line number of the source code file after the 2183 change to add this comment on. Optional, leave unset if this is an ECAD 2184 file or the comment must be on the entire file. 2185 :param old_position: The line number of the source code file before the 2186 change to add this comment on. Optional, leave unset if this is an ECAD 2187 file or the comment must be on the entire file. 2188 """ 2189 2190 body: str 2191 path: str 2192 sub_path: Optional[str] = None 2193 new_position: Optional[int] = None 2194 old_position: Optional[int] = None 2195 2196 def __init__(self, allspice_client): 2197 super().__init__(allspice_client) 2198 2199 _fields_to_parsers: ClassVar[dict] = { 2200 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 2201 "state": lambda _, s: DesignReviewReview.ReviewEvent(s), 2202 } 2203 2204 def _get_dr_properties(self) -> dict[str, str]: 2205 """ 2206 Get the owner, repo name and design review number from the URL of this 2207 review's DR. 2208 """ 2209 2210 parts = self.pull_request_url.strip("/").split("/") 2211 2212 try: 2213 index = parts[-1] 2214 assert parts[-2] == "pulls" or parts[-2] == "pull", ( 2215 "Expected the second last part of the URL to be 'pulls' or 'pull', " 2216 ) 2217 repo = parts[-3] 2218 owner = parts[-4] 2219 2220 return { 2221 "owner": owner, 2222 "repo": repo, 2223 "index": index, 2224 } 2225 except IndexError: 2226 raise ValueError("Malformed design review URL: {}".format(self.pull_request_url)) 2227 2228 @cached_property 2229 def owner_name(self) -> str: 2230 """ 2231 The owner of the repository this review is on. 2232 """ 2233 2234 return self._get_dr_properties()["owner"] 2235 2236 @cached_property 2237 def repository_name(self) -> str: 2238 """ 2239 The name of the repository this review is on. 2240 """ 2241 2242 return self._get_dr_properties()["repo"] 2243 2244 @cached_property 2245 def index(self) -> str: 2246 """ 2247 The index of the design review this review is on. 2248 """ 2249 2250 return self._get_dr_properties()["index"] 2251 2252 def delete(self): 2253 """ 2254 Delete this review. 2255 """ 2256 2257 self.allspice_client.requests_delete( 2258 self.API_OBJECT.format(**self._get_dr_properties(), id=self.id) 2259 ) 2260 self.deleted = True 2261 2262 def get_comments(self) -> List[DesignReviewReviewComment]: 2263 """ 2264 Get the comments on this review. 2265 """ 2266 2267 result = self.allspice_client.requests_get( 2268 self.GET_COMMENTS.format(**self._get_dr_properties(), id=self.id) 2269 ) 2270 2271 return [ 2272 DesignReviewReviewComment.parse_response(self.allspice_client, comment) 2273 for comment in result 2274 ] 2275 2276 2277class DesignReview(ApiObject): 2278 """ 2279 A Design Review. See 2280 https://hub.allspice.io/api/swagger#/repository/repoGetPullRequest. 2281 2282 Note: The base and head fields are not `Branch` objects - they are plain strings 2283 referring to the branch names. This is because DRs can exist for branches that have 2284 been deleted, which don't have an associated `Branch` object from the API. You can use 2285 the `Repository.get_branch` method to get a `Branch` object for a branch if you know 2286 it exists. 2287 """ 2288 2289 additions: Optional[int] 2290 allow_maintainer_edit: bool 2291 allow_maintainer_edits: Any 2292 assignee: User 2293 assignees: List["User"] 2294 base: str 2295 body: str 2296 changed_files: Optional[int] 2297 closed_at: Optional[str] 2298 comments: int 2299 created_at: str 2300 deletions: Optional[int] 2301 diff_url: str 2302 draft: bool 2303 due_date: Optional[str] 2304 head: str 2305 html_url: str 2306 id: int 2307 is_locked: bool 2308 labels: List[Any] 2309 merge_base: str 2310 merge_commit_sha: Optional[str] 2311 mergeable: bool 2312 merged: bool 2313 merged_at: Optional[str] 2314 merged_by: Any 2315 milestone: Any 2316 number: int 2317 patch_url: str 2318 pin_order: int 2319 repository: Optional["Repository"] 2320 requested_reviewers: Any 2321 requested_reviewers_teams: Any 2322 review_comments: int 2323 state: str 2324 title: str 2325 updated_at: str 2326 url: str 2327 user: User 2328 2329 API_OBJECT = "/repos/{owner}/{repo}/pulls/{index}" 2330 MERGE_DESIGN_REVIEW = "/repos/{owner}/{repo}/pulls/{index}/merge" 2331 GET_REVIEWS = "/repos/{owner}/{repo}/pulls/{index}/reviews" 2332 GET_REVIEW = "/repos/{owner}/{repo}/pulls/{index}/reviews/{review_id}" 2333 GET_COMMENTS = "/repos/{owner}/{repo}/issues/{index}/comments" 2334 2335 OPEN = "open" 2336 CLOSED = "closed" 2337 2338 class MergeType(Enum): 2339 MERGE = "merge" 2340 REBASE = "rebase" 2341 REBASE_MERGE = "rebase-merge" 2342 SQUASH = "squash" 2343 MANUALLY_MERGED = "manually-merged" 2344 2345 def __init__(self, allspice_client): 2346 super().__init__(allspice_client) 2347 2348 def __eq__(self, other): 2349 if not isinstance(other, DesignReview): 2350 return False 2351 return self.repository == other.repository and self.id == other.id 2352 2353 def __hash__(self): 2354 return hash(self.repository) ^ hash(self.id) 2355 2356 @classmethod 2357 def parse_response(cls, allspice_client, result) -> "DesignReview": 2358 api_object = super().parse_response(allspice_client, result) 2359 cls._add_read_property( 2360 "repository", 2361 Repository.parse_response(allspice_client, result["base"]["repo"]), 2362 api_object, 2363 ) 2364 2365 return api_object 2366 2367 @classmethod 2368 def request(cls, allspice_client, owner: str, repo: str, number: str): 2369 """See https://hub.allspice.io/api/swagger#/repository/repoGetPullRequest""" 2370 return cls._request(allspice_client, {"owner": owner, "repo": repo, "index": number}) 2371 2372 _fields_to_parsers: ClassVar[dict] = { 2373 "assignee": lambda allspice_client, u: User.parse_response(allspice_client, u), 2374 "assignees": lambda allspice_client, us: [ 2375 User.parse_response(allspice_client, u) for u in us 2376 ], 2377 "base": lambda _, b: b["ref"], 2378 "head": lambda _, h: h["ref"], 2379 "merged_by": lambda allspice_client, u: User.parse_response(allspice_client, u), 2380 "milestone": lambda allspice_client, m: Milestone.parse_response(allspice_client, m), 2381 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 2382 } 2383 2384 _patchable_fields: ClassVar[set[str]] = { 2385 "allow_maintainer_edits", 2386 "assignee", 2387 "assignees", 2388 "base", 2389 "body", 2390 "due_date", 2391 "milestone", 2392 "state", 2393 "title", 2394 } 2395 2396 _parsers_to_fields: ClassVar[dict] = { 2397 "assignee": lambda u: u.username, 2398 "assignees": lambda us: [u.username for u in us], 2399 "base": lambda b: b.name if isinstance(b, Branch) else b, 2400 "milestone": lambda m: m.id, 2401 } 2402 2403 def commit(self): 2404 data = self.get_dirty_fields() 2405 if "due_date" in data and data["due_date"] is None: 2406 data["unset_due_date"] = True 2407 2408 args = { 2409 "owner": self.repository.owner.username, 2410 "repo": self.repository.name, 2411 "index": self.number, 2412 } 2413 self._commit(args, data) 2414 2415 def merge(self, merge_type: MergeType): 2416 """ 2417 Merge the pull request. See 2418 https://hub.allspice.io/api/swagger#/repository/repoMergePullRequest 2419 2420 :param merge_type: The type of merge to perform. See the MergeType enum. 2421 """ 2422 2423 self.allspice_client.requests_post( 2424 self.MERGE_DESIGN_REVIEW.format( 2425 owner=self.repository.owner.username, 2426 repo=self.repository.name, 2427 index=self.number, 2428 ), 2429 data={"Do": merge_type.value}, 2430 ) 2431 2432 def get_comments(self) -> List[Comment]: 2433 """ 2434 Get the comments on this pull request, but not specifically on a review. 2435 2436 https://hub.allspice.io/api/swagger#/issue/issueGetComments 2437 2438 :return: A list of comments on this pull request. 2439 """ 2440 2441 results = self.allspice_client.requests_get( 2442 self.GET_COMMENTS.format( 2443 owner=self.repository.owner.username, 2444 repo=self.repository.name, 2445 index=self.number, 2446 ) 2447 ) 2448 return [Comment.parse_response(self.allspice_client, result) for result in results] 2449 2450 def create_comment(self, body: str): 2451 """ 2452 Create a comment on this pull request. This uses the same endpoint as the 2453 comments on issues, and will not be associated with any reviews. 2454 2455 https://hub.allspice.io/api/swagger#/issue/issueCreateComment 2456 2457 :param body: The body of the comment. 2458 :return: The comment that was created. 2459 """ 2460 2461 result = self.allspice_client.requests_post( 2462 self.GET_COMMENTS.format( 2463 owner=self.repository.owner.username, 2464 repo=self.repository.name, 2465 index=self.number, 2466 ), 2467 data={"body": body}, 2468 ) 2469 return Comment.parse_response(self.allspice_client, result) 2470 2471 def create_review( 2472 self, 2473 *, 2474 body: Optional[str] = None, 2475 event: Optional[DesignReviewReview.ReviewEvent] = None, 2476 comments: Optional[List[DesignReviewReview.ReviewComment]] = None, 2477 commit_id: Optional[str] = None, 2478 ) -> DesignReviewReview: 2479 """ 2480 Create a review on this design review. 2481 2482 https://hub.allspice.io/api/swagger#/repository/repoCreatePullRequestReview 2483 2484 Note: in most cases, you should not set the body or event when creating 2485 a review. The event is automatically set to "PENDING" when the review 2486 is created. You should then use `submit_review` to submit the review 2487 with the desired event and body. 2488 2489 :param body: The body of the review. This is the top-level comment on 2490 the review. If not provided, the review will be created with no body. 2491 :param event: The event of the review. This is the overall status of the 2492 review. See the ReviewEvent enum. If not provided, the API will 2493 default to "PENDING". 2494 :param comments: A list of comments on the review. Each comment should 2495 be a ReviewComment object. If not provided, only the base comment 2496 will be created. 2497 :param commit_id: The commit SHA to associate with the review. This is 2498 optional. 2499 """ 2500 2501 data: dict[str, Any] = {} 2502 2503 if body is not None: 2504 data["body"] = body 2505 if event is not None: 2506 data["event"] = event.value 2507 if commit_id is not None: 2508 data["commit_id"] = commit_id 2509 if comments is not None: 2510 data["comments"] = [asdict(comment) for comment in comments] 2511 2512 result = self.allspice_client.requests_post( 2513 self.GET_REVIEWS.format( 2514 owner=self.repository.owner.username, 2515 repo=self.repository.name, 2516 index=self.number, 2517 ), 2518 data=data, 2519 ) 2520 2521 return DesignReviewReview.parse_response(self.allspice_client, result) 2522 2523 def get_reviews(self) -> List[DesignReviewReview]: 2524 """ 2525 Get all reviews on this design review. 2526 2527 https://hub.allspice.io/api/swagger#/repository/repoListPullReviews 2528 """ 2529 2530 results = self.allspice_client.requests_get( 2531 self.GET_REVIEWS.format( 2532 owner=self.repository.owner.username, 2533 repo=self.repository.name, 2534 index=self.number, 2535 ) 2536 ) 2537 2538 return [ 2539 DesignReviewReview.parse_response(self.allspice_client, result) for result in results 2540 ] 2541 2542 def submit_review( 2543 self, 2544 review_id: int, 2545 event: DesignReviewReview.ReviewEvent, 2546 *, 2547 body: Optional[str] = None, 2548 ): 2549 """ 2550 Submit a review on this design review. 2551 2552 https://hub.allspice.io/api/swagger#/repository/repoSubmitPullReview 2553 2554 :param review_id: The ID of the review to submit. 2555 :param event: The event to submit the review with. See the ReviewEvent 2556 enum for the possible values. 2557 :param body: Optional body text for the review submission. 2558 """ 2559 2560 data = { 2561 "event": event.value, 2562 } 2563 if body is not None: 2564 data["body"] = body 2565 2566 result = self.allspice_client.requests_post( 2567 self.GET_REVIEW.format( 2568 owner=self.repository.owner.username, 2569 repo=self.repository.name, 2570 index=self.number, 2571 review_id=review_id, 2572 ), 2573 data=data, 2574 ) 2575 2576 return result 2577 2578 2579class Team(ApiObject): 2580 can_create_org_repo: bool 2581 description: str 2582 id: int 2583 includes_all_repositories: bool 2584 name: str 2585 organization: Optional["Organization"] 2586 permission: str 2587 units: List[str] 2588 units_map: Dict[str, str] 2589 2590 API_OBJECT = """/teams/{id}""" # <id> 2591 ADD_REPO = """/teams/%s/repos/%s/%s""" # <id, org, repo> 2592 TEAM_DELETE = """/teams/%s""" # <id> 2593 GET_MEMBERS = """/teams/%s/members""" # <id> 2594 GET_REPOS = """/teams/%s/repos""" # <id> 2595 2596 def __init__(self, allspice_client): 2597 super().__init__(allspice_client) 2598 2599 def __eq__(self, other): 2600 if not isinstance(other, Team): 2601 return False 2602 return self.organization == other.organization and self.id == other.id 2603 2604 def __hash__(self): 2605 return hash(self.organization) ^ hash(self.id) 2606 2607 _fields_to_parsers: ClassVar[dict] = { 2608 "organization": lambda allspice_client, o: Organization.parse_response(allspice_client, o) 2609 } 2610 2611 _patchable_fields: ClassVar[set[str]] = { 2612 "can_create_org_repo", 2613 "description", 2614 "includes_all_repositories", 2615 "name", 2616 "permission", 2617 "units", 2618 "units_map", 2619 } 2620 2621 @classmethod 2622 def request(cls, allspice_client, id: int): 2623 return cls._request(allspice_client, {"id": id}) 2624 2625 def commit(self): 2626 args = {"id": self.id} 2627 self._commit(args) 2628 2629 def add_user(self, user: User): 2630 """https://hub.allspice.io/api/swagger#/organization/orgAddTeamMember""" 2631 url = f"/teams/{self.id}/members/{user.login}" 2632 self.allspice_client.requests_put(url) 2633 2634 def add_repo(self, org: Organization, repo: Union[Repository, str]): 2635 if isinstance(repo, Repository): 2636 repo_name = repo.name 2637 else: 2638 repo_name = repo 2639 self.allspice_client.requests_put(Team.ADD_REPO % (self.id, org.username, repo_name)) 2640 2641 def get_members(self): 2642 """Get all users assigned to the team.""" 2643 results = self.allspice_client.requests_get_paginated( 2644 Team.GET_MEMBERS % self.id, 2645 ) 2646 return [User.parse_response(self.allspice_client, result) for result in results] 2647 2648 def get_repos(self): 2649 """Get all repos of this Team.""" 2650 results = self.allspice_client.requests_get(Team.GET_REPOS % self.id) 2651 return [Repository.parse_response(self.allspice_client, result) for result in results] 2652 2653 def delete(self): 2654 self.allspice_client.requests_delete(Team.TEAM_DELETE % self.id) 2655 self.deleted = True 2656 2657 def remove_team_member(self, user_name: str): 2658 url = f"/teams/{self.id}/members/{user_name}" 2659 self.allspice_client.requests_delete(url) 2660 2661 2662class Release(ApiObject): 2663 """ 2664 A release on a repo. 2665 """ 2666 2667 assets: List[Union[Any, Dict[str, Union[int, str]], "ReleaseAsset"]] 2668 author: User 2669 body: str 2670 created_at: str 2671 draft: bool 2672 html_url: str 2673 id: int 2674 name: str 2675 prerelease: bool 2676 published_at: str 2677 repo: Optional["Repository"] 2678 repository: Optional["Repository"] 2679 tag_name: str 2680 tarball_url: str 2681 target_commitish: str 2682 upload_url: str 2683 url: str 2684 zipball_url: str 2685 2686 API_OBJECT = "/repos/{owner}/{repo}/releases/{id}" 2687 RELEASE_CREATE_ASSET = "/repos/{owner}/{repo}/releases/{id}/assets" 2688 # Note that we don't strictly need the get_assets route, as the release 2689 # object already contains the assets. 2690 2691 def __init__(self, allspice_client): 2692 super().__init__(allspice_client) 2693 2694 def __eq__(self, other): 2695 if not isinstance(other, Release): 2696 return False 2697 return self.repo == other.repo and self.id == other.id 2698 2699 def __hash__(self): 2700 return hash(self.repo) ^ hash(self.id) 2701 2702 _fields_to_parsers: ClassVar[dict] = { 2703 "author": lambda allspice_client, author: User.parse_response(allspice_client, author), 2704 } 2705 _patchable_fields: ClassVar[set[str]] = { 2706 "body", 2707 "draft", 2708 "name", 2709 "prerelease", 2710 "tag_name", 2711 "target_commitish", 2712 } 2713 2714 @classmethod 2715 def parse_response(cls, allspice_client, result, repo) -> Release: 2716 release = super().parse_response(allspice_client, result) 2717 Release._add_read_property("repository", repo, release) 2718 # For legacy reasons 2719 Release._add_read_property("repo", repo, release) 2720 setattr( 2721 release, 2722 "_assets", 2723 [ 2724 ReleaseAsset.parse_response(allspice_client, asset, release) 2725 for asset in result["assets"] 2726 ], 2727 ) 2728 return release 2729 2730 @classmethod 2731 def request( 2732 cls, 2733 allspice_client, 2734 owner: str, 2735 repo: str, 2736 id: Optional[int] = None, 2737 ) -> Release: 2738 args = {"owner": owner, "repo": repo, "id": id} 2739 release_response = cls._get_gitea_api_object(allspice_client, args) 2740 repository = Repository.request(allspice_client, owner, repo) 2741 release = cls.parse_response(allspice_client, release_response, repository) 2742 return release 2743 2744 def commit(self): 2745 if self.repo is None: 2746 raise ValueError("Cannot commit a release without a repository.") 2747 2748 args = {"owner": self.repo.owner.username, "repo": self.repo.name, "id": self.id} 2749 self._commit(args) 2750 2751 def create_asset(self, file: IO, name: Optional[str] = None) -> ReleaseAsset: 2752 """ 2753 Create an asset for this release. 2754 2755 https://hub.allspice.io/api/swagger#/repository/repoCreateReleaseAsset 2756 2757 :param file: The file to upload. This should be a file-like object. 2758 :param name: The name of the file. 2759 :return: The created asset. 2760 """ 2761 2762 if self.repo is None: 2763 raise ValueError("Cannot commit a release without a repository.") 2764 2765 args: dict[str, Any] = {"files": {"attachment": file}} 2766 if name is not None: 2767 args["params"] = {"name": name} 2768 2769 result = self.allspice_client.requests_post( 2770 self.RELEASE_CREATE_ASSET.format( 2771 owner=self.repo.owner.username, 2772 repo=self.repo.name, 2773 id=self.id, 2774 ), 2775 **args, 2776 ) 2777 return ReleaseAsset.parse_response(self.allspice_client, result, self) 2778 2779 def delete(self): 2780 if self.repo is None: 2781 raise ValueError("Cannot commit a release without a repository.") 2782 2783 args = {"owner": self.repo.owner.username, "repo": self.repo.name, "id": self.id} 2784 self.allspice_client.requests_delete(self.API_OBJECT.format(**args)) 2785 self.deleted = True 2786 2787 2788class ReleaseAsset(ApiObject): 2789 browser_download_url: str 2790 created_at: str 2791 download_count: int 2792 id: int 2793 name: str 2794 release: Optional["Release"] 2795 size: int 2796 uuid: str 2797 2798 API_OBJECT = "/repos/{owner}/{repo}/releases/{release_id}/assets/{id}" 2799 2800 def __init__(self, allspice_client): 2801 super().__init__(allspice_client) 2802 2803 def __eq__(self, other): 2804 if not isinstance(other, ReleaseAsset): 2805 return False 2806 return self.release == other.release and self.id == other.id 2807 2808 def __hash__(self): 2809 return hash(self.release) ^ hash(self.id) 2810 2811 _fields_to_parsers: ClassVar[dict] = {} 2812 _patchable_fields: ClassVar[set[str]] = { 2813 "name", 2814 } 2815 2816 @classmethod 2817 def parse_response(cls, allspice_client, result, release) -> ReleaseAsset: 2818 asset = super().parse_response(allspice_client, result) 2819 ReleaseAsset._add_read_property("release", release, asset) 2820 return asset 2821 2822 @classmethod 2823 def request( 2824 cls, 2825 allspice_client, 2826 owner: str, 2827 repo: str, 2828 release_id: int, 2829 id: int, 2830 ) -> ReleaseAsset: 2831 args = {"owner": owner, "repo": repo, "release_id": release_id, "id": id} 2832 asset_response = cls._get_gitea_api_object(allspice_client, args) 2833 release = Release.request(allspice_client, owner, repo, release_id) 2834 asset = cls.parse_response(allspice_client, asset_response, release) 2835 return asset 2836 2837 def commit(self): 2838 if self.release is None or self.release.repo is None: 2839 raise ValueError("Cannot commit a release asset without a release or a repository.") 2840 2841 args = { 2842 "owner": self.release.repo.owner.username, 2843 "repo": self.release.repo.name, 2844 "release_id": self.release.id, 2845 "id": self.id, 2846 } 2847 self._commit(args) 2848 2849 def download(self) -> bytes: 2850 """ 2851 Download the raw, binary data of this asset. 2852 2853 Note 1: if the file you are requesting is a text file, you might want to 2854 use .decode() on the result to get a string. For example: 2855 2856 asset.download().decode("utf-8") 2857 2858 Note 2: this method will store the entire file in memory. If you are 2859 downloading a large file, you might want to use download_to_file instead. 2860 """ 2861 2862 return self.allspice_client.requests.get( 2863 self.browser_download_url, 2864 headers=self.allspice_client.headers, 2865 ).content 2866 2867 def download_to_file(self, io: IO): 2868 """ 2869 Download the raw, binary data of this asset to a file-like object. 2870 2871 Example: 2872 2873 with open("my_file.zip", "wb") as f: 2874 asset.download_to_file(f) 2875 2876 :param io: The file-like object to write the data to. 2877 """ 2878 2879 response = self.allspice_client.requests.get( 2880 self.browser_download_url, 2881 headers=self.allspice_client.headers, 2882 stream=True, 2883 ) 2884 # 4kb chunks 2885 for chunk in response.iter_content(chunk_size=4096): 2886 if chunk: 2887 io.write(chunk) 2888 2889 def delete(self): 2890 if self.release is None or self.release.repo is None: 2891 raise ValueError("Cannot commit a release asset without a release or a repository.") 2892 2893 args = { 2894 "owner": self.release.repo.owner.username, 2895 "repo": self.release.repo.name, 2896 "release_id": self.release.id, 2897 "id": self.id, 2898 } 2899 self.allspice_client.requests_delete(self.API_OBJECT.format(**args)) 2900 self.deleted = True 2901 2902 2903class Content(ReadonlyApiObject): 2904 content: Any 2905 download_url: str 2906 encoding: Any 2907 git_url: str 2908 html_url: str 2909 last_commit_sha: str 2910 name: str 2911 path: str 2912 sha: str 2913 size: int 2914 submodule_git_url: Any 2915 target: Any 2916 type: str 2917 url: str 2918 2919 FILE = "file" 2920 2921 def __init__(self, allspice_client): 2922 super().__init__(allspice_client) 2923 2924 def __eq__(self, other): 2925 if not isinstance(other, Content): 2926 return False 2927 2928 return self.sha == other.sha and self.name == other.name 2929 2930 def __hash__(self): 2931 return hash(self.sha) ^ hash(self.name) 2932 2933 2934Ref = Union[Branch, Commit, str] 2935 2936 2937class Util: 2938 @staticmethod 2939 def convert_time(time: str) -> datetime: 2940 """Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" 2941 try: 2942 return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") 2943 except ValueError: 2944 return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S") 2945 2946 @staticmethod 2947 def format_time(time: datetime) -> str: 2948 """ 2949 Format a datetime object to Gitea's time format. 2950 2951 :param time: The time to format 2952 :return: Formatted time 2953 """ 2954 2955 return time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + "Z" 2956 2957 @staticmethod 2958 def data_params_for_ref(ref: Optional[Ref]) -> Dict: 2959 """ 2960 Given a "ref", returns a dict with the ref parameter for the API call. 2961 2962 If the ref is None, returns an empty dict. You can pass this to the API 2963 directly. 2964 """ 2965 2966 if isinstance(ref, Branch): 2967 return {"ref": ref.name} 2968 elif isinstance(ref, Commit): 2969 return {"ref": ref.sha} 2970 elif ref: 2971 return {"ref": ref} 2972 else: 2973 return {}
36class Organization(ApiObject): 37 """see https://hub.allspice.io/api/swagger#/organization/orgGetAll""" 38 39 active: Optional[bool] 40 avatar_url: str 41 created: Optional[str] 42 description: str 43 email: str 44 followers_count: Optional[int] 45 following_count: Optional[int] 46 full_name: str 47 html_url: Optional[str] 48 id: int 49 is_admin: Optional[bool] 50 language: Optional[str] 51 last_login: Optional[str] 52 location: str 53 login: Optional[str] 54 login_name: Optional[str] 55 name: Optional[str] 56 prohibit_login: Optional[bool] 57 repo_admin_change_team_access: Optional[bool] 58 restricted: Optional[bool] 59 source_id: Optional[int] 60 starred_repos_count: Optional[int] 61 username: str 62 visibility: str 63 website: str 64 65 API_OBJECT = """/orgs/{name}""" # <org> 66 ORG_REPOS_REQUEST = """/orgs/%s/repos""" # <org> 67 ORG_TEAMS_REQUEST = """/orgs/%s/teams""" # <org> 68 ORG_TEAMS_CREATE = """/orgs/%s/teams""" # <org> 69 ORG_GET_MEMBERS = """/orgs/%s/members""" # <org> 70 ORG_IS_MEMBER = """/orgs/%s/members/%s""" # <org>, <username> 71 ORG_HEATMAP = """/users/%s/heatmap""" # <username> 72 73 def __init__(self, allspice_client): 74 super().__init__(allspice_client) 75 76 def __eq__(self, other): 77 if not isinstance(other, Organization): 78 return False 79 return self.allspice_client == other.allspice_client and self.name == other.name 80 81 def __hash__(self): 82 return hash(self.allspice_client) ^ hash(self.name) 83 84 @classmethod 85 def request(cls, allspice_client, name: str) -> Self: 86 return cls._request(allspice_client, {"name": name}) 87 88 @classmethod 89 def parse_response(cls, allspice_client, result) -> "Organization": 90 api_object = super().parse_response(allspice_client, result) 91 # add "name" field to make this behave similar to users for gitea < 1.18 92 # also necessary for repository-owner when org is repo owner 93 if not hasattr(api_object, "name"): 94 Organization._add_read_property("name", result["username"], api_object) 95 return api_object 96 97 _patchable_fields: ClassVar[set[str]] = { 98 "description", 99 "full_name", 100 "location", 101 "visibility", 102 "website", 103 } 104 105 def commit(self): 106 args = {"name": self.name} 107 self._commit(args) 108 109 def create_repo( 110 self, 111 repoName: str, 112 description: str = "", 113 private: bool = False, 114 autoInit=True, 115 gitignores: Optional[str] = None, 116 license: Optional[str] = None, 117 readme: str = "Default", 118 issue_labels: Optional[str] = None, 119 default_branch="master", 120 ): 121 """Create an organization Repository 122 123 Throws: 124 AlreadyExistsException: If the Repository exists already. 125 Exception: If something else went wrong. 126 """ 127 result = self.allspice_client.requests_post( 128 f"/orgs/{self.name}/repos", 129 data={ 130 "name": repoName, 131 "description": description, 132 "private": private, 133 "auto_init": autoInit, 134 "gitignores": gitignores, 135 "license": license, 136 "issue_labels": issue_labels, 137 "readme": readme, 138 "default_branch": default_branch, 139 }, 140 ) 141 if "id" in result: 142 self.allspice_client.logger.info("Successfully created Repository %s " % result["name"]) 143 else: 144 self.allspice_client.logger.error(result["message"]) 145 raise Exception("Repository not created... (gitea: %s)" % result["message"]) 146 return Repository.parse_response(self.allspice_client, result) 147 148 def get_repositories(self) -> List["Repository"]: 149 results = self.allspice_client.requests_get_paginated( 150 Organization.ORG_REPOS_REQUEST % self.username 151 ) 152 return [Repository.parse_response(self.allspice_client, result) for result in results] 153 154 def get_repository(self, name) -> "Repository": 155 repos = self.get_repositories() 156 for repo in repos: 157 if repo.name == name: 158 return repo 159 raise NotFoundException("Repository %s not existent in organization." % name) 160 161 def get_teams(self) -> List["Team"]: 162 results = self.allspice_client.requests_get(Organization.ORG_TEAMS_REQUEST % self.username) 163 teams = [Team.parse_response(self.allspice_client, result) for result in results] 164 # organisation seems to be missing using this request, so we add org manually 165 for t in teams: 166 setattr(t, "_organization", self) 167 return teams 168 169 def get_team(self, name) -> "Team": 170 teams = self.get_teams() 171 for team in teams: 172 if team.name == name: 173 return team 174 raise NotFoundException("Team not existent in organization.") 175 176 def create_team( 177 self, 178 name: str, 179 description: str = "", 180 permission: str = "read", 181 can_create_org_repo: bool = False, 182 includes_all_repositories: bool = False, 183 units=( 184 "repo.code", 185 "repo.issues", 186 "repo.ext_issues", 187 "repo.wiki", 188 "repo.pulls", 189 "repo.releases", 190 "repo.ext_wiki", 191 ), 192 units_map={}, 193 ) -> "Team": 194 """Alias for AllSpice#create_team""" 195 # TODO: Move AllSpice#create_team to Organization#create_team and 196 # deprecate AllSpice#create_team. 197 return self.allspice_client.create_team( 198 org=self, 199 name=name, 200 description=description, 201 permission=permission, 202 can_create_org_repo=can_create_org_repo, 203 includes_all_repositories=includes_all_repositories, 204 units=units, 205 units_map=units_map, 206 ) 207 208 def get_members(self) -> List["User"]: 209 results = self.allspice_client.requests_get(Organization.ORG_GET_MEMBERS % self.username) 210 return [User.parse_response(self.allspice_client, result) for result in results] 211 212 def is_member(self, username) -> bool: 213 if isinstance(username, User): 214 username = username.username 215 try: 216 # returns 204 if its ok, 404 if its not 217 self.allspice_client.requests_get( 218 Organization.ORG_IS_MEMBER % (self.username, username) 219 ) 220 return True 221 except Exception: 222 return False 223 224 def remove_member(self, user: "User"): 225 path = f"/orgs/{self.username}/members/{user.username}" 226 self.allspice_client.requests_delete(path) 227 228 def delete(self): 229 """Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" 230 for repo in self.get_repositories(): 231 repo.delete() 232 self.allspice_client.requests_delete(Organization.API_OBJECT.format(name=self.username)) 233 self.deleted = True 234 235 def get_heatmap(self) -> List[Tuple[datetime, int]]: 236 results = self.allspice_client.requests_get(User.USER_HEATMAP % self.username) 237 results = [ 238 (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) 239 for result in results 240 ] 241 return results
see allspice.allspice.io/api/swagger#/organization/orgGetAll">https://huballspice.allspice.io/api/swagger#/organization/orgGetAll
88 @classmethod 89 def parse_response(cls, allspice_client, result) -> "Organization": 90 api_object = super().parse_response(allspice_client, result) 91 # add "name" field to make this behave similar to users for gitea < 1.18 92 # also necessary for repository-owner when org is repo owner 93 if not hasattr(api_object, "name"): 94 Organization._add_read_property("name", result["username"], api_object) 95 return api_object
def
create_repo( self, repoName: str, description: str = '', private: bool = False, autoInit=True, gitignores: Optional[str] = None, license: Optional[str] = None, readme: str = 'Default', issue_labels: Optional[str] = None, default_branch='master'):
109 def create_repo( 110 self, 111 repoName: str, 112 description: str = "", 113 private: bool = False, 114 autoInit=True, 115 gitignores: Optional[str] = None, 116 license: Optional[str] = None, 117 readme: str = "Default", 118 issue_labels: Optional[str] = None, 119 default_branch="master", 120 ): 121 """Create an organization Repository 122 123 Throws: 124 AlreadyExistsException: If the Repository exists already. 125 Exception: If something else went wrong. 126 """ 127 result = self.allspice_client.requests_post( 128 f"/orgs/{self.name}/repos", 129 data={ 130 "name": repoName, 131 "description": description, 132 "private": private, 133 "auto_init": autoInit, 134 "gitignores": gitignores, 135 "license": license, 136 "issue_labels": issue_labels, 137 "readme": readme, 138 "default_branch": default_branch, 139 }, 140 ) 141 if "id" in result: 142 self.allspice_client.logger.info("Successfully created Repository %s " % result["name"]) 143 else: 144 self.allspice_client.logger.error(result["message"]) 145 raise Exception("Repository not created... (gitea: %s)" % result["message"]) 146 return Repository.parse_response(self.allspice_client, result)
Create an organization Repository
Throws: AlreadyExistsException: If the Repository exists already. Exception: If something else went wrong.
161 def get_teams(self) -> List["Team"]: 162 results = self.allspice_client.requests_get(Organization.ORG_TEAMS_REQUEST % self.username) 163 teams = [Team.parse_response(self.allspice_client, result) for result in results] 164 # organisation seems to be missing using this request, so we add org manually 165 for t in teams: 166 setattr(t, "_organization", self) 167 return teams
def
create_team( self, name: str, description: str = '', permission: str = 'read', can_create_org_repo: bool = False, includes_all_repositories: bool = False, units=('repo.code', 'repo.issues', 'repo.ext_issues', 'repo.wiki', 'repo.pulls', 'repo.releases', 'repo.ext_wiki'), units_map={}) -> Team:
176 def create_team( 177 self, 178 name: str, 179 description: str = "", 180 permission: str = "read", 181 can_create_org_repo: bool = False, 182 includes_all_repositories: bool = False, 183 units=( 184 "repo.code", 185 "repo.issues", 186 "repo.ext_issues", 187 "repo.wiki", 188 "repo.pulls", 189 "repo.releases", 190 "repo.ext_wiki", 191 ), 192 units_map={}, 193 ) -> "Team": 194 """Alias for AllSpice#create_team""" 195 # TODO: Move AllSpice#create_team to Organization#create_team and 196 # deprecate AllSpice#create_team. 197 return self.allspice_client.create_team( 198 org=self, 199 name=name, 200 description=description, 201 permission=permission, 202 can_create_org_repo=can_create_org_repo, 203 includes_all_repositories=includes_all_repositories, 204 units=units, 205 units_map=units_map, 206 )
Alias for AllSpice#create_team
def
is_member(self, username) -> bool:
212 def is_member(self, username) -> bool: 213 if isinstance(username, User): 214 username = username.username 215 try: 216 # returns 204 if its ok, 404 if its not 217 self.allspice_client.requests_get( 218 Organization.ORG_IS_MEMBER % (self.username, username) 219 ) 220 return True 221 except Exception: 222 return False
def
delete(self):
228 def delete(self): 229 """Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User""" 230 for repo in self.get_repositories(): 231 repo.delete() 232 self.allspice_client.requests_delete(Organization.API_OBJECT.format(name=self.username)) 233 self.deleted = True
Delete this Organization. Invalidates this Objects data. Also deletes all Repositories owned by the User
244class User(ApiObject): 245 active: bool 246 admin: Any 247 allow_create_organization: Any 248 allow_git_hook: Any 249 allow_import_local: Any 250 avatar_url: str 251 created: str 252 description: str 253 email: str 254 emails: List[Any] 255 followers_count: int 256 following_count: int 257 full_name: str 258 html_url: str 259 id: int 260 is_admin: bool 261 language: str 262 last_login: str 263 location: str 264 login: str 265 login_name: str 266 max_repo_creation: Any 267 must_change_password: Any 268 password: Any 269 prohibit_login: bool 270 restricted: bool 271 source_id: int 272 starred_repos_count: int 273 username: str 274 visibility: str 275 website: str 276 277 API_OBJECT = """/users/{name}""" # <org> 278 USER_MAIL = """/user/emails?sudo=%s""" # <name> 279 USER_PATCH = """/admin/users/%s""" # <username> 280 ADMIN_DELETE_USER = """/admin/users/%s""" # <username> 281 ADMIN_EDIT_USER = """/admin/users/{username}""" # <username> 282 USER_HEATMAP = """/users/%s/heatmap""" # <username> 283 284 def __init__(self, allspice_client): 285 super().__init__(allspice_client) 286 self._emails = [] 287 288 def __eq__(self, other): 289 if not isinstance(other, User): 290 return False 291 return self.allspice_client == other.allspice_client and self.id == other.id 292 293 def __hash__(self): 294 return hash(self.allspice_client) ^ hash(self.id) 295 296 @property 297 def emails(self): 298 self.__request_emails() 299 return self._emails 300 301 @classmethod 302 def request(cls, allspice_client, name: str) -> "User": 303 api_object = cls._request(allspice_client, {"name": name}) 304 return api_object 305 306 _patchable_fields: ClassVar[set[str]] = { 307 "active", 308 "admin", 309 "allow_create_organization", 310 "allow_git_hook", 311 "allow_import_local", 312 "email", 313 "full_name", 314 "location", 315 "login_name", 316 "max_repo_creation", 317 "must_change_password", 318 "password", 319 "prohibit_login", 320 "website", 321 } 322 323 def commit(self, login_name: str, source_id: int = 0): 324 """ 325 Unfortunately it is necessary to require the login name 326 as well as the login source (that is not supplied when getting a user) for 327 changing a user. 328 Usually source_id is 0 and the login_name is equal to the username. 329 """ 330 values = self.get_dirty_fields() 331 values.update( 332 # api-doc says that the "source_id" is necessary; works without though 333 {"login_name": login_name, "source_id": source_id} 334 ) 335 args = {"username": self.username} 336 self.allspice_client.requests_patch(User.ADMIN_EDIT_USER.format(**args), data=values) 337 self._dirty_fields = {} 338 339 def create_repo( 340 self, 341 repoName: str, 342 description: str = "", 343 private: bool = False, 344 autoInit=True, 345 gitignores: Optional[str] = None, 346 license: Optional[str] = None, 347 readme: str = "Default", 348 issue_labels: Optional[str] = None, 349 default_branch="master", 350 ): 351 """Create a user Repository 352 353 Throws: 354 AlreadyExistsException: If the Repository exists already. 355 Exception: If something else went wrong. 356 """ 357 result = self.allspice_client.requests_post( 358 "/user/repos", 359 data={ 360 "name": repoName, 361 "description": description, 362 "private": private, 363 "auto_init": autoInit, 364 "gitignores": gitignores, 365 "license": license, 366 "issue_labels": issue_labels, 367 "readme": readme, 368 "default_branch": default_branch, 369 }, 370 ) 371 if "id" in result: 372 self.allspice_client.logger.info("Successfully created Repository %s " % result["name"]) 373 else: 374 self.allspice_client.logger.error(result["message"]) 375 raise Exception("Repository not created... (gitea: %s)" % result["message"]) 376 return Repository.parse_response(self.allspice_client, result) 377 378 def get_repositories(self) -> List["Repository"]: 379 """Get all Repositories owned by this User.""" 380 url = f"/users/{self.username}/repos" 381 results = self.allspice_client.requests_get_paginated(url) 382 return [Repository.parse_response(self.allspice_client, result) for result in results] 383 384 def get_orgs(self) -> List[Organization]: 385 """Get all Organizations this user is a member of.""" 386 url = f"/users/{self.username}/orgs" 387 results = self.allspice_client.requests_get_paginated(url) 388 return [Organization.parse_response(self.allspice_client, result) for result in results] 389 390 def get_teams(self) -> List["Team"]: 391 url = "/user/teams" 392 results = self.allspice_client.requests_get_paginated(url, sudo=self) 393 return [Team.parse_response(self.allspice_client, result) for result in results] 394 395 def get_accessible_repos(self) -> List["Repository"]: 396 """Get all Repositories accessible by the logged in User.""" 397 results = self.allspice_client.requests_get("/user/repos", sudo=self) 398 return [Repository.parse_response(self.allspice_client, result) for result in results] 399 400 def __request_emails(self): 401 result = self.allspice_client.requests_get(User.USER_MAIL % self.login) 402 # report if the adress changed by this 403 for mail in result: 404 self._emails.append(mail["email"]) 405 if mail["primary"]: 406 self._email = mail["email"] 407 408 def delete(self): 409 """Deletes this User. Also deletes all Repositories he owns.""" 410 self.allspice_client.requests_delete(User.ADMIN_DELETE_USER % self.username) 411 self.deleted = True 412 413 def get_heatmap(self) -> List[Tuple[datetime, int]]: 414 results = self.allspice_client.requests_get(User.USER_HEATMAP % self.username) 415 results = [ 416 (datetime.fromtimestamp(result["timestamp"]), result["contributions"]) 417 for result in results 418 ] 419 return results
def
commit(self, login_name: str, source_id: int = 0):
323 def commit(self, login_name: str, source_id: int = 0): 324 """ 325 Unfortunately it is necessary to require the login name 326 as well as the login source (that is not supplied when getting a user) for 327 changing a user. 328 Usually source_id is 0 and the login_name is equal to the username. 329 """ 330 values = self.get_dirty_fields() 331 values.update( 332 # api-doc says that the "source_id" is necessary; works without though 333 {"login_name": login_name, "source_id": source_id} 334 ) 335 args = {"username": self.username} 336 self.allspice_client.requests_patch(User.ADMIN_EDIT_USER.format(**args), data=values) 337 self._dirty_fields = {}
Unfortunately it is necessary to require the login name as well as the login source (that is not supplied when getting a user) for changing a user. Usually source_id is 0 and the login_name is equal to the username.
def
create_repo( self, repoName: str, description: str = '', private: bool = False, autoInit=True, gitignores: Optional[str] = None, license: Optional[str] = None, readme: str = 'Default', issue_labels: Optional[str] = None, default_branch='master'):
339 def create_repo( 340 self, 341 repoName: str, 342 description: str = "", 343 private: bool = False, 344 autoInit=True, 345 gitignores: Optional[str] = None, 346 license: Optional[str] = None, 347 readme: str = "Default", 348 issue_labels: Optional[str] = None, 349 default_branch="master", 350 ): 351 """Create a user Repository 352 353 Throws: 354 AlreadyExistsException: If the Repository exists already. 355 Exception: If something else went wrong. 356 """ 357 result = self.allspice_client.requests_post( 358 "/user/repos", 359 data={ 360 "name": repoName, 361 "description": description, 362 "private": private, 363 "auto_init": autoInit, 364 "gitignores": gitignores, 365 "license": license, 366 "issue_labels": issue_labels, 367 "readme": readme, 368 "default_branch": default_branch, 369 }, 370 ) 371 if "id" in result: 372 self.allspice_client.logger.info("Successfully created Repository %s " % result["name"]) 373 else: 374 self.allspice_client.logger.error(result["message"]) 375 raise Exception("Repository not created... (gitea: %s)" % result["message"]) 376 return Repository.parse_response(self.allspice_client, result)
Create a user Repository
Throws: AlreadyExistsException: If the Repository exists already. Exception: If something else went wrong.
378 def get_repositories(self) -> List["Repository"]: 379 """Get all Repositories owned by this User.""" 380 url = f"/users/{self.username}/repos" 381 results = self.allspice_client.requests_get_paginated(url) 382 return [Repository.parse_response(self.allspice_client, result) for result in results]
Get all Repositories owned by this User.
384 def get_orgs(self) -> List[Organization]: 385 """Get all Organizations this user is a member of.""" 386 url = f"/users/{self.username}/orgs" 387 results = self.allspice_client.requests_get_paginated(url) 388 return [Organization.parse_response(self.allspice_client, result) for result in results]
Get all Organizations this user is a member of.
395 def get_accessible_repos(self) -> List["Repository"]: 396 """Get all Repositories accessible by the logged in User.""" 397 results = self.allspice_client.requests_get("/user/repos", sudo=self) 398 return [Repository.parse_response(self.allspice_client, result) for result in results]
Get all Repositories accessible by the logged in User.
def
delete(self):
408 def delete(self): 409 """Deletes this User. Also deletes all Repositories he owns.""" 410 self.allspice_client.requests_delete(User.ADMIN_DELETE_USER % self.username) 411 self.deleted = True
Deletes this User. Also deletes all Repositories he owns.
422class Branch(ReadonlyApiObject): 423 commit: Dict[str, Optional[Union[str, Dict[str, str], Dict[str, Optional[Union[bool, str]]]]]] 424 effective_branch_protection_name: str 425 enable_status_check: bool 426 name: str 427 protected: bool 428 required_approvals: int 429 status_check_contexts: List[Any] 430 user_can_merge: bool 431 user_can_push: bool 432 433 API_OBJECT = """/repos/{owner}/{repo}/branches/{branch}""" 434 435 def __init__(self, allspice_client): 436 super().__init__(allspice_client) 437 438 def __eq__(self, other): 439 if not isinstance(other, Branch): 440 return False 441 return self.commit == other.commit and self.name == other.name 442 443 def __hash__(self): 444 return hash(self.commit["id"]) ^ hash(self.name) 445 446 _fields_to_parsers: ClassVar[dict] = { 447 # This is not a commit object 448 # "commit": lambda allspice_client, c: Commit.parse_response(allspice_client, c) 449 } 450 451 @classmethod 452 def request(cls, allspice_client, owner: str, repo: str, branch: str): 453 return cls._request(allspice_client, {"owner": owner, "repo": repo, "branch": branch})
Inherited Members
456class GitEntry(ReadonlyApiObject): 457 """ 458 An object representing a file or directory in the Git tree. 459 """ 460 461 mode: str 462 path: str 463 sha: str 464 size: int 465 type: str 466 url: str 467 468 def __init__(self, allspice_client): 469 super().__init__(allspice_client) 470 471 def __eq__(self, other) -> bool: 472 if not isinstance(other, GitEntry): 473 return False 474 return self.sha == other.sha 475 476 def __hash__(self) -> int: 477 return hash(self.sha)
An object representing a file or directory in the Git tree.
Inherited Members
480class Repository(ApiObject): 481 allow_fast_forward_only_merge: bool 482 allow_manual_merge: Any 483 allow_merge_commits: bool 484 allow_rebase: bool 485 allow_rebase_explicit: bool 486 allow_rebase_update: bool 487 allow_squash_merge: bool 488 archived: bool 489 archived_at: str 490 autodetect_manual_merge: Any 491 avatar_url: str 492 clone_url: str 493 created_at: str 494 default_allow_maintainer_edit: bool 495 default_branch: str 496 default_delete_branch_after_merge: bool 497 default_merge_style: str 498 description: str 499 empty: bool 500 enable_prune: Any 501 external_tracker: Any 502 external_wiki: Any 503 fork: bool 504 forks_count: int 505 full_name: str 506 has_actions: bool 507 has_issues: bool 508 has_packages: bool 509 has_projects: bool 510 has_pull_requests: bool 511 has_releases: bool 512 has_wiki: bool 513 html_url: str 514 id: int 515 ignore_whitespace_conflicts: bool 516 internal: bool 517 internal_tracker: Dict[str, bool] 518 language: str 519 languages_url: str 520 licenses: Any 521 link: str 522 mirror: bool 523 mirror_interval: str 524 mirror_updated: str 525 name: str 526 object_format_name: str 527 open_issues_count: int 528 open_pr_counter: int 529 original_url: str 530 owner: Union["User", "Organization"] 531 parent: Any 532 permissions: Dict[str, bool] 533 private: bool 534 projects_mode: str 535 release_counter: int 536 repo_transfer: Any 537 size: int 538 ssh_url: str 539 stars_count: int 540 template: bool 541 topics: List[Union[Any, str]] 542 updated_at: datetime 543 url: str 544 watchers_count: int 545 website: str 546 547 API_OBJECT = """/repos/{owner}/{name}""" # <owner>, <reponame> 548 REPO_IS_COLLABORATOR = """/repos/%s/%s/collaborators/%s""" # <owner>, <reponame>, <username> 549 REPO_SEARCH = """/repos/search/""" 550 REPO_BRANCHES = """/repos/%s/%s/branches""" # <owner>, <reponame> 551 REPO_BRANCH = """/repos/{owner}/{repo}/branches/{branch}""" 552 REPO_ISSUES = """/repos/{owner}/{repo}/issues""" # <owner, reponame> 553 REPO_DESIGN_REVIEWS = """/repos/{owner}/{repo}/pulls""" 554 REPO_DELETE = """/repos/%s/%s""" # <owner>, <reponame> 555 REPO_TIMES = """/repos/%s/%s/times""" # <owner>, <reponame> 556 REPO_USER_TIME = """/repos/%s/%s/times/%s""" # <owner>, <reponame>, <username> 557 REPO_COMMITS = "/repos/%s/%s/commits" # <owner>, <reponame> 558 REPO_TRANSFER = "/repos/{owner}/{repo}/transfer" 559 REPO_MILESTONES = """/repos/{owner}/{repo}/milestones""" 560 REPO_GET_ARCHIVE = "/repos/{owner}/{repo}/archive/{ref}.{format}" 561 REPO_GET_ALLSPICE_JSON = "/repos/{owner}/{repo}/allspice_generated/json/{content}" 562 REPO_GET_ALLSPICE_SVG = "/repos/{owner}/{repo}/allspice_generated/svg/{content}" 563 REPO_GET_ALLSPICE_PROJECT = "/repos/{owner}/{repo}/allspice_generated/project/{content}" 564 REPO_GET_TOPICS = "/repos/{owner}/{repo}/topics" 565 REPO_ADD_TOPIC = "/repos/{owner}/{repo}/topics/{topic}" 566 REPO_GET_RELEASES = "/repos/{owner}/{repo}/releases" 567 REPO_GET_LATEST_RELEASE = "/repos/{owner}/{repo}/releases/latest" 568 REPO_GET_RELEASE_BY_TAG = "/repos/{owner}/{repo}/releases/tags/{tag}" 569 REPO_GET_COMMIT_STATUS = "/repos/{owner}/{repo}/statuses/{sha}" 570 REPO_GET_MEDIA = "/repos/{owner}/{repo}/media/{path}" 571 REPO_GET_TREE = "/repos/{owner}/{repo}/git/trees/{ref}" 572 573 class ArchiveFormat(Enum): 574 """ 575 Archive formats for Repository.get_archive 576 """ 577 578 TAR = "tar.gz" 579 ZIP = "zip" 580 581 class CommitStatusSort(Enum): 582 """ 583 Sort order for Repository.get_commit_status 584 """ 585 586 OLDEST = "oldest" 587 RECENT_UPDATE = "recentupdate" 588 LEAST_UPDATE = "leastupdate" 589 LEAST_INDEX = "leastindex" 590 HIGHEST_INDEX = "highestindex" 591 592 def __init__(self, allspice_client): 593 super().__init__(allspice_client) 594 595 def __eq__(self, other): 596 if not isinstance(other, Repository): 597 return False 598 return self.owner == other.owner and self.name == other.name 599 600 def __hash__(self): 601 return hash(self.owner) ^ hash(self.name) 602 603 _fields_to_parsers: ClassVar[dict] = { 604 # dont know how to tell apart user and org as owner except form email being empty. 605 "owner": lambda allspice_client, r: ( 606 Organization.parse_response(allspice_client, r) 607 if r["email"] == "" 608 else User.parse_response(allspice_client, r) 609 ), 610 "updated_at": lambda _, t: Util.convert_time(t), 611 } 612 613 @classmethod 614 def request( 615 cls, 616 allspice_client, 617 owner: str, 618 name: str, 619 ) -> Repository: 620 return cls._request(allspice_client, {"owner": owner, "name": name}) 621 622 @classmethod 623 def search( 624 cls, 625 allspice_client, 626 query: Optional[str] = None, 627 topic: bool = False, 628 include_description: bool = False, 629 user: Optional[User] = None, 630 owner_to_prioritize: Union[User, Organization, None] = None, 631 ) -> list[Repository]: 632 """ 633 Search for repositories. 634 635 See https://hub.allspice.io/api/swagger#/repository/repoSearch 636 637 :param query: The query string to search for 638 :param topic: If true, the query string will only be matched against the 639 repository's topic. 640 :param include_description: If true, the query string will be matched 641 against the repository's description as well. 642 :param user: If specified, only repositories that this user owns or 643 contributes to will be searched. 644 :param owner_to_prioritize: If specified, repositories owned by the 645 given entity will be prioritized in the search. 646 :returns: All repositories matching the query. If there are many 647 repositories matching this query, this may take some time. 648 """ 649 650 params = {} 651 652 if query is not None: 653 params["q"] = query 654 if topic: 655 params["topic"] = topic 656 if include_description: 657 params["include_description"] = include_description 658 if user is not None: 659 params["user"] = user.id 660 if owner_to_prioritize is not None: 661 params["owner_to_prioritize"] = owner_to_prioritize.id 662 663 responses = allspice_client.requests_get_paginated(cls.REPO_SEARCH, params=params) 664 665 return [Repository.parse_response(allspice_client, response) for response in responses] 666 667 _patchable_fields: ClassVar[set[str]] = { 668 "allow_manual_merge", 669 "allow_merge_commits", 670 "allow_rebase", 671 "allow_rebase_explicit", 672 "allow_rebase_update", 673 "allow_squash_merge", 674 "archived", 675 "autodetect_manual_merge", 676 "default_branch", 677 "default_delete_branch_after_merge", 678 "default_merge_style", 679 "description", 680 "enable_prune", 681 "external_tracker", 682 "external_wiki", 683 "has_actions", 684 "has_issues", 685 "has_projects", 686 "has_pull_requests", 687 "has_wiki", 688 "ignore_whitespace_conflicts", 689 "internal_tracker", 690 "mirror_interval", 691 "name", 692 "private", 693 "template", 694 "website", 695 } 696 697 def commit(self): 698 args = {"owner": self.owner.username, "name": self.name} 699 self._commit(args) 700 701 def get_branches(self) -> List["Branch"]: 702 """Get all the Branches of this Repository.""" 703 704 results = self.allspice_client.requests_get_paginated( 705 Repository.REPO_BRANCHES % (self.owner.username, self.name) 706 ) 707 return [Branch.parse_response(self.allspice_client, result) for result in results] 708 709 def get_branch(self, name: str) -> "Branch": 710 """Get a specific Branch of this Repository.""" 711 result = self.allspice_client.requests_get( 712 Repository.REPO_BRANCH.format(owner=self.owner.username, repo=self.name, branch=name) 713 ) 714 return Branch.parse_response(self.allspice_client, result) 715 716 def add_branch(self, create_from: Ref, newname: str) -> "Branch": 717 """Add a branch to the repository""" 718 # Note: will only work with gitea 1.13 or higher! 719 720 ref_name = Util.data_params_for_ref(create_from) 721 if "ref" not in ref_name: 722 raise ValueError("create_from must be a Branch, Commit or string") 723 ref_name = ref_name["ref"] 724 725 data = {"new_branch_name": newname, "old_ref_name": ref_name} 726 result = self.allspice_client.requests_post( 727 Repository.REPO_BRANCHES % (self.owner.username, self.name), data=data 728 ) 729 return Branch.parse_response(self.allspice_client, result) 730 731 def get_issues( 732 self, 733 state: Literal["open", "closed", "all"] = "all", 734 search_query: Optional[str] = None, 735 labels: Optional[List[str]] = None, 736 milestones: Optional[List[Union[Milestone, str]]] = None, 737 assignee: Optional[Union[User, str]] = None, 738 since: Optional[datetime] = None, 739 before: Optional[datetime] = None, 740 ) -> List["Issue"]: 741 """ 742 Get all Issues of this Repository (open and closed) 743 744 https://hub.allspice.io/api/swagger#/repository/repoListIssues 745 746 All params of this method are optional filters. If you don't specify a filter, it 747 will not be applied. 748 749 :param state: The state of the Issues to get. If None, all Issues are returned. 750 :param search_query: Filter issues by text. This is equivalent to searching for 751 `search_query` in the Issues on the web interface. 752 :param labels: Filter issues by labels. 753 :param milestones: Filter issues by milestones. 754 :param assignee: Filter issues by the assigned user. 755 :param since: Filter issues by the date they were created. 756 :param before: Filter issues by the date they were created. 757 :return: A list of Issues. 758 """ 759 760 data = { 761 "state": state, 762 } 763 if search_query: 764 data["q"] = search_query 765 if labels: 766 data["labels"] = ",".join(labels) 767 if milestones: 768 data["milestone"] = ",".join( 769 [ 770 milestone.name if isinstance(milestone, Milestone) else milestone 771 for milestone in milestones 772 ] 773 ) 774 if assignee: 775 if isinstance(assignee, User): 776 data["assignee"] = assignee.username 777 else: 778 data["assignee"] = assignee 779 if since: 780 data["since"] = Util.format_time(since) 781 if before: 782 data["before"] = Util.format_time(before) 783 784 results = self.allspice_client.requests_get_paginated( 785 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 786 params=data, 787 ) 788 789 issues = [] 790 for result in results: 791 issue = Issue.parse_response(self.allspice_client, result) 792 # See Issue.request 793 setattr(issue, "_repository", self) 794 # This is mostly for compatibility with an older implementation 795 Issue._add_read_property("repo", self, issue) 796 issues.append(issue) 797 798 return issues 799 800 def get_design_reviews( 801 self, 802 state: Literal["open", "closed", "all"] = "all", 803 milestone: Optional[Union[Milestone, str]] = None, 804 labels: Optional[List[str]] = None, 805 ) -> List["DesignReview"]: 806 """ 807 Get all Design Reviews of this Repository. 808 809 https://hub.allspice.io/api/swagger#/repository/repoListPullRequests 810 811 :param state: The state of the Design Reviews to get. If None, all Design Reviews 812 are returned. 813 :param milestone: The milestone of the Design Reviews to get. 814 :param labels: A list of label IDs to filter DRs by. 815 :return: A list of Design Reviews. 816 """ 817 818 params = { 819 "state": state, 820 } 821 if milestone: 822 if isinstance(milestone, Milestone): 823 params["milestone"] = milestone.name 824 else: 825 params["milestone"] = milestone 826 if labels: 827 params["labels"] = ",".join(labels) 828 829 results = self.allspice_client.requests_get_paginated( 830 self.REPO_DESIGN_REVIEWS.format(owner=self.owner.username, repo=self.name), 831 params=params, 832 ) 833 return [DesignReview.parse_response(self.allspice_client, result) for result in results] 834 835 def get_commits( 836 self, 837 sha: Optional[str] = None, 838 path: Optional[str] = None, 839 stat: bool = True, 840 ) -> List["Commit"]: 841 """ 842 Get all the Commits of this Repository. 843 844 https://hub.allspice.io/api/swagger#/repository/repoGetAllCommits 845 846 :param sha: The SHA of the commit to start listing commits from. 847 :param path: filepath of a file/dir. 848 :param stat: Include the number of additions and deletions in the response. 849 Disable for speedup. 850 :return: A list of Commits. 851 """ 852 853 data = {} 854 if sha: 855 data["sha"] = sha 856 if path: 857 data["path"] = path 858 if not stat: 859 data["stat"] = False 860 861 try: 862 results = self.allspice_client.requests_get_paginated( 863 Repository.REPO_COMMITS % (self.owner.username, self.name), 864 params=data, 865 ) 866 except ConflictException as err: 867 logging.warning(err) 868 logging.warning("Repository %s/%s is Empty" % (self.owner.username, self.name)) 869 results = [] 870 return [Commit.parse_response(self.allspice_client, result) for result in results] 871 872 def get_issues_state(self, state) -> List["Issue"]: 873 """ 874 DEPRECATED: Use get_issues() instead. 875 876 Get issues of state Issue.open or Issue.closed of a repository. 877 """ 878 879 assert state in [Issue.OPENED, Issue.CLOSED] 880 issues = [] 881 data = {"state": state} 882 results = self.allspice_client.requests_get_paginated( 883 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 884 params=data, 885 ) 886 for result in results: 887 issue = Issue.parse_response(self.allspice_client, result) 888 # adding data not contained in the issue response 889 # See Issue.request() 890 setattr(issue, "_repository", self) 891 Issue._add_read_property("repo", self, issue) 892 Issue._add_read_property("owner", self.owner, issue) 893 issues.append(issue) 894 return issues 895 896 def get_times(self): 897 results = self.allspice_client.requests_get( 898 Repository.REPO_TIMES % (self.owner.username, self.name) 899 ) 900 return results 901 902 def get_user_time(self, username) -> float: 903 if isinstance(username, User): 904 username = username.username 905 results = self.allspice_client.requests_get( 906 Repository.REPO_USER_TIME % (self.owner.username, self.name, username) 907 ) 908 time = sum(r["time"] for r in results) 909 return time 910 911 def get_full_name(self) -> str: 912 return self.owner.username + "/" + self.name 913 914 def create_issue(self, title, assignees=frozenset(), description="") -> ApiObject: 915 data = { 916 "assignees": assignees, 917 "body": description, 918 "closed": False, 919 "title": title, 920 } 921 result = self.allspice_client.requests_post( 922 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 923 data=data, 924 ) 925 926 issue = Issue.parse_response(self.allspice_client, result) 927 setattr(issue, "_repository", self) 928 Issue._add_read_property("repo", self, issue) 929 return issue 930 931 def create_design_review( 932 self, 933 title: str, 934 head: Union[Branch, str], 935 base: Union[Branch, str], 936 assignees: Optional[Set[Union[User, str]]] = None, 937 body: Optional[str] = None, 938 due_date: Optional[datetime] = None, 939 milestone: Optional["Milestone"] = None, 940 ) -> "DesignReview": 941 """ 942 Create a new Design Review. 943 944 See https://hub.allspice.io/api/swagger#/repository/repoCreatePullRequest 945 946 :param title: Title of the Design Review 947 :param head: Branch or name of the branch to merge into the base branch 948 :param base: Branch or name of the branch to merge into 949 :param assignees: Optional. A list of users to assign this review. List can be of 950 User objects or of usernames. 951 :param body: An Optional Description for the Design Review. 952 :param due_date: An Optional Due date for the Design Review. 953 :param milestone: An Optional Milestone for the Design Review 954 :return: The created Design Review 955 """ 956 957 data: dict[str, Any] = { 958 "title": title, 959 } 960 961 if isinstance(head, Branch): 962 data["head"] = head.name 963 else: 964 data["head"] = head 965 if isinstance(base, Branch): 966 data["base"] = base.name 967 else: 968 data["base"] = base 969 if assignees: 970 data["assignees"] = [a.username if isinstance(a, User) else a for a in assignees] 971 if body: 972 data["body"] = body 973 if due_date: 974 data["due_date"] = Util.format_time(due_date) 975 if milestone: 976 data["milestone"] = milestone.id 977 978 result = self.allspice_client.requests_post( 979 self.REPO_DESIGN_REVIEWS.format(owner=self.owner.username, repo=self.name), 980 data=data, 981 ) 982 983 return DesignReview.parse_response(self.allspice_client, result) 984 985 def create_milestone( 986 self, 987 title: str, 988 description: str, 989 due_date: Optional[str] = None, 990 state: str = "open", 991 ) -> "Milestone": 992 url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) 993 data = {"title": title, "description": description, "state": state} 994 if due_date: 995 data["due_date"] = due_date 996 result = self.allspice_client.requests_post(url, data=data) 997 return Milestone.parse_response(self.allspice_client, result) 998 999 def create_gitea_hook(self, hook_url: str, events: List[str]): 1000 url = f"/repos/{self.owner.username}/{self.name}/hooks" 1001 data = { 1002 "type": "gitea", 1003 "config": {"content_type": "json", "url": hook_url}, 1004 "events": events, 1005 "active": True, 1006 } 1007 return self.allspice_client.requests_post(url, data=data) 1008 1009 def list_hooks(self): 1010 url = f"/repos/{self.owner.username}/{self.name}/hooks" 1011 return self.allspice_client.requests_get(url) 1012 1013 def delete_hook(self, id: str): 1014 url = f"/repos/{self.owner.username}/{self.name}/hooks/{id}" 1015 self.allspice_client.requests_delete(url) 1016 1017 def is_collaborator(self, username) -> bool: 1018 if isinstance(username, User): 1019 username = username.username 1020 try: 1021 # returns 204 if its ok, 404 if its not 1022 self.allspice_client.requests_get( 1023 Repository.REPO_IS_COLLABORATOR % (self.owner.username, self.name, username) 1024 ) 1025 return True 1026 except Exception: 1027 return False 1028 1029 def get_users_with_access(self) -> Sequence[User]: 1030 url = f"/repos/{self.owner.username}/{self.name}/collaborators" 1031 response = self.allspice_client.requests_get(url) 1032 collabs = [User.parse_response(self.allspice_client, user) for user in response] 1033 if isinstance(self.owner, User): 1034 return [*collabs, self.owner] 1035 else: 1036 # owner must be org 1037 teams = self.owner.get_teams() 1038 for team in teams: 1039 team_repos = team.get_repos() 1040 if self.name in [n.name for n in team_repos]: 1041 collabs += team.get_members() 1042 return collabs 1043 1044 def remove_collaborator(self, user_name: str): 1045 url = f"/repos/{self.owner.username}/{self.name}/collaborators/{user_name}" 1046 self.allspice_client.requests_delete(url) 1047 1048 def transfer_ownership( 1049 self, 1050 new_owner: Union[User, Organization], 1051 new_teams: Set[Team] | FrozenSet[Team] = frozenset(), 1052 ): 1053 url = Repository.REPO_TRANSFER.format(owner=self.owner.username, repo=self.name) 1054 data: dict[str, Any] = {"new_owner": new_owner.username} 1055 if isinstance(new_owner, Organization): 1056 new_team_ids = [team.id for team in new_teams if team in new_owner.get_teams()] 1057 data["team_ids"] = new_team_ids 1058 self.allspice_client.requests_post(url, data=data) 1059 # TODO: make sure this instance is either updated or discarded 1060 1061 def get_git_content( 1062 self, 1063 ref: Optional["Ref"] = None, 1064 commit: "Optional[Commit]" = None, 1065 ) -> List[Content]: 1066 """ 1067 Get the metadata for all files in the root directory. 1068 1069 https://hub.allspice.io/api/swagger#/repository/repoGetContentsList 1070 1071 :param ref: branch or commit to get content from 1072 :param commit: commit to get content from (deprecated) 1073 """ 1074 url = f"/repos/{self.owner.username}/{self.name}/contents" 1075 data = Util.data_params_for_ref(ref or commit) 1076 1077 result = [ 1078 Content.parse_response(self.allspice_client, f) 1079 for f in self.allspice_client.requests_get(url, data) 1080 ] 1081 return result 1082 1083 def get_tree(self, ref: Optional[Ref] = None, recursive: bool = False) -> List[GitEntry]: 1084 """ 1085 Get the repository's tree on a given ref. 1086 1087 By default, this will only return the top-level entries in the tree. If you want 1088 to get the entire tree, set `recursive` to True. 1089 1090 :param ref: The ref to get the tree from. If not provided, the default branch is used. 1091 :param recursive: Whether to get the entire tree or just the top-level entries. 1092 """ 1093 1094 ref = Util.data_params_for_ref(ref).get("ref", self.default_branch) 1095 url = self.REPO_GET_TREE.format(owner=self.owner.username, repo=self.name, ref=ref) 1096 params = {"recursive": recursive} 1097 results = self.allspice_client.requests_get_paginated(url, params=params) 1098 return [GitEntry.parse_response(self.allspice_client, result) for result in results] 1099 1100 def get_file_content( 1101 self, 1102 content: Content, 1103 ref: Optional[Ref] = None, 1104 commit: Optional[Commit] = None, 1105 ) -> Union[str, List["Content"]]: 1106 """https://hub.allspice.io/api/swagger#/repository/repoGetContents""" 1107 url = f"/repos/{self.owner.username}/{self.name}/contents/{content.path}" 1108 data = Util.data_params_for_ref(ref or commit) 1109 1110 if content.type == Content.FILE: 1111 return self.allspice_client.requests_get(url, data)["content"] 1112 else: 1113 return [ 1114 Content.parse_response(self.allspice_client, f) 1115 for f in self.allspice_client.requests_get(url, data) 1116 ] 1117 1118 def get_raw_file( 1119 self, 1120 file_path: str, 1121 ref: Optional[Ref] = None, 1122 ) -> bytes: 1123 """ 1124 Get the raw, binary data of a single file. 1125 1126 Note 1: if the file you are requesting is a text file, you might want to 1127 use .decode() on the result to get a string. For example: 1128 1129 content = repo.get_raw_file("file.txt").decode("utf-8") 1130 1131 Note 2: this method will store the entire file in memory. If you want 1132 to download a large file, you might want to use `download_to_file` 1133 instead. 1134 1135 See https://hub.allspice.io/api/swagger#/repository/repoGetRawFileOrLFS 1136 1137 :param file_path: The path to the file to get. 1138 :param ref: The branch or commit to get the file from. If not provided, 1139 the default branch is used. 1140 """ 1141 1142 url = self.REPO_GET_MEDIA.format( 1143 owner=self.owner.username, 1144 repo=self.name, 1145 path=file_path, 1146 ) 1147 params = Util.data_params_for_ref(ref) 1148 return self.allspice_client.requests_get_raw(url, params=params) 1149 1150 def download_to_file( 1151 self, 1152 file_path: str, 1153 io: IO, 1154 ref: Optional[Ref] = None, 1155 ) -> None: 1156 """ 1157 Download the binary data of a file to a file-like object. 1158 1159 Example: 1160 1161 with open("schematic.DSN", "wb") as f: 1162 Repository.download_to_file("Schematics/my_schematic.DSN", f) 1163 1164 :param file_path: The path to the file in the repository from the root 1165 of the repository. 1166 :param io: The file-like object to write the data to. 1167 """ 1168 1169 url = self.allspice_client._AllSpice__get_url( 1170 self.REPO_GET_MEDIA.format( 1171 owner=self.owner.username, 1172 repo=self.name, 1173 path=file_path, 1174 ) 1175 ) 1176 params = Util.data_params_for_ref(ref) 1177 response = self.allspice_client.requests.get( 1178 url, 1179 params=params, 1180 headers=self.allspice_client.headers, 1181 stream=True, 1182 ) 1183 1184 for chunk in response.iter_content(chunk_size=4096): 1185 if chunk: 1186 io.write(chunk) 1187 1188 def get_generated_json(self, content: Union[Content, str], ref: Optional[Ref] = None) -> dict: 1189 """ 1190 Get the json blob for a cad file if it exists, otherwise enqueue 1191 a new job and return a 503 status. 1192 1193 WARNING: This is still experimental and not recommended for critical 1194 applications. The structure and content of the returned dictionary can 1195 change at any time. 1196 1197 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceJSON 1198 """ 1199 1200 if isinstance(content, Content): 1201 content = content.path 1202 1203 url = self.REPO_GET_ALLSPICE_JSON.format( 1204 owner=self.owner.username, 1205 repo=self.name, 1206 content=content, 1207 ) 1208 data = Util.data_params_for_ref(ref) 1209 return self.allspice_client.requests_get(url, data) 1210 1211 def get_generated_svg(self, content: Union[Content, str], ref: Optional[Ref] = None) -> bytes: 1212 """ 1213 Get the svg blob for a cad file if it exists, otherwise enqueue 1214 a new job and return a 503 status. 1215 1216 WARNING: This is still experimental and not yet recommended for 1217 critical applications. The content of the returned svg can change 1218 at any time. 1219 1220 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceSVG 1221 """ 1222 1223 if isinstance(content, Content): 1224 content = content.path 1225 1226 url = self.REPO_GET_ALLSPICE_SVG.format( 1227 owner=self.owner.username, 1228 repo=self.name, 1229 content=content, 1230 ) 1231 data = Util.data_params_for_ref(ref) 1232 return self.allspice_client.requests_get_raw(url, data) 1233 1234 def get_generated_projectdata( 1235 self, content: Union[Content, str], ref: Optional[Ref] = None 1236 ) -> dict: 1237 """ 1238 Get the json project data based on the cad file provided 1239 1240 WARNING: This is still experimental and not yet recommended for 1241 critical applications. The content of the returned dictionary can change 1242 at any time. 1243 1244 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceProject 1245 """ 1246 if isinstance(content, Content): 1247 content = content.path 1248 1249 url = self.REPO_GET_ALLSPICE_PROJECT.format( 1250 owner=self.owner.username, 1251 repo=self.name, 1252 content=content, 1253 ) 1254 data = Util.data_params_for_ref(ref) 1255 return self.allspice_client.requests_get(url, data) 1256 1257 def create_file(self, file_path: str, content: str, data: Optional[dict] = None): 1258 """https://hub.allspice.io/api/swagger#/repository/repoCreateFile""" 1259 if not data: 1260 data = {} 1261 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1262 data.update({"content": content}) 1263 return self.allspice_client.requests_post(url, data) 1264 1265 def change_file(self, file_path: str, file_sha: str, content: str, data: Optional[dict] = None): 1266 """https://hub.allspice.io/api/swagger#/repository/repoCreateFile""" 1267 if not data: 1268 data = {} 1269 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1270 data.update({"sha": file_sha, "content": content}) 1271 return self.allspice_client.requests_put(url, data) 1272 1273 def delete_file(self, file_path: str, file_sha: str, data: Optional[dict] = None): 1274 """https://hub.allspice.io/api/swagger#/repository/repoDeleteFile""" 1275 if not data: 1276 data = {} 1277 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1278 data.update({"sha": file_sha}) 1279 return self.allspice_client.requests_delete(url, data) 1280 1281 def get_archive( 1282 self, 1283 ref: Ref = "main", 1284 archive_format: ArchiveFormat = ArchiveFormat.ZIP, 1285 ) -> bytes: 1286 """ 1287 Download all the files in a specific ref of a repository as a zip or tarball 1288 archive. 1289 1290 https://hub.allspice.io/api/swagger#/repository/repoGetArchive 1291 1292 :param ref: branch or commit to get content from, defaults to the "main" branch 1293 :param archive_format: zip or tar, defaults to zip 1294 """ 1295 1296 ref_string = Util.data_params_for_ref(ref)["ref"] 1297 url = self.REPO_GET_ARCHIVE.format( 1298 owner=self.owner.username, 1299 repo=self.name, 1300 ref=ref_string, 1301 format=archive_format.value, 1302 ) 1303 return self.allspice_client.requests_get_raw(url) 1304 1305 def get_topics(self) -> list[str]: 1306 """ 1307 Gets the list of topics on this repository. 1308 1309 See http://localhost:3000/api/swagger#/repository/repoListTopics 1310 """ 1311 1312 url = self.REPO_GET_TOPICS.format( 1313 owner=self.owner.username, 1314 repo=self.name, 1315 ) 1316 return self.allspice_client.requests_get(url)["topics"] 1317 1318 def add_topic(self, topic: str): 1319 """ 1320 Adds a topic to the repository. 1321 1322 See https://hub.allspice.io/api/swagger#/repository/repoAddTopic 1323 1324 :param topic: The topic to add. Topic names must consist only of 1325 lowercase letters, numnbers and dashes (-), and cannot start with 1326 dashes. Topic names also must be under 35 characters long. 1327 """ 1328 1329 url = self.REPO_ADD_TOPIC.format(owner=self.owner.username, repo=self.name, topic=topic) 1330 self.allspice_client.requests_put(url) 1331 1332 def create_release( 1333 self, 1334 tag_name: str, 1335 name: Optional[str] = None, 1336 body: Optional[str] = None, 1337 draft: bool = False, 1338 ): 1339 """ 1340 Create a release for this repository. The release will be created for 1341 the tag with the given name. If there is no tag with this name, create 1342 the tag first. 1343 1344 See https://hub.allspice.io/api/swagger#/repository/repoCreateRelease 1345 """ 1346 1347 url = self.REPO_GET_RELEASES.format(owner=self.owner.username, repo=self.name) 1348 data = { 1349 "tag_name": tag_name, 1350 "draft": draft, 1351 } 1352 if name is not None: 1353 data["name"] = name 1354 if body is not None: 1355 data["body"] = body 1356 response = self.allspice_client.requests_post(url, data) 1357 return Release.parse_response(self.allspice_client, response, self) 1358 1359 def get_releases( 1360 self, draft: Optional[bool] = None, pre_release: Optional[bool] = None 1361 ) -> List[Release]: 1362 """ 1363 Get the list of releases for this repository. 1364 1365 See https://hub.allspice.io/api/swagger#/repository/repoListReleases 1366 """ 1367 1368 data = {} 1369 1370 if draft is not None: 1371 data["draft"] = draft 1372 if pre_release is not None: 1373 data["pre-release"] = pre_release 1374 1375 url = self.REPO_GET_RELEASES.format(owner=self.owner.username, repo=self.name) 1376 responses = self.allspice_client.requests_get_paginated(url, params=data) 1377 1378 return [ 1379 Release.parse_response(self.allspice_client, response, self) for response in responses 1380 ] 1381 1382 def get_latest_release(self) -> Release: 1383 """ 1384 Get the latest release for this repository. 1385 1386 See https://hub.allspice.io/api/swagger#/repository/repoGetLatestRelease 1387 """ 1388 1389 url = self.REPO_GET_LATEST_RELEASE.format(owner=self.owner.username, repo=self.name) 1390 response = self.allspice_client.requests_get(url) 1391 release = Release.parse_response(self.allspice_client, response, self) 1392 return release 1393 1394 def get_release_by_tag(self, tag: str) -> Release: 1395 """ 1396 Get a release by its tag. 1397 1398 See https://hub.allspice.io/api/swagger#/repository/repoGetReleaseByTag 1399 """ 1400 1401 url = self.REPO_GET_RELEASE_BY_TAG.format( 1402 owner=self.owner.username, repo=self.name, tag=tag 1403 ) 1404 response = self.allspice_client.requests_get(url) 1405 release = Release.parse_response(self.allspice_client, response, self) 1406 return release 1407 1408 def get_commit_statuses( 1409 self, 1410 commit: Union[str, Commit], 1411 sort: Optional[CommitStatusSort] = None, 1412 state: Optional[CommitStatusState] = None, 1413 ) -> List[CommitStatus]: 1414 """ 1415 Get a list of statuses for a commit. 1416 1417 This is roughly equivalent to the Commit.get_statuses method, but this 1418 method allows you to sort and filter commits and is more convenient if 1419 you have a commit SHA and don't need to get the commit itself. 1420 1421 See https://hub.allspice.io/api/swagger#/repository/repoListStatuses 1422 """ 1423 1424 if isinstance(commit, Commit): 1425 commit = commit.sha 1426 1427 params = {} 1428 if sort is not None: 1429 params["sort"] = sort.value 1430 if state is not None: 1431 params["state"] = state.value 1432 1433 url = self.REPO_GET_COMMIT_STATUS.format( 1434 owner=self.owner.username, repo=self.name, sha=commit 1435 ) 1436 response = self.allspice_client.requests_get_paginated(url, params=params) 1437 return [CommitStatus.parse_response(self.allspice_client, status) for status in response] 1438 1439 def create_commit_status( 1440 self, 1441 commit: Union[str, Commit], 1442 context: Optional[str] = None, 1443 description: Optional[str] = None, 1444 state: Optional[CommitStatusState] = None, 1445 target_url: Optional[str] = None, 1446 ) -> CommitStatus: 1447 """ 1448 Create a status on a commit. 1449 1450 See https://hub.allspice.io/api/swagger#/repository/repoCreateStatus 1451 """ 1452 1453 if isinstance(commit, Commit): 1454 commit = commit.sha 1455 1456 data = {} 1457 if context is not None: 1458 data["context"] = context 1459 if description is not None: 1460 data["description"] = description 1461 if state is not None: 1462 data["state"] = state.value 1463 if target_url is not None: 1464 data["target_url"] = target_url 1465 1466 url = self.REPO_GET_COMMIT_STATUS.format( 1467 owner=self.owner.username, repo=self.name, sha=commit 1468 ) 1469 response = self.allspice_client.requests_post(url, data=data) 1470 return CommitStatus.parse_response(self.allspice_client, response) 1471 1472 def delete(self): 1473 self.allspice_client.requests_delete( 1474 Repository.REPO_DELETE % (self.owner.username, self.name) 1475 ) 1476 self.deleted = True
owner: Union[User, Organization]
@classmethod
def
search( cls, allspice_client, query: Optional[str] = None, topic: bool = False, include_description: bool = False, user: Optional[User] = None, owner_to_prioritize: Union[User, Organization, NoneType] = None) -> list[Repository]:
622 @classmethod 623 def search( 624 cls, 625 allspice_client, 626 query: Optional[str] = None, 627 topic: bool = False, 628 include_description: bool = False, 629 user: Optional[User] = None, 630 owner_to_prioritize: Union[User, Organization, None] = None, 631 ) -> list[Repository]: 632 """ 633 Search for repositories. 634 635 See https://hub.allspice.io/api/swagger#/repository/repoSearch 636 637 :param query: The query string to search for 638 :param topic: If true, the query string will only be matched against the 639 repository's topic. 640 :param include_description: If true, the query string will be matched 641 against the repository's description as well. 642 :param user: If specified, only repositories that this user owns or 643 contributes to will be searched. 644 :param owner_to_prioritize: If specified, repositories owned by the 645 given entity will be prioritized in the search. 646 :returns: All repositories matching the query. If there are many 647 repositories matching this query, this may take some time. 648 """ 649 650 params = {} 651 652 if query is not None: 653 params["q"] = query 654 if topic: 655 params["topic"] = topic 656 if include_description: 657 params["include_description"] = include_description 658 if user is not None: 659 params["user"] = user.id 660 if owner_to_prioritize is not None: 661 params["owner_to_prioritize"] = owner_to_prioritize.id 662 663 responses = allspice_client.requests_get_paginated(cls.REPO_SEARCH, params=params) 664 665 return [Repository.parse_response(allspice_client, response) for response in responses]
Search for repositories.
See allspice.allspice.io/api/swagger#/repository/repoSearch">https://huballspice.allspice.io/api/swagger#/repository/repoSearch
Parameters
- query: The query string to search for
- topic: If true, the query string will only be matched against the repository's topic.
- include_description: If true, the query string will be matched against the repository's description as well.
- user: If specified, only repositories that this user owns or contributes to will be searched.
- owner_to_prioritize: If specified, repositories owned by the given entity will be prioritized in the search. :returns: All repositories matching the query. If there are many repositories matching this query, this may take some time.
701 def get_branches(self) -> List["Branch"]: 702 """Get all the Branches of this Repository.""" 703 704 results = self.allspice_client.requests_get_paginated( 705 Repository.REPO_BRANCHES % (self.owner.username, self.name) 706 ) 707 return [Branch.parse_response(self.allspice_client, result) for result in results]
Get all the Branches of this Repository.
709 def get_branch(self, name: str) -> "Branch": 710 """Get a specific Branch of this Repository.""" 711 result = self.allspice_client.requests_get( 712 Repository.REPO_BRANCH.format(owner=self.owner.username, repo=self.name, branch=name) 713 ) 714 return Branch.parse_response(self.allspice_client, result)
Get a specific Branch of this Repository.
716 def add_branch(self, create_from: Ref, newname: str) -> "Branch": 717 """Add a branch to the repository""" 718 # Note: will only work with gitea 1.13 or higher! 719 720 ref_name = Util.data_params_for_ref(create_from) 721 if "ref" not in ref_name: 722 raise ValueError("create_from must be a Branch, Commit or string") 723 ref_name = ref_name["ref"] 724 725 data = {"new_branch_name": newname, "old_ref_name": ref_name} 726 result = self.allspice_client.requests_post( 727 Repository.REPO_BRANCHES % (self.owner.username, self.name), data=data 728 ) 729 return Branch.parse_response(self.allspice_client, result)
Add a branch to the repository
def
get_issues( self, state: Literal['open', 'closed', 'all'] = 'all', search_query: Optional[str] = None, labels: Optional[List[str]] = None, milestones: Optional[List[Union[Milestone, str]]] = None, assignee: Union[User, str, NoneType] = None, since: Optional[datetime.datetime] = None, before: Optional[datetime.datetime] = None) -> List[Issue]:
731 def get_issues( 732 self, 733 state: Literal["open", "closed", "all"] = "all", 734 search_query: Optional[str] = None, 735 labels: Optional[List[str]] = None, 736 milestones: Optional[List[Union[Milestone, str]]] = None, 737 assignee: Optional[Union[User, str]] = None, 738 since: Optional[datetime] = None, 739 before: Optional[datetime] = None, 740 ) -> List["Issue"]: 741 """ 742 Get all Issues of this Repository (open and closed) 743 744 https://hub.allspice.io/api/swagger#/repository/repoListIssues 745 746 All params of this method are optional filters. If you don't specify a filter, it 747 will not be applied. 748 749 :param state: The state of the Issues to get. If None, all Issues are returned. 750 :param search_query: Filter issues by text. This is equivalent to searching for 751 `search_query` in the Issues on the web interface. 752 :param labels: Filter issues by labels. 753 :param milestones: Filter issues by milestones. 754 :param assignee: Filter issues by the assigned user. 755 :param since: Filter issues by the date they were created. 756 :param before: Filter issues by the date they were created. 757 :return: A list of Issues. 758 """ 759 760 data = { 761 "state": state, 762 } 763 if search_query: 764 data["q"] = search_query 765 if labels: 766 data["labels"] = ",".join(labels) 767 if milestones: 768 data["milestone"] = ",".join( 769 [ 770 milestone.name if isinstance(milestone, Milestone) else milestone 771 for milestone in milestones 772 ] 773 ) 774 if assignee: 775 if isinstance(assignee, User): 776 data["assignee"] = assignee.username 777 else: 778 data["assignee"] = assignee 779 if since: 780 data["since"] = Util.format_time(since) 781 if before: 782 data["before"] = Util.format_time(before) 783 784 results = self.allspice_client.requests_get_paginated( 785 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 786 params=data, 787 ) 788 789 issues = [] 790 for result in results: 791 issue = Issue.parse_response(self.allspice_client, result) 792 # See Issue.request 793 setattr(issue, "_repository", self) 794 # This is mostly for compatibility with an older implementation 795 Issue._add_read_property("repo", self, issue) 796 issues.append(issue) 797 798 return issues
Get all Issues of this Repository (open and closed)
allspice.allspice.io/api/swagger#/repository/repoListIssues">https://huballspice.allspice.io/api/swagger#/repository/repoListIssues
All params of this method are optional filters. If you don't specify a filter, it will not be applied.
Parameters
- state: The state of the Issues to get. If None, all Issues are returned.
- search_query: Filter issues by text. This is equivalent to searching for
search_queryin the Issues on the web interface. - labels: Filter issues by labels.
- milestones: Filter issues by milestones.
- assignee: Filter issues by the assigned user.
- since: Filter issues by the date they were created.
- before: Filter issues by the date they were created.
Returns
A list of Issues.
def
get_design_reviews( self, state: Literal['open', 'closed', 'all'] = 'all', milestone: Union[Milestone, str, NoneType] = None, labels: Optional[List[str]] = None) -> List[DesignReview]:
800 def get_design_reviews( 801 self, 802 state: Literal["open", "closed", "all"] = "all", 803 milestone: Optional[Union[Milestone, str]] = None, 804 labels: Optional[List[str]] = None, 805 ) -> List["DesignReview"]: 806 """ 807 Get all Design Reviews of this Repository. 808 809 https://hub.allspice.io/api/swagger#/repository/repoListPullRequests 810 811 :param state: The state of the Design Reviews to get. If None, all Design Reviews 812 are returned. 813 :param milestone: The milestone of the Design Reviews to get. 814 :param labels: A list of label IDs to filter DRs by. 815 :return: A list of Design Reviews. 816 """ 817 818 params = { 819 "state": state, 820 } 821 if milestone: 822 if isinstance(milestone, Milestone): 823 params["milestone"] = milestone.name 824 else: 825 params["milestone"] = milestone 826 if labels: 827 params["labels"] = ",".join(labels) 828 829 results = self.allspice_client.requests_get_paginated( 830 self.REPO_DESIGN_REVIEWS.format(owner=self.owner.username, repo=self.name), 831 params=params, 832 ) 833 return [DesignReview.parse_response(self.allspice_client, result) for result in results]
Get all Design Reviews of this Repository.
allspice.allspice.io/api/swagger#/repository/repoListPullRequests">https://huballspice.allspice.io/api/swagger#/repository/repoListPullRequests
Parameters
- state: The state of the Design Reviews to get. If None, all Design Reviews are returned.
- milestone: The milestone of the Design Reviews to get.
- labels: A list of label IDs to filter DRs by.
Returns
A list of Design Reviews.
def
get_commits( self, sha: Optional[str] = None, path: Optional[str] = None, stat: bool = True) -> List[Commit]:
835 def get_commits( 836 self, 837 sha: Optional[str] = None, 838 path: Optional[str] = None, 839 stat: bool = True, 840 ) -> List["Commit"]: 841 """ 842 Get all the Commits of this Repository. 843 844 https://hub.allspice.io/api/swagger#/repository/repoGetAllCommits 845 846 :param sha: The SHA of the commit to start listing commits from. 847 :param path: filepath of a file/dir. 848 :param stat: Include the number of additions and deletions in the response. 849 Disable for speedup. 850 :return: A list of Commits. 851 """ 852 853 data = {} 854 if sha: 855 data["sha"] = sha 856 if path: 857 data["path"] = path 858 if not stat: 859 data["stat"] = False 860 861 try: 862 results = self.allspice_client.requests_get_paginated( 863 Repository.REPO_COMMITS % (self.owner.username, self.name), 864 params=data, 865 ) 866 except ConflictException as err: 867 logging.warning(err) 868 logging.warning("Repository %s/%s is Empty" % (self.owner.username, self.name)) 869 results = [] 870 return [Commit.parse_response(self.allspice_client, result) for result in results]
Get all the Commits of this Repository.
allspice.allspice.io/api/swagger#/repository/repoGetAllCommits">https://huballspice.allspice.io/api/swagger#/repository/repoGetAllCommits
Parameters
- sha: The SHA of the commit to start listing commits from.
- path: filepath of a file/dir.
- stat: Include the number of additions and deletions in the response. Disable for speedup.
Returns
A list of Commits.
872 def get_issues_state(self, state) -> List["Issue"]: 873 """ 874 DEPRECATED: Use get_issues() instead. 875 876 Get issues of state Issue.open or Issue.closed of a repository. 877 """ 878 879 assert state in [Issue.OPENED, Issue.CLOSED] 880 issues = [] 881 data = {"state": state} 882 results = self.allspice_client.requests_get_paginated( 883 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 884 params=data, 885 ) 886 for result in results: 887 issue = Issue.parse_response(self.allspice_client, result) 888 # adding data not contained in the issue response 889 # See Issue.request() 890 setattr(issue, "_repository", self) 891 Issue._add_read_property("repo", self, issue) 892 Issue._add_read_property("owner", self.owner, issue) 893 issues.append(issue) 894 return issues
DEPRECATED: Use get_issues() instead.
Get issues of state Issue.open or Issue.closed of a repository.
def
get_user_time(self, username) -> float:
902 def get_user_time(self, username) -> float: 903 if isinstance(username, User): 904 username = username.username 905 results = self.allspice_client.requests_get( 906 Repository.REPO_USER_TIME % (self.owner.username, self.name, username) 907 ) 908 time = sum(r["time"] for r in results) 909 return time
def
create_issue( self, title, assignees=frozenset(), description='') -> allspice.baseapiobject.ApiObject:
914 def create_issue(self, title, assignees=frozenset(), description="") -> ApiObject: 915 data = { 916 "assignees": assignees, 917 "body": description, 918 "closed": False, 919 "title": title, 920 } 921 result = self.allspice_client.requests_post( 922 Repository.REPO_ISSUES.format(owner=self.owner.username, repo=self.name), 923 data=data, 924 ) 925 926 issue = Issue.parse_response(self.allspice_client, result) 927 setattr(issue, "_repository", self) 928 Issue._add_read_property("repo", self, issue) 929 return issue
def
create_design_review( self, title: str, head: Union[Branch, str], base: Union[Branch, str], assignees: Optional[Set[Union[User, str]]] = None, body: Optional[str] = None, due_date: Optional[datetime.datetime] = None, milestone: Optional[Milestone] = None) -> DesignReview:
931 def create_design_review( 932 self, 933 title: str, 934 head: Union[Branch, str], 935 base: Union[Branch, str], 936 assignees: Optional[Set[Union[User, str]]] = None, 937 body: Optional[str] = None, 938 due_date: Optional[datetime] = None, 939 milestone: Optional["Milestone"] = None, 940 ) -> "DesignReview": 941 """ 942 Create a new Design Review. 943 944 See https://hub.allspice.io/api/swagger#/repository/repoCreatePullRequest 945 946 :param title: Title of the Design Review 947 :param head: Branch or name of the branch to merge into the base branch 948 :param base: Branch or name of the branch to merge into 949 :param assignees: Optional. A list of users to assign this review. List can be of 950 User objects or of usernames. 951 :param body: An Optional Description for the Design Review. 952 :param due_date: An Optional Due date for the Design Review. 953 :param milestone: An Optional Milestone for the Design Review 954 :return: The created Design Review 955 """ 956 957 data: dict[str, Any] = { 958 "title": title, 959 } 960 961 if isinstance(head, Branch): 962 data["head"] = head.name 963 else: 964 data["head"] = head 965 if isinstance(base, Branch): 966 data["base"] = base.name 967 else: 968 data["base"] = base 969 if assignees: 970 data["assignees"] = [a.username if isinstance(a, User) else a for a in assignees] 971 if body: 972 data["body"] = body 973 if due_date: 974 data["due_date"] = Util.format_time(due_date) 975 if milestone: 976 data["milestone"] = milestone.id 977 978 result = self.allspice_client.requests_post( 979 self.REPO_DESIGN_REVIEWS.format(owner=self.owner.username, repo=self.name), 980 data=data, 981 ) 982 983 return DesignReview.parse_response(self.allspice_client, result)
Create a new Design Review.
See allspice.allspice.io/api/swagger#/repository/repoCreatePullRequest">https://huballspice.allspice.io/api/swagger#/repository/repoCreatePullRequest
Parameters
- title: Title of the Design Review
- head: Branch or name of the branch to merge into the base branch
- base: Branch or name of the branch to merge into
- assignees: Optional. A list of users to assign this review. List can be of User objects or of usernames.
- body: An Optional Description for the Design Review.
- due_date: An Optional Due date for the Design Review.
- milestone: An Optional Milestone for the Design Review
Returns
The created Design Review
def
create_milestone( self, title: str, description: str, due_date: Optional[str] = None, state: str = 'open') -> Milestone:
985 def create_milestone( 986 self, 987 title: str, 988 description: str, 989 due_date: Optional[str] = None, 990 state: str = "open", 991 ) -> "Milestone": 992 url = Repository.REPO_MILESTONES.format(owner=self.owner.username, repo=self.name) 993 data = {"title": title, "description": description, "state": state} 994 if due_date: 995 data["due_date"] = due_date 996 result = self.allspice_client.requests_post(url, data=data) 997 return Milestone.parse_response(self.allspice_client, result)
def
create_gitea_hook(self, hook_url: str, events: List[str]):
999 def create_gitea_hook(self, hook_url: str, events: List[str]): 1000 url = f"/repos/{self.owner.username}/{self.name}/hooks" 1001 data = { 1002 "type": "gitea", 1003 "config": {"content_type": "json", "url": hook_url}, 1004 "events": events, 1005 "active": True, 1006 } 1007 return self.allspice_client.requests_post(url, data=data)
def
is_collaborator(self, username) -> bool:
1017 def is_collaborator(self, username) -> bool: 1018 if isinstance(username, User): 1019 username = username.username 1020 try: 1021 # returns 204 if its ok, 404 if its not 1022 self.allspice_client.requests_get( 1023 Repository.REPO_IS_COLLABORATOR % (self.owner.username, self.name, username) 1024 ) 1025 return True 1026 except Exception: 1027 return False
1029 def get_users_with_access(self) -> Sequence[User]: 1030 url = f"/repos/{self.owner.username}/{self.name}/collaborators" 1031 response = self.allspice_client.requests_get(url) 1032 collabs = [User.parse_response(self.allspice_client, user) for user in response] 1033 if isinstance(self.owner, User): 1034 return [*collabs, self.owner] 1035 else: 1036 # owner must be org 1037 teams = self.owner.get_teams() 1038 for team in teams: 1039 team_repos = team.get_repos() 1040 if self.name in [n.name for n in team_repos]: 1041 collabs += team.get_members() 1042 return collabs
def
transfer_ownership( self, new_owner: Union[User, Organization], new_teams: Union[Set[Team], FrozenSet[Team]] = frozenset()):
1048 def transfer_ownership( 1049 self, 1050 new_owner: Union[User, Organization], 1051 new_teams: Set[Team] | FrozenSet[Team] = frozenset(), 1052 ): 1053 url = Repository.REPO_TRANSFER.format(owner=self.owner.username, repo=self.name) 1054 data: dict[str, Any] = {"new_owner": new_owner.username} 1055 if isinstance(new_owner, Organization): 1056 new_team_ids = [team.id for team in new_teams if team in new_owner.get_teams()] 1057 data["team_ids"] = new_team_ids 1058 self.allspice_client.requests_post(url, data=data) 1059 # TODO: make sure this instance is either updated or discarded
def
get_git_content( self, ref: Union[Branch, Commit, str, NoneType] = None, commit: Optional[Commit] = None) -> List[Content]:
1061 def get_git_content( 1062 self, 1063 ref: Optional["Ref"] = None, 1064 commit: "Optional[Commit]" = None, 1065 ) -> List[Content]: 1066 """ 1067 Get the metadata for all files in the root directory. 1068 1069 https://hub.allspice.io/api/swagger#/repository/repoGetContentsList 1070 1071 :param ref: branch or commit to get content from 1072 :param commit: commit to get content from (deprecated) 1073 """ 1074 url = f"/repos/{self.owner.username}/{self.name}/contents" 1075 data = Util.data_params_for_ref(ref or commit) 1076 1077 result = [ 1078 Content.parse_response(self.allspice_client, f) 1079 for f in self.allspice_client.requests_get(url, data) 1080 ] 1081 return result
Get the metadata for all files in the root directory.
allspice.allspice.io/api/swagger#/repository/repoGetContentsList">https://huballspice.allspice.io/api/swagger#/repository/repoGetContentsList
Parameters
- ref: branch or commit to get content from
- commit: commit to get content from (deprecated)
def
get_tree( self, ref: Union[Branch, Commit, str, NoneType] = None, recursive: bool = False) -> List[GitEntry]:
1083 def get_tree(self, ref: Optional[Ref] = None, recursive: bool = False) -> List[GitEntry]: 1084 """ 1085 Get the repository's tree on a given ref. 1086 1087 By default, this will only return the top-level entries in the tree. If you want 1088 to get the entire tree, set `recursive` to True. 1089 1090 :param ref: The ref to get the tree from. If not provided, the default branch is used. 1091 :param recursive: Whether to get the entire tree or just the top-level entries. 1092 """ 1093 1094 ref = Util.data_params_for_ref(ref).get("ref", self.default_branch) 1095 url = self.REPO_GET_TREE.format(owner=self.owner.username, repo=self.name, ref=ref) 1096 params = {"recursive": recursive} 1097 results = self.allspice_client.requests_get_paginated(url, params=params) 1098 return [GitEntry.parse_response(self.allspice_client, result) for result in results]
Get the repository's tree on a given ref.
By default, this will only return the top-level entries in the tree. If you want
to get the entire tree, set recursive to True.
Parameters
- ref: The ref to get the tree from. If not provided, the default branch is used.
- recursive: Whether to get the entire tree or just the top-level entries.
def
get_file_content( self, content: Content, ref: Union[Branch, Commit, str, NoneType] = None, commit: Optional[Commit] = None) -> Union[str, List[Content]]:
1100 def get_file_content( 1101 self, 1102 content: Content, 1103 ref: Optional[Ref] = None, 1104 commit: Optional[Commit] = None, 1105 ) -> Union[str, List["Content"]]: 1106 """https://hub.allspice.io/api/swagger#/repository/repoGetContents""" 1107 url = f"/repos/{self.owner.username}/{self.name}/contents/{content.path}" 1108 data = Util.data_params_for_ref(ref or commit) 1109 1110 if content.type == Content.FILE: 1111 return self.allspice_client.requests_get(url, data)["content"] 1112 else: 1113 return [ 1114 Content.parse_response(self.allspice_client, f) 1115 for f in self.allspice_client.requests_get(url, data) 1116 ]
allspice.allspice.io/api/swagger#/repository/repoGetContents">https://huballspice.allspice.io/api/swagger#/repository/repoGetContents
1118 def get_raw_file( 1119 self, 1120 file_path: str, 1121 ref: Optional[Ref] = None, 1122 ) -> bytes: 1123 """ 1124 Get the raw, binary data of a single file. 1125 1126 Note 1: if the file you are requesting is a text file, you might want to 1127 use .decode() on the result to get a string. For example: 1128 1129 content = repo.get_raw_file("file.txt").decode("utf-8") 1130 1131 Note 2: this method will store the entire file in memory. If you want 1132 to download a large file, you might want to use `download_to_file` 1133 instead. 1134 1135 See https://hub.allspice.io/api/swagger#/repository/repoGetRawFileOrLFS 1136 1137 :param file_path: The path to the file to get. 1138 :param ref: The branch or commit to get the file from. If not provided, 1139 the default branch is used. 1140 """ 1141 1142 url = self.REPO_GET_MEDIA.format( 1143 owner=self.owner.username, 1144 repo=self.name, 1145 path=file_path, 1146 ) 1147 params = Util.data_params_for_ref(ref) 1148 return self.allspice_client.requests_get_raw(url, params=params)
Get the raw, binary data of a single file.
Note 1: if the file you are requesting is a text file, you might want to use .decode() on the result to get a string. For example:
content = repo.get_raw_file("file.txt").decode("utf-8")
Note 2: this method will store the entire file in memory. If you want
to download a large file, you might want to use download_to_file
instead.
See allspice.allspice.io/api/swagger#/repository/repoGetRawFileOrLFS">https://huballspice.allspice.io/api/swagger#/repository/repoGetRawFileOrLFS
Parameters
- file_path: The path to the file to get.
- ref: The branch or commit to get the file from. If not provided, the default branch is used.
def
download_to_file( self, file_path: str, io: <class 'IO'>, ref: Union[Branch, Commit, str, NoneType] = None) -> None:
1150 def download_to_file( 1151 self, 1152 file_path: str, 1153 io: IO, 1154 ref: Optional[Ref] = None, 1155 ) -> None: 1156 """ 1157 Download the binary data of a file to a file-like object. 1158 1159 Example: 1160 1161 with open("schematic.DSN", "wb") as f: 1162 Repository.download_to_file("Schematics/my_schematic.DSN", f) 1163 1164 :param file_path: The path to the file in the repository from the root 1165 of the repository. 1166 :param io: The file-like object to write the data to. 1167 """ 1168 1169 url = self.allspice_client._AllSpice__get_url( 1170 self.REPO_GET_MEDIA.format( 1171 owner=self.owner.username, 1172 repo=self.name, 1173 path=file_path, 1174 ) 1175 ) 1176 params = Util.data_params_for_ref(ref) 1177 response = self.allspice_client.requests.get( 1178 url, 1179 params=params, 1180 headers=self.allspice_client.headers, 1181 stream=True, 1182 ) 1183 1184 for chunk in response.iter_content(chunk_size=4096): 1185 if chunk: 1186 io.write(chunk)
Download the binary data of a file to a file-like object.
Example:
with open("schematic.DSN", "wb") as f:
Repository.download_to_file("Schematics/my_schematic.DSN", f)
Parameters
- file_path: The path to the file in the repository from the root of the repository.
- io: The file-like object to write the data to.
def
get_generated_json( self, content: Union[Content, str], ref: Union[Branch, Commit, str, NoneType] = None) -> dict:
1188 def get_generated_json(self, content: Union[Content, str], ref: Optional[Ref] = None) -> dict: 1189 """ 1190 Get the json blob for a cad file if it exists, otherwise enqueue 1191 a new job and return a 503 status. 1192 1193 WARNING: This is still experimental and not recommended for critical 1194 applications. The structure and content of the returned dictionary can 1195 change at any time. 1196 1197 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceJSON 1198 """ 1199 1200 if isinstance(content, Content): 1201 content = content.path 1202 1203 url = self.REPO_GET_ALLSPICE_JSON.format( 1204 owner=self.owner.username, 1205 repo=self.name, 1206 content=content, 1207 ) 1208 data = Util.data_params_for_ref(ref) 1209 return self.allspice_client.requests_get(url, data)
Get the json blob for a cad file if it exists, otherwise enqueue a new job and return a 503 status.
WARNING: This is still experimental and not recommended for critical applications. The structure and content of the returned dictionary can change at any time.
See allspice.allspice.io/api/swagger#/repository/repoGetAllSpiceJSON">https://huballspice.allspice.io/api/swagger#/repository/repoGetAllSpiceJSON
def
get_generated_svg( self, content: Union[Content, str], ref: Union[Branch, Commit, str, NoneType] = None) -> bytes:
1211 def get_generated_svg(self, content: Union[Content, str], ref: Optional[Ref] = None) -> bytes: 1212 """ 1213 Get the svg blob for a cad file if it exists, otherwise enqueue 1214 a new job and return a 503 status. 1215 1216 WARNING: This is still experimental and not yet recommended for 1217 critical applications. The content of the returned svg can change 1218 at any time. 1219 1220 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceSVG 1221 """ 1222 1223 if isinstance(content, Content): 1224 content = content.path 1225 1226 url = self.REPO_GET_ALLSPICE_SVG.format( 1227 owner=self.owner.username, 1228 repo=self.name, 1229 content=content, 1230 ) 1231 data = Util.data_params_for_ref(ref) 1232 return self.allspice_client.requests_get_raw(url, data)
Get the svg blob for a cad file if it exists, otherwise enqueue a new job and return a 503 status.
WARNING: This is still experimental and not yet recommended for critical applications. The content of the returned svg can change at any time.
See allspice.allspice.io/api/swagger#/repository/repoGetAllSpiceSVG">https://huballspice.allspice.io/api/swagger#/repository/repoGetAllSpiceSVG
def
get_generated_projectdata( self, content: Union[Content, str], ref: Union[Branch, Commit, str, NoneType] = None) -> dict:
1234 def get_generated_projectdata( 1235 self, content: Union[Content, str], ref: Optional[Ref] = None 1236 ) -> dict: 1237 """ 1238 Get the json project data based on the cad file provided 1239 1240 WARNING: This is still experimental and not yet recommended for 1241 critical applications. The content of the returned dictionary can change 1242 at any time. 1243 1244 See https://hub.allspice.io/api/swagger#/repository/repoGetAllSpiceProject 1245 """ 1246 if isinstance(content, Content): 1247 content = content.path 1248 1249 url = self.REPO_GET_ALLSPICE_PROJECT.format( 1250 owner=self.owner.username, 1251 repo=self.name, 1252 content=content, 1253 ) 1254 data = Util.data_params_for_ref(ref) 1255 return self.allspice_client.requests_get(url, data)
Get the json project data based on the cad file provided
WARNING: This is still experimental and not yet recommended for critical applications. The content of the returned dictionary can change at any time.
See allspice.allspice.io/api/swagger#/repository/repoGetAllSpiceProject">https://huballspice.allspice.io/api/swagger#/repository/repoGetAllSpiceProject
def
create_file(self, file_path: str, content: str, data: Optional[dict] = None):
1257 def create_file(self, file_path: str, content: str, data: Optional[dict] = None): 1258 """https://hub.allspice.io/api/swagger#/repository/repoCreateFile""" 1259 if not data: 1260 data = {} 1261 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1262 data.update({"content": content}) 1263 return self.allspice_client.requests_post(url, data)
allspice.allspice.io/api/swagger#/repository/repoCreateFile">https://huballspice.allspice.io/api/swagger#/repository/repoCreateFile
def
change_file( self, file_path: str, file_sha: str, content: str, data: Optional[dict] = None):
1265 def change_file(self, file_path: str, file_sha: str, content: str, data: Optional[dict] = None): 1266 """https://hub.allspice.io/api/swagger#/repository/repoCreateFile""" 1267 if not data: 1268 data = {} 1269 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1270 data.update({"sha": file_sha, "content": content}) 1271 return self.allspice_client.requests_put(url, data)
allspice.allspice.io/api/swagger#/repository/repoCreateFile">https://huballspice.allspice.io/api/swagger#/repository/repoCreateFile
def
delete_file(self, file_path: str, file_sha: str, data: Optional[dict] = None):
1273 def delete_file(self, file_path: str, file_sha: str, data: Optional[dict] = None): 1274 """https://hub.allspice.io/api/swagger#/repository/repoDeleteFile""" 1275 if not data: 1276 data = {} 1277 url = f"/repos/{self.owner.username}/{self.name}/contents/{file_path}" 1278 data.update({"sha": file_sha}) 1279 return self.allspice_client.requests_delete(url, data)
allspice.allspice.io/api/swagger#/repository/repoDeleteFile">https://huballspice.allspice.io/api/swagger#/repository/repoDeleteFile
def
get_archive( self, ref: Union[Branch, Commit, str] = 'main', archive_format: Repository.ArchiveFormat = <ArchiveFormat.ZIP: 'zip'>) -> bytes:
1281 def get_archive( 1282 self, 1283 ref: Ref = "main", 1284 archive_format: ArchiveFormat = ArchiveFormat.ZIP, 1285 ) -> bytes: 1286 """ 1287 Download all the files in a specific ref of a repository as a zip or tarball 1288 archive. 1289 1290 https://hub.allspice.io/api/swagger#/repository/repoGetArchive 1291 1292 :param ref: branch or commit to get content from, defaults to the "main" branch 1293 :param archive_format: zip or tar, defaults to zip 1294 """ 1295 1296 ref_string = Util.data_params_for_ref(ref)["ref"] 1297 url = self.REPO_GET_ARCHIVE.format( 1298 owner=self.owner.username, 1299 repo=self.name, 1300 ref=ref_string, 1301 format=archive_format.value, 1302 ) 1303 return self.allspice_client.requests_get_raw(url)
Download all the files in a specific ref of a repository as a zip or tarball archive.
allspice.allspice.io/api/swagger#/repository/repoGetArchive">https://huballspice.allspice.io/api/swagger#/repository/repoGetArchive
Parameters
- ref: branch or commit to get content from, defaults to the "main" branch
- archive_format: zip or tar, defaults to zip
def
get_topics(self) -> list[str]:
1305 def get_topics(self) -> list[str]: 1306 """ 1307 Gets the list of topics on this repository. 1308 1309 See http://localhost:3000/api/swagger#/repository/repoListTopics 1310 """ 1311 1312 url = self.REPO_GET_TOPICS.format( 1313 owner=self.owner.username, 1314 repo=self.name, 1315 ) 1316 return self.allspice_client.requests_get(url)["topics"]
Gets the list of topics on this repository.
See http://localhost:3000/api/swagger#/repository/repoListTopics
def
add_topic(self, topic: str):
1318 def add_topic(self, topic: str): 1319 """ 1320 Adds a topic to the repository. 1321 1322 See https://hub.allspice.io/api/swagger#/repository/repoAddTopic 1323 1324 :param topic: The topic to add. Topic names must consist only of 1325 lowercase letters, numnbers and dashes (-), and cannot start with 1326 dashes. Topic names also must be under 35 characters long. 1327 """ 1328 1329 url = self.REPO_ADD_TOPIC.format(owner=self.owner.username, repo=self.name, topic=topic) 1330 self.allspice_client.requests_put(url)
Adds a topic to the repository.
See allspice.allspice.io/api/swagger#/repository/repoAddTopic">https://huballspice.allspice.io/api/swagger#/repository/repoAddTopic
Parameters
- topic: The topic to add. Topic names must consist only of lowercase letters, numnbers and dashes (-), and cannot start with dashes. Topic names also must be under 35 characters long.
def
create_release( self, tag_name: str, name: Optional[str] = None, body: Optional[str] = None, draft: bool = False):
1332 def create_release( 1333 self, 1334 tag_name: str, 1335 name: Optional[str] = None, 1336 body: Optional[str] = None, 1337 draft: bool = False, 1338 ): 1339 """ 1340 Create a release for this repository. The release will be created for 1341 the tag with the given name. If there is no tag with this name, create 1342 the tag first. 1343 1344 See https://hub.allspice.io/api/swagger#/repository/repoCreateRelease 1345 """ 1346 1347 url = self.REPO_GET_RELEASES.format(owner=self.owner.username, repo=self.name) 1348 data = { 1349 "tag_name": tag_name, 1350 "draft": draft, 1351 } 1352 if name is not None: 1353 data["name"] = name 1354 if body is not None: 1355 data["body"] = body 1356 response = self.allspice_client.requests_post(url, data) 1357 return Release.parse_response(self.allspice_client, response, self)
Create a release for this repository. The release will be created for the tag with the given name. If there is no tag with this name, create the tag first.
See allspice.allspice.io/api/swagger#/repository/repoCreateRelease">https://huballspice.allspice.io/api/swagger#/repository/repoCreateRelease
def
get_releases( self, draft: Optional[bool] = None, pre_release: Optional[bool] = None) -> List[Release]:
1359 def get_releases( 1360 self, draft: Optional[bool] = None, pre_release: Optional[bool] = None 1361 ) -> List[Release]: 1362 """ 1363 Get the list of releases for this repository. 1364 1365 See https://hub.allspice.io/api/swagger#/repository/repoListReleases 1366 """ 1367 1368 data = {} 1369 1370 if draft is not None: 1371 data["draft"] = draft 1372 if pre_release is not None: 1373 data["pre-release"] = pre_release 1374 1375 url = self.REPO_GET_RELEASES.format(owner=self.owner.username, repo=self.name) 1376 responses = self.allspice_client.requests_get_paginated(url, params=data) 1377 1378 return [ 1379 Release.parse_response(self.allspice_client, response, self) for response in responses 1380 ]
Get the list of releases for this repository.
See allspice.allspice.io/api/swagger#/repository/repoListReleases">https://huballspice.allspice.io/api/swagger#/repository/repoListReleases
1382 def get_latest_release(self) -> Release: 1383 """ 1384 Get the latest release for this repository. 1385 1386 See https://hub.allspice.io/api/swagger#/repository/repoGetLatestRelease 1387 """ 1388 1389 url = self.REPO_GET_LATEST_RELEASE.format(owner=self.owner.username, repo=self.name) 1390 response = self.allspice_client.requests_get(url) 1391 release = Release.parse_response(self.allspice_client, response, self) 1392 return release
Get the latest release for this repository.
See allspice.allspice.io/api/swagger#/repository/repoGetLatestRelease">https://huballspice.allspice.io/api/swagger#/repository/repoGetLatestRelease
1394 def get_release_by_tag(self, tag: str) -> Release: 1395 """ 1396 Get a release by its tag. 1397 1398 See https://hub.allspice.io/api/swagger#/repository/repoGetReleaseByTag 1399 """ 1400 1401 url = self.REPO_GET_RELEASE_BY_TAG.format( 1402 owner=self.owner.username, repo=self.name, tag=tag 1403 ) 1404 response = self.allspice_client.requests_get(url) 1405 release = Release.parse_response(self.allspice_client, response, self) 1406 return release
Get a release by its tag.
See allspice.allspice.io/api/swagger#/repository/repoGetReleaseByTag">https://huballspice.allspice.io/api/swagger#/repository/repoGetReleaseByTag
def
get_commit_statuses( self, commit: Union[str, Commit], sort: Optional[Repository.CommitStatusSort] = None, state: Optional[CommitStatusState] = None) -> List[CommitStatus]:
1408 def get_commit_statuses( 1409 self, 1410 commit: Union[str, Commit], 1411 sort: Optional[CommitStatusSort] = None, 1412 state: Optional[CommitStatusState] = None, 1413 ) -> List[CommitStatus]: 1414 """ 1415 Get a list of statuses for a commit. 1416 1417 This is roughly equivalent to the Commit.get_statuses method, but this 1418 method allows you to sort and filter commits and is more convenient if 1419 you have a commit SHA and don't need to get the commit itself. 1420 1421 See https://hub.allspice.io/api/swagger#/repository/repoListStatuses 1422 """ 1423 1424 if isinstance(commit, Commit): 1425 commit = commit.sha 1426 1427 params = {} 1428 if sort is not None: 1429 params["sort"] = sort.value 1430 if state is not None: 1431 params["state"] = state.value 1432 1433 url = self.REPO_GET_COMMIT_STATUS.format( 1434 owner=self.owner.username, repo=self.name, sha=commit 1435 ) 1436 response = self.allspice_client.requests_get_paginated(url, params=params) 1437 return [CommitStatus.parse_response(self.allspice_client, status) for status in response]
Get a list of statuses for a commit.
This is roughly equivalent to the Commit.get_statuses method, but this method allows you to sort and filter commits and is more convenient if you have a commit SHA and don't need to get the commit itself.
See allspice.allspice.io/api/swagger#/repository/repoListStatuses">https://huballspice.allspice.io/api/swagger#/repository/repoListStatuses
def
create_commit_status( self, commit: Union[str, Commit], context: Optional[str] = None, description: Optional[str] = None, state: Optional[CommitStatusState] = None, target_url: Optional[str] = None) -> CommitStatus:
1439 def create_commit_status( 1440 self, 1441 commit: Union[str, Commit], 1442 context: Optional[str] = None, 1443 description: Optional[str] = None, 1444 state: Optional[CommitStatusState] = None, 1445 target_url: Optional[str] = None, 1446 ) -> CommitStatus: 1447 """ 1448 Create a status on a commit. 1449 1450 See https://hub.allspice.io/api/swagger#/repository/repoCreateStatus 1451 """ 1452 1453 if isinstance(commit, Commit): 1454 commit = commit.sha 1455 1456 data = {} 1457 if context is not None: 1458 data["context"] = context 1459 if description is not None: 1460 data["description"] = description 1461 if state is not None: 1462 data["state"] = state.value 1463 if target_url is not None: 1464 data["target_url"] = target_url 1465 1466 url = self.REPO_GET_COMMIT_STATUS.format( 1467 owner=self.owner.username, repo=self.name, sha=commit 1468 ) 1469 response = self.allspice_client.requests_post(url, data=data) 1470 return CommitStatus.parse_response(self.allspice_client, response)
Create a status on a commit.
See allspice.allspice.io/api/swagger#/repository/repoCreateStatus">https://huballspice.allspice.io/api/swagger#/repository/repoCreateStatus
class
Repository.ArchiveFormat(enum.Enum):
573 class ArchiveFormat(Enum): 574 """ 575 Archive formats for Repository.get_archive 576 """ 577 578 TAR = "tar.gz" 579 ZIP = "zip"
Archive formats for Repository.get_archive
class
Repository.CommitStatusSort(enum.Enum):
581 class CommitStatusSort(Enum): 582 """ 583 Sort order for Repository.get_commit_status 584 """ 585 586 OLDEST = "oldest" 587 RECENT_UPDATE = "recentupdate" 588 LEAST_UPDATE = "leastupdate" 589 LEAST_INDEX = "leastindex" 590 HIGHEST_INDEX = "highestindex"
Sort order for Repository.get_commit_status
1479class Milestone(ApiObject): 1480 allow_merge_commits: Any 1481 allow_rebase: Any 1482 allow_rebase_explicit: Any 1483 allow_squash_merge: Any 1484 archived: Any 1485 closed_at: Any 1486 closed_issues: int 1487 created_at: str 1488 default_branch: Any 1489 description: str 1490 due_on: Any 1491 has_issues: Any 1492 has_pull_requests: Any 1493 has_wiki: Any 1494 id: int 1495 ignore_whitespace_conflicts: Any 1496 name: Any 1497 open_issues: int 1498 private: Any 1499 state: str 1500 title: str 1501 updated_at: str 1502 website: Any 1503 1504 API_OBJECT = """/repos/{owner}/{repo}/milestones/{number}""" # <owner, repo> 1505 1506 def __init__(self, allspice_client): 1507 super().__init__(allspice_client) 1508 1509 def __eq__(self, other): 1510 if not isinstance(other, Milestone): 1511 return False 1512 return self.allspice_client == other.allspice_client and self.id == other.id 1513 1514 def __hash__(self): 1515 return hash(self.allspice_client) ^ hash(self.id) 1516 1517 _fields_to_parsers: ClassVar[dict] = { 1518 "closed_at": lambda _, t: Util.convert_time(t), 1519 "due_on": lambda _, t: Util.convert_time(t), 1520 } 1521 1522 _patchable_fields: ClassVar[set[str]] = { 1523 "allow_merge_commits", 1524 "allow_rebase", 1525 "allow_rebase_explicit", 1526 "allow_squash_merge", 1527 "archived", 1528 "default_branch", 1529 "description", 1530 "has_issues", 1531 "has_pull_requests", 1532 "has_wiki", 1533 "ignore_whitespace_conflicts", 1534 "name", 1535 "private", 1536 "website", 1537 } 1538 1539 @classmethod 1540 def request(cls, allspice_client, owner: str, repo: str, number: str): 1541 return cls._request(allspice_client, {"owner": owner, "repo": repo, "number": number})
1544class Attachment(ReadonlyApiObject): 1545 """ 1546 An asset attached to a comment. 1547 1548 You cannot edit or delete the attachment from this object - see the instance methods 1549 Comment.edit_attachment and delete_attachment for that. 1550 """ 1551 1552 browser_download_url: str 1553 created_at: str 1554 download_count: int 1555 id: int 1556 name: str 1557 size: int 1558 uuid: str 1559 1560 def __init__(self, allspice_client): 1561 super().__init__(allspice_client) 1562 1563 def __eq__(self, other): 1564 if not isinstance(other, Attachment): 1565 return False 1566 1567 return self.uuid == other.uuid 1568 1569 def __hash__(self): 1570 return hash(self.uuid) 1571 1572 def download_to_file(self, io: IO): 1573 """ 1574 Download the raw, binary data of this Attachment to a file-like object. 1575 1576 Example: 1577 1578 with open("my_file.zip", "wb") as f: 1579 attachment.download_to_file(f) 1580 1581 :param io: The file-like object to write the data to. 1582 """ 1583 1584 response = self.allspice_client.requests.get( 1585 self.browser_download_url, 1586 headers=self.allspice_client.headers, 1587 stream=True, 1588 ) 1589 # 4kb chunks 1590 for chunk in response.iter_content(chunk_size=4096): 1591 if chunk: 1592 io.write(chunk)
An asset attached to a comment.
You cannot edit or delete the attachment from this object - see the instance methods Comment.edit_attachment and delete_attachment for that.
def
download_to_file(self, io: <class 'IO'>):
1572 def download_to_file(self, io: IO): 1573 """ 1574 Download the raw, binary data of this Attachment to a file-like object. 1575 1576 Example: 1577 1578 with open("my_file.zip", "wb") as f: 1579 attachment.download_to_file(f) 1580 1581 :param io: The file-like object to write the data to. 1582 """ 1583 1584 response = self.allspice_client.requests.get( 1585 self.browser_download_url, 1586 headers=self.allspice_client.headers, 1587 stream=True, 1588 ) 1589 # 4kb chunks 1590 for chunk in response.iter_content(chunk_size=4096): 1591 if chunk: 1592 io.write(chunk)
Download the raw, binary data of this Attachment to a file-like object.
Example:
with open("my_file.zip", "wb") as f:
attachment.download_to_file(f)
Parameters
- io: The file-like object to write the data to.
Inherited Members
1595class Comment(ApiObject): 1596 assets: List[Union[Any, Dict[str, Union[int, str]]]] 1597 body: str 1598 created_at: datetime 1599 html_url: str 1600 id: int 1601 issue_url: str 1602 original_author: str 1603 original_author_id: int 1604 pull_request_url: str 1605 updated_at: datetime 1606 user: User 1607 1608 API_OBJECT = """/repos/{owner}/{repo}/issues/comments/{id}""" 1609 GET_ATTACHMENTS_PATH = """/repos/{owner}/{repo}/issues/comments/{id}/assets""" 1610 ATTACHMENT_PATH = """/repos/{owner}/{repo}/issues/comments/{id}/assets/{attachment_id}""" 1611 1612 def __init__(self, allspice_client): 1613 super().__init__(allspice_client) 1614 1615 def __eq__(self, other): 1616 if not isinstance(other, Comment): 1617 return False 1618 return self.repository == other.repository and self.id == other.id 1619 1620 def __hash__(self): 1621 return hash(self.repository) ^ hash(self.id) 1622 1623 @classmethod 1624 def request(cls, allspice_client, owner: str, repo: str, id: str) -> "Comment": 1625 return cls._request(allspice_client, {"owner": owner, "repo": repo, "id": id}) 1626 1627 _fields_to_parsers: ClassVar[dict] = { 1628 "user": lambda allspice_client, r: User.parse_response(allspice_client, r), 1629 "created_at": lambda _, t: Util.convert_time(t), 1630 "updated_at": lambda _, t: Util.convert_time(t), 1631 } 1632 1633 _patchable_fields: ClassVar[set[str]] = {"body"} 1634 1635 @property 1636 def parent_url(self) -> str: 1637 """URL of the parent of this comment (the issue or the pull request)""" 1638 1639 if self.issue_url is not None and self.issue_url != "": 1640 return self.issue_url 1641 else: 1642 return self.pull_request_url 1643 1644 @cached_property 1645 def repository(self) -> Repository: 1646 """The repository this comment was posted on.""" 1647 1648 owner_name, repo_name = self.parent_url.split("/")[-4:-2] 1649 return Repository.request(self.allspice_client, owner_name, repo_name) 1650 1651 def __fields_for_path(self): 1652 return { 1653 "owner": self.repository.owner.username, 1654 "repo": self.repository.name, 1655 "id": self.id, 1656 } 1657 1658 def commit(self): 1659 self._commit(self.__fields_for_path()) 1660 1661 def delete(self): 1662 self.allspice_client.requests_delete(self.API_OBJECT.format(**self.__fields_for_path())) 1663 self.deleted = True 1664 1665 def get_attachments(self) -> List[Attachment]: 1666 """ 1667 Get all attachments on this comment. This returns Attachment objects, which 1668 contain a link to download the attachment. 1669 1670 https://hub.allspice.io/api/swagger#/issue/issueListIssueCommentAttachments 1671 """ 1672 1673 results = self.allspice_client.requests_get( 1674 self.GET_ATTACHMENTS_PATH.format(**self.__fields_for_path()) 1675 ) 1676 return [Attachment.parse_response(self.allspice_client, result) for result in results] 1677 1678 def create_attachment(self, file: IO, name: Optional[str] = None) -> Attachment: 1679 """ 1680 Create an attachment on this comment. 1681 1682 https://hub.allspice.io/api/swagger#/issue/issueCreateIssueCommentAttachment 1683 1684 :param file: The file to attach. This should be a file-like object. 1685 :param name: The name of the file. If not provided, the name of the file will be 1686 used. 1687 :return: The created attachment. 1688 """ 1689 1690 args: dict[str, Any] = { 1691 "files": {"attachment": file}, 1692 } 1693 if name is not None: 1694 args["params"] = {"name": name} 1695 1696 result = self.allspice_client.requests_post( 1697 self.GET_ATTACHMENTS_PATH.format(**self.__fields_for_path()), 1698 **args, 1699 ) 1700 return Attachment.parse_response(self.allspice_client, result) 1701 1702 def edit_attachment(self, attachment: Attachment, data: dict) -> Attachment: 1703 """ 1704 Edit an attachment. 1705 1706 The list of params that can be edited is available at 1707 https://hub.allspice.io/api/swagger#/issue/issueEditIssueCommentAttachment 1708 1709 :param attachment: The attachment to be edited 1710 :param data: The data parameter should be a dictionary of the fields to edit. 1711 :return: The edited attachment 1712 """ 1713 1714 args = { 1715 **self.__fields_for_path(), 1716 "attachment_id": attachment.id, 1717 } 1718 result = self.allspice_client.requests_patch( 1719 self.ATTACHMENT_PATH.format(**args), 1720 data=data, 1721 ) 1722 return Attachment.parse_response(self.allspice_client, result) 1723 1724 def delete_attachment(self, attachment: Attachment): 1725 """https://hub.allspice.io/api/swagger#/issue/issueDeleteIssueCommentAttachment""" 1726 1727 args = { 1728 **self.__fields_for_path(), 1729 "attachment_id": attachment.id, 1730 } 1731 self.allspice_client.requests_delete(self.ATTACHMENT_PATH.format(**args)) 1732 attachment.deleted = True
user: User
parent_url: str
1635 @property 1636 def parent_url(self) -> str: 1637 """URL of the parent of this comment (the issue or the pull request)""" 1638 1639 if self.issue_url is not None and self.issue_url != "": 1640 return self.issue_url 1641 else: 1642 return self.pull_request_url
URL of the parent of this comment (the issue or the pull request)
repository: Repository
1644 @cached_property 1645 def repository(self) -> Repository: 1646 """The repository this comment was posted on.""" 1647 1648 owner_name, repo_name = self.parent_url.split("/")[-4:-2] 1649 return Repository.request(self.allspice_client, owner_name, repo_name)
The repository this comment was posted on.
1665 def get_attachments(self) -> List[Attachment]: 1666 """ 1667 Get all attachments on this comment. This returns Attachment objects, which 1668 contain a link to download the attachment. 1669 1670 https://hub.allspice.io/api/swagger#/issue/issueListIssueCommentAttachments 1671 """ 1672 1673 results = self.allspice_client.requests_get( 1674 self.GET_ATTACHMENTS_PATH.format(**self.__fields_for_path()) 1675 ) 1676 return [Attachment.parse_response(self.allspice_client, result) for result in results]
Get all attachments on this comment. This returns Attachment objects, which contain a link to download the attachment.
allspice.allspice.io/api/swagger#/issue/issueListIssueCommentAttachments">https://huballspice.allspice.io/api/swagger#/issue/issueListIssueCommentAttachments
1678 def create_attachment(self, file: IO, name: Optional[str] = None) -> Attachment: 1679 """ 1680 Create an attachment on this comment. 1681 1682 https://hub.allspice.io/api/swagger#/issue/issueCreateIssueCommentAttachment 1683 1684 :param file: The file to attach. This should be a file-like object. 1685 :param name: The name of the file. If not provided, the name of the file will be 1686 used. 1687 :return: The created attachment. 1688 """ 1689 1690 args: dict[str, Any] = { 1691 "files": {"attachment": file}, 1692 } 1693 if name is not None: 1694 args["params"] = {"name": name} 1695 1696 result = self.allspice_client.requests_post( 1697 self.GET_ATTACHMENTS_PATH.format(**self.__fields_for_path()), 1698 **args, 1699 ) 1700 return Attachment.parse_response(self.allspice_client, result)
Create an attachment on this comment.
allspice.allspice.io/api/swagger#/issue/issueCreateIssueCommentAttachment">https://huballspice.allspice.io/api/swagger#/issue/issueCreateIssueCommentAttachment
Parameters
- file: The file to attach. This should be a file-like object.
- name: The name of the file. If not provided, the name of the file will be used.
Returns
The created attachment.
1702 def edit_attachment(self, attachment: Attachment, data: dict) -> Attachment: 1703 """ 1704 Edit an attachment. 1705 1706 The list of params that can be edited is available at 1707 https://hub.allspice.io/api/swagger#/issue/issueEditIssueCommentAttachment 1708 1709 :param attachment: The attachment to be edited 1710 :param data: The data parameter should be a dictionary of the fields to edit. 1711 :return: The edited attachment 1712 """ 1713 1714 args = { 1715 **self.__fields_for_path(), 1716 "attachment_id": attachment.id, 1717 } 1718 result = self.allspice_client.requests_patch( 1719 self.ATTACHMENT_PATH.format(**args), 1720 data=data, 1721 ) 1722 return Attachment.parse_response(self.allspice_client, result)
Edit an attachment.
The list of params that can be edited is available at allspice.allspice.io/api/swagger#/issue/issueEditIssueCommentAttachment">https://huballspice.allspice.io/api/swagger#/issue/issueEditIssueCommentAttachment
Parameters
- attachment: The attachment to be edited
- data: The data parameter should be a dictionary of the fields to edit.
Returns
The edited attachment
1724 def delete_attachment(self, attachment: Attachment): 1725 """https://hub.allspice.io/api/swagger#/issue/issueDeleteIssueCommentAttachment""" 1726 1727 args = { 1728 **self.__fields_for_path(), 1729 "attachment_id": attachment.id, 1730 } 1731 self.allspice_client.requests_delete(self.ATTACHMENT_PATH.format(**args)) 1732 attachment.deleted = True
allspice.allspice.io/api/swagger#/issue/issueDeleteIssueCommentAttachment">https://huballspice.allspice.io/api/swagger#/issue/issueDeleteIssueCommentAttachment
1735class Commit(ReadonlyApiObject): 1736 author: User 1737 commit: Dict[str, Union[str, Dict[str, str], Dict[str, Optional[Union[bool, str]]]]] 1738 committer: Dict[str, Union[int, str, bool]] 1739 created: str 1740 files: List[Dict[str, str]] 1741 html_url: str 1742 inner_commit: Dict[str, Union[str, Dict[str, str], Dict[str, Optional[Union[bool, str]]]]] 1743 parents: List[Union[Dict[str, str], Any]] 1744 sha: str 1745 stats: Dict[str, int] 1746 url: str 1747 1748 API_OBJECT = """/repos/{owner}/{repo}/commits/{sha}""" 1749 COMMIT_GET_STATUS = """/repos/{owner}/{repo}/commits/{sha}/status""" 1750 COMMIT_GET_STATUSES = """/repos/{owner}/{repo}/commits/{sha}/statuses""" 1751 1752 # Regex to extract owner and repo names from the url property 1753 URL_REGEXP = re.compile(r"/repos/([^/]+)/([^/]+)/git/commits") 1754 1755 def __init__(self, allspice_client): 1756 super().__init__(allspice_client) 1757 1758 _fields_to_parsers: ClassVar[dict] = { 1759 # NOTE: api may return None for commiters that are no allspice users 1760 "author": lambda allspice_client, u: User.parse_response(allspice_client, u) if u else None 1761 } 1762 1763 def __eq__(self, other): 1764 if not isinstance(other, Commit): 1765 return False 1766 return self.sha == other.sha 1767 1768 def __hash__(self): 1769 return hash(self.sha) 1770 1771 @classmethod 1772 def parse_response(cls, allspice_client, result) -> "Commit": 1773 commit_cache = result["commit"] 1774 api_object = cls(allspice_client) 1775 cls._initialize(allspice_client, api_object, result) 1776 # inner_commit for legacy reasons 1777 Commit._add_read_property("inner_commit", commit_cache, api_object) 1778 return api_object 1779 1780 def get_status(self) -> CommitCombinedStatus: 1781 """ 1782 Get a combined status consisting of all statues on this commit. 1783 1784 Note that the returned object is a CommitCombinedStatus object, which 1785 also contains a list of all statuses on the commit. 1786 1787 https://hub.allspice.io/api/swagger#/repository/repoGetCommitStatus 1788 """ 1789 1790 result = self.allspice_client.requests_get( 1791 self.COMMIT_GET_STATUS.format(**self._fields_for_path) 1792 ) 1793 return CommitCombinedStatus.parse_response(self.allspice_client, result) 1794 1795 def get_statuses(self) -> List[CommitStatus]: 1796 """ 1797 Get a list of all statuses on this commit. 1798 1799 https://hub.allspice.io/api/swagger#/repository/repoListCommitStatuses 1800 """ 1801 1802 results = self.allspice_client.requests_get( 1803 self.COMMIT_GET_STATUSES.format(**self._fields_for_path) 1804 ) 1805 return [CommitStatus.parse_response(self.allspice_client, result) for result in results] 1806 1807 @cached_property 1808 def _fields_for_path(self) -> dict[str, str]: 1809 matches = self.URL_REGEXP.search(self.url) 1810 if not matches: 1811 raise ValueError(f"Invalid commit URL: {self.url}") 1812 1813 return { 1814 "owner": matches.group(1), 1815 "repo": matches.group(2), 1816 "sha": self.sha, 1817 }
1771 @classmethod 1772 def parse_response(cls, allspice_client, result) -> "Commit": 1773 commit_cache = result["commit"] 1774 api_object = cls(allspice_client) 1775 cls._initialize(allspice_client, api_object, result) 1776 # inner_commit for legacy reasons 1777 Commit._add_read_property("inner_commit", commit_cache, api_object) 1778 return api_object
1780 def get_status(self) -> CommitCombinedStatus: 1781 """ 1782 Get a combined status consisting of all statues on this commit. 1783 1784 Note that the returned object is a CommitCombinedStatus object, which 1785 also contains a list of all statuses on the commit. 1786 1787 https://hub.allspice.io/api/swagger#/repository/repoGetCommitStatus 1788 """ 1789 1790 result = self.allspice_client.requests_get( 1791 self.COMMIT_GET_STATUS.format(**self._fields_for_path) 1792 ) 1793 return CommitCombinedStatus.parse_response(self.allspice_client, result)
Get a combined status consisting of all statues on this commit.
Note that the returned object is a CommitCombinedStatus object, which also contains a list of all statuses on the commit.
allspice.allspice.io/api/swagger#/repository/repoGetCommitStatus">https://huballspice.allspice.io/api/swagger#/repository/repoGetCommitStatus
1795 def get_statuses(self) -> List[CommitStatus]: 1796 """ 1797 Get a list of all statuses on this commit. 1798 1799 https://hub.allspice.io/api/swagger#/repository/repoListCommitStatuses 1800 """ 1801 1802 results = self.allspice_client.requests_get( 1803 self.COMMIT_GET_STATUSES.format(**self._fields_for_path) 1804 ) 1805 return [CommitStatus.parse_response(self.allspice_client, result) for result in results]
Get a list of all statuses on this commit.
allspice.allspice.io/api/swagger#/repository/repoListCommitStatuses">https://huballspice.allspice.io/api/swagger#/repository/repoListCommitStatuses
Inherited Members
class
CommitStatusState(enum.Enum):
1820class CommitStatusState(Enum): 1821 PENDING = "pending" 1822 SUCCESS = "success" 1823 ERROR = "error" 1824 FAILURE = "failure" 1825 WARNING = "warning" 1826 1827 @classmethod 1828 def try_init(cls, value: str) -> CommitStatusState | str: 1829 """ 1830 Try converting a string to the enum, and if that fails, return the 1831 string itself. 1832 """ 1833 1834 try: 1835 return cls(value) 1836 except ValueError: 1837 return value
PENDING =
<CommitStatusState.PENDING: 'pending'>
SUCCESS =
<CommitStatusState.SUCCESS: 'success'>
ERROR =
<CommitStatusState.ERROR: 'error'>
FAILURE =
<CommitStatusState.FAILURE: 'failure'>
WARNING =
<CommitStatusState.WARNING: 'warning'>
1827 @classmethod 1828 def try_init(cls, value: str) -> CommitStatusState | str: 1829 """ 1830 Try converting a string to the enum, and if that fails, return the 1831 string itself. 1832 """ 1833 1834 try: 1835 return cls(value) 1836 except ValueError: 1837 return value
Try converting a string to the enum, and if that fails, return the string itself.
1840class CommitStatus(ReadonlyApiObject): 1841 context: str 1842 created_at: str 1843 creator: User 1844 description: str 1845 id: int 1846 status: CommitStatusState 1847 target_url: str 1848 updated_at: str 1849 url: str 1850 1851 def __init__(self, allspice_client): 1852 super().__init__(allspice_client) 1853 1854 _fields_to_parsers: ClassVar[dict] = { 1855 # Gitea/ASH doesn't actually validate that the status is a "valid" 1856 # status, so we can expect empty or unknown strings in the status field. 1857 "status": lambda _, s: CommitStatusState.try_init(s), 1858 "creator": lambda allspice_client, u: ( 1859 User.parse_response(allspice_client, u) if u else None 1860 ), 1861 } 1862 1863 def __eq__(self, other): 1864 if not isinstance(other, CommitStatus): 1865 return False 1866 return self.id == other.id 1867 1868 def __hash__(self): 1869 return hash(self.id)
creator: User
status: CommitStatusState
Inherited Members
1872class CommitCombinedStatus(ReadonlyApiObject): 1873 commit_url: str 1874 repository: Repository 1875 sha: str 1876 state: CommitStatusState 1877 statuses: List["CommitStatus"] 1878 total_count: int 1879 url: str 1880 1881 def __init__(self, allspice_client): 1882 super().__init__(allspice_client) 1883 1884 _fields_to_parsers: ClassVar[dict] = { 1885 # See CommitStatus 1886 "state": lambda _, s: CommitStatusState.try_init(s), 1887 "statuses": lambda allspice_client, statuses: [ 1888 CommitStatus.parse_response(allspice_client, status) for status in statuses 1889 ], 1890 "repository": lambda allspice_client, r: Repository.parse_response(allspice_client, r), 1891 } 1892 1893 def __eq__(self, other): 1894 if not isinstance(other, CommitCombinedStatus): 1895 return False 1896 return self.sha == other.sha 1897 1898 def __hash__(self): 1899 return hash(self.sha) 1900 1901 @classmethod 1902 def parse_response(cls, allspice_client, result) -> "CommitCombinedStatus": 1903 api_object = cls(allspice_client) 1904 cls._initialize(allspice_client, api_object, result) 1905 return api_object
repository: Repository
state: CommitStatusState
statuses: List[CommitStatus]
Inherited Members
1908class Issue(ApiObject): 1909 """ 1910 An issue on a repository. 1911 1912 Note: `Issue.assets` may not have any entries even if the issue has 1913 attachments. This happens when an issue is fetched via a bulk method like 1914 `Repository.get_issues`. In most cases, prefer using 1915 `Issue.get_attachments` to get the attachments on an issue. 1916 """ 1917 1918 assets: List[Union[Any, "Attachment"]] 1919 assignee: Any 1920 assignees: Any 1921 body: str 1922 closed_at: Any 1923 comments: int 1924 created_at: str 1925 due_date: Any 1926 html_url: str 1927 id: int 1928 is_locked: bool 1929 labels: List[Any] 1930 milestone: Optional["Milestone"] 1931 number: int 1932 original_author: str 1933 original_author_id: int 1934 pin_order: int 1935 pull_request: Any 1936 ref: str 1937 repository: Dict[str, Union[int, str]] 1938 state: str 1939 title: str 1940 updated_at: str 1941 url: str 1942 user: User 1943 1944 API_OBJECT = """/repos/{owner}/{repo}/issues/{index}""" # <owner, repo, index> 1945 GET_TIME = """/repos/%s/%s/issues/%s/times""" # <owner, repo, index> 1946 GET_COMMENTS = """/repos/{owner}/{repo}/issues/{index}/comments""" 1947 CREATE_ISSUE = """/repos/{owner}/{repo}/issues""" 1948 GET_ATTACHMENTS = """/repos/{owner}/{repo}/issues/{index}/assets""" 1949 1950 OPENED = "open" 1951 CLOSED = "closed" 1952 1953 def __init__(self, allspice_client): 1954 super().__init__(allspice_client) 1955 1956 def __eq__(self, other): 1957 if not isinstance(other, Issue): 1958 return False 1959 return self.repository == other.repository and self.id == other.id 1960 1961 def __hash__(self): 1962 return hash(self.repository) ^ hash(self.id) 1963 1964 _fields_to_parsers: ClassVar[dict] = { 1965 "milestone": lambda allspice_client, m: Milestone.parse_response(allspice_client, m), 1966 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 1967 "assets": lambda allspice_client, assets: [ 1968 Attachment.parse_response(allspice_client, a) for a in assets 1969 ], 1970 "assignee": lambda allspice_client, u: User.parse_response(allspice_client, u), 1971 "assignees": lambda allspice_client, us: [ 1972 User.parse_response(allspice_client, u) for u in us 1973 ], 1974 "state": lambda _, s: Issue.CLOSED if s == "closed" else Issue.OPENED, 1975 } 1976 1977 _parsers_to_fields: ClassVar[dict] = { 1978 "milestone": lambda m: m.id, 1979 } 1980 1981 _patchable_fields: ClassVar[set[str]] = { 1982 "assignee", 1983 "assignees", 1984 "body", 1985 "due_date", 1986 "milestone", 1987 "state", 1988 "title", 1989 } 1990 1991 def commit(self): 1992 args = { 1993 "owner": self.repository.owner.username, 1994 "repo": self.repository.name, 1995 "index": self.number, 1996 } 1997 self._commit(args) 1998 1999 @classmethod 2000 def request(cls, allspice_client, owner: str, repo: str, number: str): 2001 api_object = cls._request(allspice_client, {"owner": owner, "repo": repo, "index": number}) 2002 # The repository in the response is a RepositoryMeta object, so request 2003 # the full repository object and add it to the issue object. 2004 repository = Repository.request(allspice_client, owner, repo) 2005 setattr(api_object, "_repository", repository) 2006 # For legacy reasons 2007 cls._add_read_property("repo", repository, api_object) 2008 return api_object 2009 2010 @classmethod 2011 def create_issue(cls, allspice_client, repo: Repository, title: str, body: str = ""): 2012 args = {"owner": repo.owner.username, "repo": repo.name} 2013 data = {"title": title, "body": body} 2014 result = allspice_client.requests_post(Issue.CREATE_ISSUE.format(**args), data=data) 2015 issue = Issue.parse_response(allspice_client, result) 2016 setattr(issue, "_repository", repo) 2017 cls._add_read_property("repo", repo, issue) 2018 return issue 2019 2020 @property 2021 def owner(self) -> Organization | User: 2022 return self.repository.owner 2023 2024 def get_time_sum(self, user: User) -> int: 2025 results = self.allspice_client.requests_get( 2026 Issue.GET_TIME % (self.owner.username, self.repository.name, self.number) 2027 ) 2028 return sum(result["time"] for result in results if result and result["user_id"] == user.id) 2029 2030 def get_times(self) -> Optional[Dict]: 2031 return self.allspice_client.requests_get( 2032 Issue.GET_TIME % (self.owner.username, self.repository.name, self.number) 2033 ) 2034 2035 def delete_time(self, time_id: str): 2036 path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times/{time_id}" 2037 self.allspice_client.requests_delete(path) 2038 2039 def add_time(self, time: int, created: Optional[str] = None, user_name: Optional[User] = None): 2040 path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times" 2041 self.allspice_client.requests_post( 2042 path, data={"created": created, "time": int(time), "user_name": user_name} 2043 ) 2044 2045 def get_comments(self) -> List[Comment]: 2046 """https://hub.allspice.io/api/swagger#/issue/issueGetComments""" 2047 2048 results = self.allspice_client.requests_get( 2049 self.GET_COMMENTS.format( 2050 owner=self.owner.username, repo=self.repository.name, index=self.number 2051 ) 2052 ) 2053 2054 return [Comment.parse_response(self.allspice_client, result) for result in results] 2055 2056 def create_comment(self, body: str) -> Comment: 2057 """https://hub.allspice.io/api/swagger#/issue/issueCreateComment""" 2058 2059 path = self.GET_COMMENTS.format( 2060 owner=self.owner.username, repo=self.repository.name, index=self.number 2061 ) 2062 2063 response = self.allspice_client.requests_post(path, data={"body": body}) 2064 return Comment.parse_response(self.allspice_client, response) 2065 2066 def get_attachments(self) -> List[Attachment]: 2067 """ 2068 Fetch all attachments on this issue. 2069 2070 Unlike the assets field, this will always fetch all attachments from the 2071 API. 2072 2073 See https://hub.allspice.io/api/swagger#/issue/issueListIssueAttachments 2074 """ 2075 2076 path = self.GET_ATTACHMENTS.format( 2077 owner=self.owner.username, repo=self.repository.name, index=self.number 2078 ) 2079 response = self.allspice_client.requests_get(path) 2080 2081 return [Attachment.parse_response(self.allspice_client, result) for result in response] 2082 2083 def create_attachment(self, file: IO, name: Optional[str] = None) -> Attachment: 2084 """ 2085 Create an attachment on this issue. 2086 2087 https://hub.allspice.io/api/swagger#/issue/issueCreateIssueAttachment 2088 2089 :param file: The file to attach. This should be a file-like object. 2090 :param name: The name of the file. If not provided, the name of the file will be 2091 used. 2092 :return: The created attachment. 2093 """ 2094 2095 args: dict[str, Any] = { 2096 "files": {"attachment": file}, 2097 } 2098 if name is not None: 2099 args["params"] = {"name": name} 2100 2101 result = self.allspice_client.requests_post( 2102 self.GET_ATTACHMENTS.format( 2103 owner=self.owner.username, repo=self.repository.name, index=self.number 2104 ), 2105 **args, 2106 ) 2107 2108 return Attachment.parse_response(self.allspice_client, result)
An issue on a repository.
Note: Issue.assets may not have any entries even if the issue has
attachments. This happens when an issue is fetched via a bulk method like
Repository.get_issues. In most cases, prefer using
Issue.get_attachments to get the attachments on an issue.
assets: List[Union[Any, Attachment]]
milestone: Optional[Milestone]
user: User
@classmethod
def
request(cls, allspice_client, owner: str, repo: str, number: str):
1999 @classmethod 2000 def request(cls, allspice_client, owner: str, repo: str, number: str): 2001 api_object = cls._request(allspice_client, {"owner": owner, "repo": repo, "index": number}) 2002 # The repository in the response is a RepositoryMeta object, so request 2003 # the full repository object and add it to the issue object. 2004 repository = Repository.request(allspice_client, owner, repo) 2005 setattr(api_object, "_repository", repository) 2006 # For legacy reasons 2007 cls._add_read_property("repo", repository, api_object) 2008 return api_object
2010 @classmethod 2011 def create_issue(cls, allspice_client, repo: Repository, title: str, body: str = ""): 2012 args = {"owner": repo.owner.username, "repo": repo.name} 2013 data = {"title": title, "body": body} 2014 result = allspice_client.requests_post(Issue.CREATE_ISSUE.format(**args), data=data) 2015 issue = Issue.parse_response(allspice_client, result) 2016 setattr(issue, "_repository", repo) 2017 cls._add_read_property("repo", repo, issue) 2018 return issue
owner: Organization | User
2039 def add_time(self, time: int, created: Optional[str] = None, user_name: Optional[User] = None): 2040 path = f"/repos/{self.owner.username}/{self.repository.name}/issues/{self.number}/times" 2041 self.allspice_client.requests_post( 2042 path, data={"created": created, "time": int(time), "user_name": user_name} 2043 )
2045 def get_comments(self) -> List[Comment]: 2046 """https://hub.allspice.io/api/swagger#/issue/issueGetComments""" 2047 2048 results = self.allspice_client.requests_get( 2049 self.GET_COMMENTS.format( 2050 owner=self.owner.username, repo=self.repository.name, index=self.number 2051 ) 2052 ) 2053 2054 return [Comment.parse_response(self.allspice_client, result) for result in results]
allspice.allspice.io/api/swagger#/issue/issueGetComments">https://huballspice.allspice.io/api/swagger#/issue/issueGetComments
2056 def create_comment(self, body: str) -> Comment: 2057 """https://hub.allspice.io/api/swagger#/issue/issueCreateComment""" 2058 2059 path = self.GET_COMMENTS.format( 2060 owner=self.owner.username, repo=self.repository.name, index=self.number 2061 ) 2062 2063 response = self.allspice_client.requests_post(path, data={"body": body}) 2064 return Comment.parse_response(self.allspice_client, response)
allspice.allspice.io/api/swagger#/issue/issueCreateComment">https://huballspice.allspice.io/api/swagger#/issue/issueCreateComment
2066 def get_attachments(self) -> List[Attachment]: 2067 """ 2068 Fetch all attachments on this issue. 2069 2070 Unlike the assets field, this will always fetch all attachments from the 2071 API. 2072 2073 See https://hub.allspice.io/api/swagger#/issue/issueListIssueAttachments 2074 """ 2075 2076 path = self.GET_ATTACHMENTS.format( 2077 owner=self.owner.username, repo=self.repository.name, index=self.number 2078 ) 2079 response = self.allspice_client.requests_get(path) 2080 2081 return [Attachment.parse_response(self.allspice_client, result) for result in response]
Fetch all attachments on this issue.
Unlike the assets field, this will always fetch all attachments from the API.
See allspice.allspice.io/api/swagger#/issue/issueListIssueAttachments">https://huballspice.allspice.io/api/swagger#/issue/issueListIssueAttachments
2083 def create_attachment(self, file: IO, name: Optional[str] = None) -> Attachment: 2084 """ 2085 Create an attachment on this issue. 2086 2087 https://hub.allspice.io/api/swagger#/issue/issueCreateIssueAttachment 2088 2089 :param file: The file to attach. This should be a file-like object. 2090 :param name: The name of the file. If not provided, the name of the file will be 2091 used. 2092 :return: The created attachment. 2093 """ 2094 2095 args: dict[str, Any] = { 2096 "files": {"attachment": file}, 2097 } 2098 if name is not None: 2099 args["params"] = {"name": name} 2100 2101 result = self.allspice_client.requests_post( 2102 self.GET_ATTACHMENTS.format( 2103 owner=self.owner.username, repo=self.repository.name, index=self.number 2104 ), 2105 **args, 2106 ) 2107 2108 return Attachment.parse_response(self.allspice_client, result)
Create an attachment on this issue.
allspice.allspice.io/api/swagger#/issue/issueCreateIssueAttachment">https://huballspice.allspice.io/api/swagger#/issue/issueCreateIssueAttachment
Parameters
- file: The file to attach. This should be a file-like object.
- name: The name of the file. If not provided, the name of the file will be used.
Returns
The created attachment.
2111class DesignReviewReviewComment(ApiObject): 2112 """ 2113 A comment on a Design Review Review. 2114 """ 2115 2116 body: str 2117 commit_id: str 2118 created_at: str 2119 diff_hunk: str 2120 html_url: str 2121 id: int 2122 original_commit_id: str 2123 original_position: int 2124 path: str 2125 position: int 2126 pull_request_review_id: int 2127 pull_request_url: str 2128 resolver: Any 2129 sub_path: str 2130 updated_at: str 2131 user: User 2132 2133 def __init__(self, allspice_client): 2134 super().__init__(allspice_client) 2135 2136 _fields_to_parsers: ClassVar[dict] = { 2137 "resolver": lambda allspice_client, r: User.parse_response(allspice_client, r), 2138 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 2139 }
A comment on a Design Review Review.
user: User
2142class DesignReviewReview(ReadonlyApiObject): 2143 """ 2144 A review on a Design Review. 2145 """ 2146 2147 body: str 2148 comments_count: int 2149 commit_id: str 2150 dismissed: bool 2151 html_url: str 2152 id: int 2153 official: bool 2154 pull_request_url: str 2155 stale: bool 2156 state: ReviewEvent 2157 submitted_at: str 2158 team: Any 2159 updated_at: str 2160 user: User 2161 2162 API_OBJECT = "/repos/{owner}/{repo}/pulls/{index}/reviews/{id}" 2163 GET_COMMENTS = "/repos/{owner}/{repo}/pulls/{index}/reviews/{id}/comments" 2164 2165 class ReviewEvent(Enum): 2166 APPROVED = "APPROVED" 2167 PENDING = "PENDING" 2168 COMMENT = "COMMENT" 2169 REQUEST_CHANGES = "REQUEST_CHANGES" 2170 REQUEST_REVIEW = "REQUEST_REVIEW" 2171 UNKNOWN = "" 2172 2173 @dataclass 2174 class ReviewComment: 2175 """ 2176 Data required to create a review comment on a design review. 2177 2178 :param body: The body of the comment. 2179 :param path: The path of the file to comment on. If you have a 2180 `Content` object, get the path using the `path` property. 2181 :param sub_path: The sub-path of the file to comment on. This is 2182 usually the page ID of the page in the multi-page document. 2183 :param new_position: The line number of the source code file after the 2184 change to add this comment on. Optional, leave unset if this is an ECAD 2185 file or the comment must be on the entire file. 2186 :param old_position: The line number of the source code file before the 2187 change to add this comment on. Optional, leave unset if this is an ECAD 2188 file or the comment must be on the entire file. 2189 """ 2190 2191 body: str 2192 path: str 2193 sub_path: Optional[str] = None 2194 new_position: Optional[int] = None 2195 old_position: Optional[int] = None 2196 2197 def __init__(self, allspice_client): 2198 super().__init__(allspice_client) 2199 2200 _fields_to_parsers: ClassVar[dict] = { 2201 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 2202 "state": lambda _, s: DesignReviewReview.ReviewEvent(s), 2203 } 2204 2205 def _get_dr_properties(self) -> dict[str, str]: 2206 """ 2207 Get the owner, repo name and design review number from the URL of this 2208 review's DR. 2209 """ 2210 2211 parts = self.pull_request_url.strip("/").split("/") 2212 2213 try: 2214 index = parts[-1] 2215 assert parts[-2] == "pulls" or parts[-2] == "pull", ( 2216 "Expected the second last part of the URL to be 'pulls' or 'pull', " 2217 ) 2218 repo = parts[-3] 2219 owner = parts[-4] 2220 2221 return { 2222 "owner": owner, 2223 "repo": repo, 2224 "index": index, 2225 } 2226 except IndexError: 2227 raise ValueError("Malformed design review URL: {}".format(self.pull_request_url)) 2228 2229 @cached_property 2230 def owner_name(self) -> str: 2231 """ 2232 The owner of the repository this review is on. 2233 """ 2234 2235 return self._get_dr_properties()["owner"] 2236 2237 @cached_property 2238 def repository_name(self) -> str: 2239 """ 2240 The name of the repository this review is on. 2241 """ 2242 2243 return self._get_dr_properties()["repo"] 2244 2245 @cached_property 2246 def index(self) -> str: 2247 """ 2248 The index of the design review this review is on. 2249 """ 2250 2251 return self._get_dr_properties()["index"] 2252 2253 def delete(self): 2254 """ 2255 Delete this review. 2256 """ 2257 2258 self.allspice_client.requests_delete( 2259 self.API_OBJECT.format(**self._get_dr_properties(), id=self.id) 2260 ) 2261 self.deleted = True 2262 2263 def get_comments(self) -> List[DesignReviewReviewComment]: 2264 """ 2265 Get the comments on this review. 2266 """ 2267 2268 result = self.allspice_client.requests_get( 2269 self.GET_COMMENTS.format(**self._get_dr_properties(), id=self.id) 2270 ) 2271 2272 return [ 2273 DesignReviewReviewComment.parse_response(self.allspice_client, comment) 2274 for comment in result 2275 ]
A review on a Design Review.
user: User
owner_name: str
2229 @cached_property 2230 def owner_name(self) -> str: 2231 """ 2232 The owner of the repository this review is on. 2233 """ 2234 2235 return self._get_dr_properties()["owner"]
The owner of the repository this review is on.
repository_name: str
2237 @cached_property 2238 def repository_name(self) -> str: 2239 """ 2240 The name of the repository this review is on. 2241 """ 2242 2243 return self._get_dr_properties()["repo"]
The name of the repository this review is on.
index: str
2245 @cached_property 2246 def index(self) -> str: 2247 """ 2248 The index of the design review this review is on. 2249 """ 2250 2251 return self._get_dr_properties()["index"]
The index of the design review this review is on.
def
delete(self):
2253 def delete(self): 2254 """ 2255 Delete this review. 2256 """ 2257 2258 self.allspice_client.requests_delete( 2259 self.API_OBJECT.format(**self._get_dr_properties(), id=self.id) 2260 ) 2261 self.deleted = True
Delete this review.
2263 def get_comments(self) -> List[DesignReviewReviewComment]: 2264 """ 2265 Get the comments on this review. 2266 """ 2267 2268 result = self.allspice_client.requests_get( 2269 self.GET_COMMENTS.format(**self._get_dr_properties(), id=self.id) 2270 ) 2271 2272 return [ 2273 DesignReviewReviewComment.parse_response(self.allspice_client, comment) 2274 for comment in result 2275 ]
Get the comments on this review.
Inherited Members
class
DesignReviewReview.ReviewEvent(enum.Enum):
@dataclass
class
DesignReviewReview.ReviewComment:
2173 @dataclass 2174 class ReviewComment: 2175 """ 2176 Data required to create a review comment on a design review. 2177 2178 :param body: The body of the comment. 2179 :param path: The path of the file to comment on. If you have a 2180 `Content` object, get the path using the `path` property. 2181 :param sub_path: The sub-path of the file to comment on. This is 2182 usually the page ID of the page in the multi-page document. 2183 :param new_position: The line number of the source code file after the 2184 change to add this comment on. Optional, leave unset if this is an ECAD 2185 file or the comment must be on the entire file. 2186 :param old_position: The line number of the source code file before the 2187 change to add this comment on. Optional, leave unset if this is an ECAD 2188 file or the comment must be on the entire file. 2189 """ 2190 2191 body: str 2192 path: str 2193 sub_path: Optional[str] = None 2194 new_position: Optional[int] = None 2195 old_position: Optional[int] = None
Data required to create a review comment on a design review.
Parameters
- body: The body of the comment.
- path: The path of the file to comment on. If you have a
Contentobject, get the path using thepathproperty. - sub_path: The sub-path of the file to comment on. This is usually the page ID of the page in the multi-page document.
- new_position: The line number of the source code file after the change to add this comment on. Optional, leave unset if this is an ECAD file or the comment must be on the entire file.
- old_position: The line number of the source code file before the change to add this comment on. Optional, leave unset if this is an ECAD file or the comment must be on the entire file.
2278class DesignReview(ApiObject): 2279 """ 2280 A Design Review. See 2281 https://hub.allspice.io/api/swagger#/repository/repoGetPullRequest. 2282 2283 Note: The base and head fields are not `Branch` objects - they are plain strings 2284 referring to the branch names. This is because DRs can exist for branches that have 2285 been deleted, which don't have an associated `Branch` object from the API. You can use 2286 the `Repository.get_branch` method to get a `Branch` object for a branch if you know 2287 it exists. 2288 """ 2289 2290 additions: Optional[int] 2291 allow_maintainer_edit: bool 2292 allow_maintainer_edits: Any 2293 assignee: User 2294 assignees: List["User"] 2295 base: str 2296 body: str 2297 changed_files: Optional[int] 2298 closed_at: Optional[str] 2299 comments: int 2300 created_at: str 2301 deletions: Optional[int] 2302 diff_url: str 2303 draft: bool 2304 due_date: Optional[str] 2305 head: str 2306 html_url: str 2307 id: int 2308 is_locked: bool 2309 labels: List[Any] 2310 merge_base: str 2311 merge_commit_sha: Optional[str] 2312 mergeable: bool 2313 merged: bool 2314 merged_at: Optional[str] 2315 merged_by: Any 2316 milestone: Any 2317 number: int 2318 patch_url: str 2319 pin_order: int 2320 repository: Optional["Repository"] 2321 requested_reviewers: Any 2322 requested_reviewers_teams: Any 2323 review_comments: int 2324 state: str 2325 title: str 2326 updated_at: str 2327 url: str 2328 user: User 2329 2330 API_OBJECT = "/repos/{owner}/{repo}/pulls/{index}" 2331 MERGE_DESIGN_REVIEW = "/repos/{owner}/{repo}/pulls/{index}/merge" 2332 GET_REVIEWS = "/repos/{owner}/{repo}/pulls/{index}/reviews" 2333 GET_REVIEW = "/repos/{owner}/{repo}/pulls/{index}/reviews/{review_id}" 2334 GET_COMMENTS = "/repos/{owner}/{repo}/issues/{index}/comments" 2335 2336 OPEN = "open" 2337 CLOSED = "closed" 2338 2339 class MergeType(Enum): 2340 MERGE = "merge" 2341 REBASE = "rebase" 2342 REBASE_MERGE = "rebase-merge" 2343 SQUASH = "squash" 2344 MANUALLY_MERGED = "manually-merged" 2345 2346 def __init__(self, allspice_client): 2347 super().__init__(allspice_client) 2348 2349 def __eq__(self, other): 2350 if not isinstance(other, DesignReview): 2351 return False 2352 return self.repository == other.repository and self.id == other.id 2353 2354 def __hash__(self): 2355 return hash(self.repository) ^ hash(self.id) 2356 2357 @classmethod 2358 def parse_response(cls, allspice_client, result) -> "DesignReview": 2359 api_object = super().parse_response(allspice_client, result) 2360 cls._add_read_property( 2361 "repository", 2362 Repository.parse_response(allspice_client, result["base"]["repo"]), 2363 api_object, 2364 ) 2365 2366 return api_object 2367 2368 @classmethod 2369 def request(cls, allspice_client, owner: str, repo: str, number: str): 2370 """See https://hub.allspice.io/api/swagger#/repository/repoGetPullRequest""" 2371 return cls._request(allspice_client, {"owner": owner, "repo": repo, "index": number}) 2372 2373 _fields_to_parsers: ClassVar[dict] = { 2374 "assignee": lambda allspice_client, u: User.parse_response(allspice_client, u), 2375 "assignees": lambda allspice_client, us: [ 2376 User.parse_response(allspice_client, u) for u in us 2377 ], 2378 "base": lambda _, b: b["ref"], 2379 "head": lambda _, h: h["ref"], 2380 "merged_by": lambda allspice_client, u: User.parse_response(allspice_client, u), 2381 "milestone": lambda allspice_client, m: Milestone.parse_response(allspice_client, m), 2382 "user": lambda allspice_client, u: User.parse_response(allspice_client, u), 2383 } 2384 2385 _patchable_fields: ClassVar[set[str]] = { 2386 "allow_maintainer_edits", 2387 "assignee", 2388 "assignees", 2389 "base", 2390 "body", 2391 "due_date", 2392 "milestone", 2393 "state", 2394 "title", 2395 } 2396 2397 _parsers_to_fields: ClassVar[dict] = { 2398 "assignee": lambda u: u.username, 2399 "assignees": lambda us: [u.username for u in us], 2400 "base": lambda b: b.name if isinstance(b, Branch) else b, 2401 "milestone": lambda m: m.id, 2402 } 2403 2404 def commit(self): 2405 data = self.get_dirty_fields() 2406 if "due_date" in data and data["due_date"] is None: 2407 data["unset_due_date"] = True 2408 2409 args = { 2410 "owner": self.repository.owner.username, 2411 "repo": self.repository.name, 2412 "index": self.number, 2413 } 2414 self._commit(args, data) 2415 2416 def merge(self, merge_type: MergeType): 2417 """ 2418 Merge the pull request. See 2419 https://hub.allspice.io/api/swagger#/repository/repoMergePullRequest 2420 2421 :param merge_type: The type of merge to perform. See the MergeType enum. 2422 """ 2423 2424 self.allspice_client.requests_post( 2425 self.MERGE_DESIGN_REVIEW.format( 2426 owner=self.repository.owner.username, 2427 repo=self.repository.name, 2428 index=self.number, 2429 ), 2430 data={"Do": merge_type.value}, 2431 ) 2432 2433 def get_comments(self) -> List[Comment]: 2434 """ 2435 Get the comments on this pull request, but not specifically on a review. 2436 2437 https://hub.allspice.io/api/swagger#/issue/issueGetComments 2438 2439 :return: A list of comments on this pull request. 2440 """ 2441 2442 results = self.allspice_client.requests_get( 2443 self.GET_COMMENTS.format( 2444 owner=self.repository.owner.username, 2445 repo=self.repository.name, 2446 index=self.number, 2447 ) 2448 ) 2449 return [Comment.parse_response(self.allspice_client, result) for result in results] 2450 2451 def create_comment(self, body: str): 2452 """ 2453 Create a comment on this pull request. This uses the same endpoint as the 2454 comments on issues, and will not be associated with any reviews. 2455 2456 https://hub.allspice.io/api/swagger#/issue/issueCreateComment 2457 2458 :param body: The body of the comment. 2459 :return: The comment that was created. 2460 """ 2461 2462 result = self.allspice_client.requests_post( 2463 self.GET_COMMENTS.format( 2464 owner=self.repository.owner.username, 2465 repo=self.repository.name, 2466 index=self.number, 2467 ), 2468 data={"body": body}, 2469 ) 2470 return Comment.parse_response(self.allspice_client, result) 2471 2472 def create_review( 2473 self, 2474 *, 2475 body: Optional[str] = None, 2476 event: Optional[DesignReviewReview.ReviewEvent] = None, 2477 comments: Optional[List[DesignReviewReview.ReviewComment]] = None, 2478 commit_id: Optional[str] = None, 2479 ) -> DesignReviewReview: 2480 """ 2481 Create a review on this design review. 2482 2483 https://hub.allspice.io/api/swagger#/repository/repoCreatePullRequestReview 2484 2485 Note: in most cases, you should not set the body or event when creating 2486 a review. The event is automatically set to "PENDING" when the review 2487 is created. You should then use `submit_review` to submit the review 2488 with the desired event and body. 2489 2490 :param body: The body of the review. This is the top-level comment on 2491 the review. If not provided, the review will be created with no body. 2492 :param event: The event of the review. This is the overall status of the 2493 review. See the ReviewEvent enum. If not provided, the API will 2494 default to "PENDING". 2495 :param comments: A list of comments on the review. Each comment should 2496 be a ReviewComment object. If not provided, only the base comment 2497 will be created. 2498 :param commit_id: The commit SHA to associate with the review. This is 2499 optional. 2500 """ 2501 2502 data: dict[str, Any] = {} 2503 2504 if body is not None: 2505 data["body"] = body 2506 if event is not None: 2507 data["event"] = event.value 2508 if commit_id is not None: 2509 data["commit_id"] = commit_id 2510 if comments is not None: 2511 data["comments"] = [asdict(comment) for comment in comments] 2512 2513 result = self.allspice_client.requests_post( 2514 self.GET_REVIEWS.format( 2515 owner=self.repository.owner.username, 2516 repo=self.repository.name, 2517 index=self.number, 2518 ), 2519 data=data, 2520 ) 2521 2522 return DesignReviewReview.parse_response(self.allspice_client, result) 2523 2524 def get_reviews(self) -> List[DesignReviewReview]: 2525 """ 2526 Get all reviews on this design review. 2527 2528 https://hub.allspice.io/api/swagger#/repository/repoListPullReviews 2529 """ 2530 2531 results = self.allspice_client.requests_get( 2532 self.GET_REVIEWS.format( 2533 owner=self.repository.owner.username, 2534 repo=self.repository.name, 2535 index=self.number, 2536 ) 2537 ) 2538 2539 return [ 2540 DesignReviewReview.parse_response(self.allspice_client, result) for result in results 2541 ] 2542 2543 def submit_review( 2544 self, 2545 review_id: int, 2546 event: DesignReviewReview.ReviewEvent, 2547 *, 2548 body: Optional[str] = None, 2549 ): 2550 """ 2551 Submit a review on this design review. 2552 2553 https://hub.allspice.io/api/swagger#/repository/repoSubmitPullReview 2554 2555 :param review_id: The ID of the review to submit. 2556 :param event: The event to submit the review with. See the ReviewEvent 2557 enum for the possible values. 2558 :param body: Optional body text for the review submission. 2559 """ 2560 2561 data = { 2562 "event": event.value, 2563 } 2564 if body is not None: 2565 data["body"] = body 2566 2567 result = self.allspice_client.requests_post( 2568 self.GET_REVIEW.format( 2569 owner=self.repository.owner.username, 2570 repo=self.repository.name, 2571 index=self.number, 2572 review_id=review_id, 2573 ), 2574 data=data, 2575 ) 2576 2577 return result
A Design Review. See allspice.allspice.io/api/swagger#/repository/repoGetPullRequest">https://huballspice.allspice.io/api/swagger#/repository/repoGetPullRequest.
Note: The base and head fields are not Branch objects - they are plain strings
referring to the branch names. This is because DRs can exist for branches that have
been deleted, which don't have an associated Branch object from the API. You can use
the Repository.get_branch method to get a Branch object for a branch if you know
it exists.
assignee: User
assignees: List[User]
repository: Optional[Repository]
user: User
2357 @classmethod 2358 def parse_response(cls, allspice_client, result) -> "DesignReview": 2359 api_object = super().parse_response(allspice_client, result) 2360 cls._add_read_property( 2361 "repository", 2362 Repository.parse_response(allspice_client, result["base"]["repo"]), 2363 api_object, 2364 ) 2365 2366 return api_object
@classmethod
def
request(cls, allspice_client, owner: str, repo: str, number: str):
2368 @classmethod 2369 def request(cls, allspice_client, owner: str, repo: str, number: str): 2370 """See https://hub.allspice.io/api/swagger#/repository/repoGetPullRequest""" 2371 return cls._request(allspice_client, {"owner": owner, "repo": repo, "index": number})
See allspice.allspice.io/api/swagger#/repository/repoGetPullRequest">https://huballspice.allspice.io/api/swagger#/repository/repoGetPullRequest
def
commit(self):
2404 def commit(self): 2405 data = self.get_dirty_fields() 2406 if "due_date" in data and data["due_date"] is None: 2407 data["unset_due_date"] = True 2408 2409 args = { 2410 "owner": self.repository.owner.username, 2411 "repo": self.repository.name, 2412 "index": self.number, 2413 } 2414 self._commit(args, data)
2416 def merge(self, merge_type: MergeType): 2417 """ 2418 Merge the pull request. See 2419 https://hub.allspice.io/api/swagger#/repository/repoMergePullRequest 2420 2421 :param merge_type: The type of merge to perform. See the MergeType enum. 2422 """ 2423 2424 self.allspice_client.requests_post( 2425 self.MERGE_DESIGN_REVIEW.format( 2426 owner=self.repository.owner.username, 2427 repo=self.repository.name, 2428 index=self.number, 2429 ), 2430 data={"Do": merge_type.value}, 2431 )
Merge the pull request. See allspice.allspice.io/api/swagger#/repository/repoMergePullRequest">https://huballspice.allspice.io/api/swagger#/repository/repoMergePullRequest
Parameters
- merge_type: The type of merge to perform. See the MergeType enum.
2433 def get_comments(self) -> List[Comment]: 2434 """ 2435 Get the comments on this pull request, but not specifically on a review. 2436 2437 https://hub.allspice.io/api/swagger#/issue/issueGetComments 2438 2439 :return: A list of comments on this pull request. 2440 """ 2441 2442 results = self.allspice_client.requests_get( 2443 self.GET_COMMENTS.format( 2444 owner=self.repository.owner.username, 2445 repo=self.repository.name, 2446 index=self.number, 2447 ) 2448 ) 2449 return [Comment.parse_response(self.allspice_client, result) for result in results]
Get the comments on this pull request, but not specifically on a review.
allspice.allspice.io/api/swagger#/issue/issueGetComments">https://huballspice.allspice.io/api/swagger#/issue/issueGetComments
Returns
A list of comments on this pull request.
def
create_comment(self, body: str):
2451 def create_comment(self, body: str): 2452 """ 2453 Create a comment on this pull request. This uses the same endpoint as the 2454 comments on issues, and will not be associated with any reviews. 2455 2456 https://hub.allspice.io/api/swagger#/issue/issueCreateComment 2457 2458 :param body: The body of the comment. 2459 :return: The comment that was created. 2460 """ 2461 2462 result = self.allspice_client.requests_post( 2463 self.GET_COMMENTS.format( 2464 owner=self.repository.owner.username, 2465 repo=self.repository.name, 2466 index=self.number, 2467 ), 2468 data={"body": body}, 2469 ) 2470 return Comment.parse_response(self.allspice_client, result)
Create a comment on this pull request. This uses the same endpoint as the comments on issues, and will not be associated with any reviews.
allspice.allspice.io/api/swagger#/issue/issueCreateComment">https://huballspice.allspice.io/api/swagger#/issue/issueCreateComment
Parameters
- body: The body of the comment.
Returns
The comment that was created.
def
create_review( self, *, body: Optional[str] = None, event: Optional[DesignReviewReview.ReviewEvent] = None, comments: Optional[List[DesignReviewReview.ReviewComment]] = None, commit_id: Optional[str] = None) -> DesignReviewReview:
2472 def create_review( 2473 self, 2474 *, 2475 body: Optional[str] = None, 2476 event: Optional[DesignReviewReview.ReviewEvent] = None, 2477 comments: Optional[List[DesignReviewReview.ReviewComment]] = None, 2478 commit_id: Optional[str] = None, 2479 ) -> DesignReviewReview: 2480 """ 2481 Create a review on this design review. 2482 2483 https://hub.allspice.io/api/swagger#/repository/repoCreatePullRequestReview 2484 2485 Note: in most cases, you should not set the body or event when creating 2486 a review. The event is automatically set to "PENDING" when the review 2487 is created. You should then use `submit_review` to submit the review 2488 with the desired event and body. 2489 2490 :param body: The body of the review. This is the top-level comment on 2491 the review. If not provided, the review will be created with no body. 2492 :param event: The event of the review. This is the overall status of the 2493 review. See the ReviewEvent enum. If not provided, the API will 2494 default to "PENDING". 2495 :param comments: A list of comments on the review. Each comment should 2496 be a ReviewComment object. If not provided, only the base comment 2497 will be created. 2498 :param commit_id: The commit SHA to associate with the review. This is 2499 optional. 2500 """ 2501 2502 data: dict[str, Any] = {} 2503 2504 if body is not None: 2505 data["body"] = body 2506 if event is not None: 2507 data["event"] = event.value 2508 if commit_id is not None: 2509 data["commit_id"] = commit_id 2510 if comments is not None: 2511 data["comments"] = [asdict(comment) for comment in comments] 2512 2513 result = self.allspice_client.requests_post( 2514 self.GET_REVIEWS.format( 2515 owner=self.repository.owner.username, 2516 repo=self.repository.name, 2517 index=self.number, 2518 ), 2519 data=data, 2520 ) 2521 2522 return DesignReviewReview.parse_response(self.allspice_client, result)
Create a review on this design review.
allspice.allspice.io/api/swagger#/repository/repoCreatePullRequestReview">https://huballspice.allspice.io/api/swagger#/repository/repoCreatePullRequestReview
Note: in most cases, you should not set the body or event when creating
a review. The event is automatically set to "PENDING" when the review
is created. You should then use submit_review to submit the review
with the desired event and body.
Parameters
- body: The body of the review. This is the top-level comment on the review. If not provided, the review will be created with no body.
- event: The event of the review. This is the overall status of the review. See the ReviewEvent enum. If not provided, the API will default to "PENDING".
- comments: A list of comments on the review. Each comment should be a ReviewComment object. If not provided, only the base comment will be created.
- commit_id: The commit SHA to associate with the review. This is optional.
2524 def get_reviews(self) -> List[DesignReviewReview]: 2525 """ 2526 Get all reviews on this design review. 2527 2528 https://hub.allspice.io/api/swagger#/repository/repoListPullReviews 2529 """ 2530 2531 results = self.allspice_client.requests_get( 2532 self.GET_REVIEWS.format( 2533 owner=self.repository.owner.username, 2534 repo=self.repository.name, 2535 index=self.number, 2536 ) 2537 ) 2538 2539 return [ 2540 DesignReviewReview.parse_response(self.allspice_client, result) for result in results 2541 ]
Get all reviews on this design review.
allspice.allspice.io/api/swagger#/repository/repoListPullReviews">https://huballspice.allspice.io/api/swagger#/repository/repoListPullReviews
def
submit_review( self, review_id: int, event: DesignReviewReview.ReviewEvent, *, body: Optional[str] = None):
2543 def submit_review( 2544 self, 2545 review_id: int, 2546 event: DesignReviewReview.ReviewEvent, 2547 *, 2548 body: Optional[str] = None, 2549 ): 2550 """ 2551 Submit a review on this design review. 2552 2553 https://hub.allspice.io/api/swagger#/repository/repoSubmitPullReview 2554 2555 :param review_id: The ID of the review to submit. 2556 :param event: The event to submit the review with. See the ReviewEvent 2557 enum for the possible values. 2558 :param body: Optional body text for the review submission. 2559 """ 2560 2561 data = { 2562 "event": event.value, 2563 } 2564 if body is not None: 2565 data["body"] = body 2566 2567 result = self.allspice_client.requests_post( 2568 self.GET_REVIEW.format( 2569 owner=self.repository.owner.username, 2570 repo=self.repository.name, 2571 index=self.number, 2572 review_id=review_id, 2573 ), 2574 data=data, 2575 ) 2576 2577 return result
Submit a review on this design review.
allspice.allspice.io/api/swagger#/repository/repoSubmitPullReview">https://huballspice.allspice.io/api/swagger#/repository/repoSubmitPullReview
Parameters
- review_id: The ID of the review to submit.
- event: The event to submit the review with. See the ReviewEvent enum for the possible values.
- body: Optional body text for the review submission.
class
DesignReview.MergeType(enum.Enum):
2580class Team(ApiObject): 2581 can_create_org_repo: bool 2582 description: str 2583 id: int 2584 includes_all_repositories: bool 2585 name: str 2586 organization: Optional["Organization"] 2587 permission: str 2588 units: List[str] 2589 units_map: Dict[str, str] 2590 2591 API_OBJECT = """/teams/{id}""" # <id> 2592 ADD_REPO = """/teams/%s/repos/%s/%s""" # <id, org, repo> 2593 TEAM_DELETE = """/teams/%s""" # <id> 2594 GET_MEMBERS = """/teams/%s/members""" # <id> 2595 GET_REPOS = """/teams/%s/repos""" # <id> 2596 2597 def __init__(self, allspice_client): 2598 super().__init__(allspice_client) 2599 2600 def __eq__(self, other): 2601 if not isinstance(other, Team): 2602 return False 2603 return self.organization == other.organization and self.id == other.id 2604 2605 def __hash__(self): 2606 return hash(self.organization) ^ hash(self.id) 2607 2608 _fields_to_parsers: ClassVar[dict] = { 2609 "organization": lambda allspice_client, o: Organization.parse_response(allspice_client, o) 2610 } 2611 2612 _patchable_fields: ClassVar[set[str]] = { 2613 "can_create_org_repo", 2614 "description", 2615 "includes_all_repositories", 2616 "name", 2617 "permission", 2618 "units", 2619 "units_map", 2620 } 2621 2622 @classmethod 2623 def request(cls, allspice_client, id: int): 2624 return cls._request(allspice_client, {"id": id}) 2625 2626 def commit(self): 2627 args = {"id": self.id} 2628 self._commit(args) 2629 2630 def add_user(self, user: User): 2631 """https://hub.allspice.io/api/swagger#/organization/orgAddTeamMember""" 2632 url = f"/teams/{self.id}/members/{user.login}" 2633 self.allspice_client.requests_put(url) 2634 2635 def add_repo(self, org: Organization, repo: Union[Repository, str]): 2636 if isinstance(repo, Repository): 2637 repo_name = repo.name 2638 else: 2639 repo_name = repo 2640 self.allspice_client.requests_put(Team.ADD_REPO % (self.id, org.username, repo_name)) 2641 2642 def get_members(self): 2643 """Get all users assigned to the team.""" 2644 results = self.allspice_client.requests_get_paginated( 2645 Team.GET_MEMBERS % self.id, 2646 ) 2647 return [User.parse_response(self.allspice_client, result) for result in results] 2648 2649 def get_repos(self): 2650 """Get all repos of this Team.""" 2651 results = self.allspice_client.requests_get(Team.GET_REPOS % self.id) 2652 return [Repository.parse_response(self.allspice_client, result) for result in results] 2653 2654 def delete(self): 2655 self.allspice_client.requests_delete(Team.TEAM_DELETE % self.id) 2656 self.deleted = True 2657 2658 def remove_team_member(self, user_name: str): 2659 url = f"/teams/{self.id}/members/{user_name}" 2660 self.allspice_client.requests_delete(url)
organization: Optional[Organization]
2630 def add_user(self, user: User): 2631 """https://hub.allspice.io/api/swagger#/organization/orgAddTeamMember""" 2632 url = f"/teams/{self.id}/members/{user.login}" 2633 self.allspice_client.requests_put(url)
allspice.allspice.io/api/swagger#/organization/orgAddTeamMember">https://huballspice.allspice.io/api/swagger#/organization/orgAddTeamMember
def
get_members(self):
2642 def get_members(self): 2643 """Get all users assigned to the team.""" 2644 results = self.allspice_client.requests_get_paginated( 2645 Team.GET_MEMBERS % self.id, 2646 ) 2647 return [User.parse_response(self.allspice_client, result) for result in results]
Get all users assigned to the team.
def
get_repos(self):
2649 def get_repos(self): 2650 """Get all repos of this Team.""" 2651 results = self.allspice_client.requests_get(Team.GET_REPOS % self.id) 2652 return [Repository.parse_response(self.allspice_client, result) for result in results]
Get all repos of this Team.
2663class Release(ApiObject): 2664 """ 2665 A release on a repo. 2666 """ 2667 2668 assets: List[Union[Any, Dict[str, Union[int, str]], "ReleaseAsset"]] 2669 author: User 2670 body: str 2671 created_at: str 2672 draft: bool 2673 html_url: str 2674 id: int 2675 name: str 2676 prerelease: bool 2677 published_at: str 2678 repo: Optional["Repository"] 2679 repository: Optional["Repository"] 2680 tag_name: str 2681 tarball_url: str 2682 target_commitish: str 2683 upload_url: str 2684 url: str 2685 zipball_url: str 2686 2687 API_OBJECT = "/repos/{owner}/{repo}/releases/{id}" 2688 RELEASE_CREATE_ASSET = "/repos/{owner}/{repo}/releases/{id}/assets" 2689 # Note that we don't strictly need the get_assets route, as the release 2690 # object already contains the assets. 2691 2692 def __init__(self, allspice_client): 2693 super().__init__(allspice_client) 2694 2695 def __eq__(self, other): 2696 if not isinstance(other, Release): 2697 return False 2698 return self.repo == other.repo and self.id == other.id 2699 2700 def __hash__(self): 2701 return hash(self.repo) ^ hash(self.id) 2702 2703 _fields_to_parsers: ClassVar[dict] = { 2704 "author": lambda allspice_client, author: User.parse_response(allspice_client, author), 2705 } 2706 _patchable_fields: ClassVar[set[str]] = { 2707 "body", 2708 "draft", 2709 "name", 2710 "prerelease", 2711 "tag_name", 2712 "target_commitish", 2713 } 2714 2715 @classmethod 2716 def parse_response(cls, allspice_client, result, repo) -> Release: 2717 release = super().parse_response(allspice_client, result) 2718 Release._add_read_property("repository", repo, release) 2719 # For legacy reasons 2720 Release._add_read_property("repo", repo, release) 2721 setattr( 2722 release, 2723 "_assets", 2724 [ 2725 ReleaseAsset.parse_response(allspice_client, asset, release) 2726 for asset in result["assets"] 2727 ], 2728 ) 2729 return release 2730 2731 @classmethod 2732 def request( 2733 cls, 2734 allspice_client, 2735 owner: str, 2736 repo: str, 2737 id: Optional[int] = None, 2738 ) -> Release: 2739 args = {"owner": owner, "repo": repo, "id": id} 2740 release_response = cls._get_gitea_api_object(allspice_client, args) 2741 repository = Repository.request(allspice_client, owner, repo) 2742 release = cls.parse_response(allspice_client, release_response, repository) 2743 return release 2744 2745 def commit(self): 2746 if self.repo is None: 2747 raise ValueError("Cannot commit a release without a repository.") 2748 2749 args = {"owner": self.repo.owner.username, "repo": self.repo.name, "id": self.id} 2750 self._commit(args) 2751 2752 def create_asset(self, file: IO, name: Optional[str] = None) -> ReleaseAsset: 2753 """ 2754 Create an asset for this release. 2755 2756 https://hub.allspice.io/api/swagger#/repository/repoCreateReleaseAsset 2757 2758 :param file: The file to upload. This should be a file-like object. 2759 :param name: The name of the file. 2760 :return: The created asset. 2761 """ 2762 2763 if self.repo is None: 2764 raise ValueError("Cannot commit a release without a repository.") 2765 2766 args: dict[str, Any] = {"files": {"attachment": file}} 2767 if name is not None: 2768 args["params"] = {"name": name} 2769 2770 result = self.allspice_client.requests_post( 2771 self.RELEASE_CREATE_ASSET.format( 2772 owner=self.repo.owner.username, 2773 repo=self.repo.name, 2774 id=self.id, 2775 ), 2776 **args, 2777 ) 2778 return ReleaseAsset.parse_response(self.allspice_client, result, self) 2779 2780 def delete(self): 2781 if self.repo is None: 2782 raise ValueError("Cannot commit a release without a repository.") 2783 2784 args = {"owner": self.repo.owner.username, "repo": self.repo.name, "id": self.id} 2785 self.allspice_client.requests_delete(self.API_OBJECT.format(**args)) 2786 self.deleted = True
A release on a repo.
assets: List[Union[Any, Dict[str, Union[int, str]], ReleaseAsset]]
repo: Optional[Repository]
repository: Optional[Repository]
2715 @classmethod 2716 def parse_response(cls, allspice_client, result, repo) -> Release: 2717 release = super().parse_response(allspice_client, result) 2718 Release._add_read_property("repository", repo, release) 2719 # For legacy reasons 2720 Release._add_read_property("repo", repo, release) 2721 setattr( 2722 release, 2723 "_assets", 2724 [ 2725 ReleaseAsset.parse_response(allspice_client, asset, release) 2726 for asset in result["assets"] 2727 ], 2728 ) 2729 return release
@classmethod
def
request( cls, allspice_client, owner: str, repo: str, id: Optional[int] = None) -> Release:
2731 @classmethod 2732 def request( 2733 cls, 2734 allspice_client, 2735 owner: str, 2736 repo: str, 2737 id: Optional[int] = None, 2738 ) -> Release: 2739 args = {"owner": owner, "repo": repo, "id": id} 2740 release_response = cls._get_gitea_api_object(allspice_client, args) 2741 repository = Repository.request(allspice_client, owner, repo) 2742 release = cls.parse_response(allspice_client, release_response, repository) 2743 return release
2752 def create_asset(self, file: IO, name: Optional[str] = None) -> ReleaseAsset: 2753 """ 2754 Create an asset for this release. 2755 2756 https://hub.allspice.io/api/swagger#/repository/repoCreateReleaseAsset 2757 2758 :param file: The file to upload. This should be a file-like object. 2759 :param name: The name of the file. 2760 :return: The created asset. 2761 """ 2762 2763 if self.repo is None: 2764 raise ValueError("Cannot commit a release without a repository.") 2765 2766 args: dict[str, Any] = {"files": {"attachment": file}} 2767 if name is not None: 2768 args["params"] = {"name": name} 2769 2770 result = self.allspice_client.requests_post( 2771 self.RELEASE_CREATE_ASSET.format( 2772 owner=self.repo.owner.username, 2773 repo=self.repo.name, 2774 id=self.id, 2775 ), 2776 **args, 2777 ) 2778 return ReleaseAsset.parse_response(self.allspice_client, result, self)
Create an asset for this release.
allspice.allspice.io/api/swagger#/repository/repoCreateReleaseAsset">https://huballspice.allspice.io/api/swagger#/repository/repoCreateReleaseAsset
Parameters
- file: The file to upload. This should be a file-like object.
- name: The name of the file.
Returns
The created asset.
def
delete(self):
2780 def delete(self): 2781 if self.repo is None: 2782 raise ValueError("Cannot commit a release without a repository.") 2783 2784 args = {"owner": self.repo.owner.username, "repo": self.repo.name, "id": self.id} 2785 self.allspice_client.requests_delete(self.API_OBJECT.format(**args)) 2786 self.deleted = True
2789class ReleaseAsset(ApiObject): 2790 browser_download_url: str 2791 created_at: str 2792 download_count: int 2793 id: int 2794 name: str 2795 release: Optional["Release"] 2796 size: int 2797 uuid: str 2798 2799 API_OBJECT = "/repos/{owner}/{repo}/releases/{release_id}/assets/{id}" 2800 2801 def __init__(self, allspice_client): 2802 super().__init__(allspice_client) 2803 2804 def __eq__(self, other): 2805 if not isinstance(other, ReleaseAsset): 2806 return False 2807 return self.release == other.release and self.id == other.id 2808 2809 def __hash__(self): 2810 return hash(self.release) ^ hash(self.id) 2811 2812 _fields_to_parsers: ClassVar[dict] = {} 2813 _patchable_fields: ClassVar[set[str]] = { 2814 "name", 2815 } 2816 2817 @classmethod 2818 def parse_response(cls, allspice_client, result, release) -> ReleaseAsset: 2819 asset = super().parse_response(allspice_client, result) 2820 ReleaseAsset._add_read_property("release", release, asset) 2821 return asset 2822 2823 @classmethod 2824 def request( 2825 cls, 2826 allspice_client, 2827 owner: str, 2828 repo: str, 2829 release_id: int, 2830 id: int, 2831 ) -> ReleaseAsset: 2832 args = {"owner": owner, "repo": repo, "release_id": release_id, "id": id} 2833 asset_response = cls._get_gitea_api_object(allspice_client, args) 2834 release = Release.request(allspice_client, owner, repo, release_id) 2835 asset = cls.parse_response(allspice_client, asset_response, release) 2836 return asset 2837 2838 def commit(self): 2839 if self.release is None or self.release.repo is None: 2840 raise ValueError("Cannot commit a release asset without a release or a repository.") 2841 2842 args = { 2843 "owner": self.release.repo.owner.username, 2844 "repo": self.release.repo.name, 2845 "release_id": self.release.id, 2846 "id": self.id, 2847 } 2848 self._commit(args) 2849 2850 def download(self) -> bytes: 2851 """ 2852 Download the raw, binary data of this asset. 2853 2854 Note 1: if the file you are requesting is a text file, you might want to 2855 use .decode() on the result to get a string. For example: 2856 2857 asset.download().decode("utf-8") 2858 2859 Note 2: this method will store the entire file in memory. If you are 2860 downloading a large file, you might want to use download_to_file instead. 2861 """ 2862 2863 return self.allspice_client.requests.get( 2864 self.browser_download_url, 2865 headers=self.allspice_client.headers, 2866 ).content 2867 2868 def download_to_file(self, io: IO): 2869 """ 2870 Download the raw, binary data of this asset to a file-like object. 2871 2872 Example: 2873 2874 with open("my_file.zip", "wb") as f: 2875 asset.download_to_file(f) 2876 2877 :param io: The file-like object to write the data to. 2878 """ 2879 2880 response = self.allspice_client.requests.get( 2881 self.browser_download_url, 2882 headers=self.allspice_client.headers, 2883 stream=True, 2884 ) 2885 # 4kb chunks 2886 for chunk in response.iter_content(chunk_size=4096): 2887 if chunk: 2888 io.write(chunk) 2889 2890 def delete(self): 2891 if self.release is None or self.release.repo is None: 2892 raise ValueError("Cannot commit a release asset without a release or a repository.") 2893 2894 args = { 2895 "owner": self.release.repo.owner.username, 2896 "repo": self.release.repo.name, 2897 "release_id": self.release.id, 2898 "id": self.id, 2899 } 2900 self.allspice_client.requests_delete(self.API_OBJECT.format(**args)) 2901 self.deleted = True
release: Optional[Release]
@classmethod
def
request( cls, allspice_client, owner: str, repo: str, release_id: int, id: int) -> ReleaseAsset:
2823 @classmethod 2824 def request( 2825 cls, 2826 allspice_client, 2827 owner: str, 2828 repo: str, 2829 release_id: int, 2830 id: int, 2831 ) -> ReleaseAsset: 2832 args = {"owner": owner, "repo": repo, "release_id": release_id, "id": id} 2833 asset_response = cls._get_gitea_api_object(allspice_client, args) 2834 release = Release.request(allspice_client, owner, repo, release_id) 2835 asset = cls.parse_response(allspice_client, asset_response, release) 2836 return asset
def
commit(self):
2838 def commit(self): 2839 if self.release is None or self.release.repo is None: 2840 raise ValueError("Cannot commit a release asset without a release or a repository.") 2841 2842 args = { 2843 "owner": self.release.repo.owner.username, 2844 "repo": self.release.repo.name, 2845 "release_id": self.release.id, 2846 "id": self.id, 2847 } 2848 self._commit(args)
def
download(self) -> bytes:
2850 def download(self) -> bytes: 2851 """ 2852 Download the raw, binary data of this asset. 2853 2854 Note 1: if the file you are requesting is a text file, you might want to 2855 use .decode() on the result to get a string. For example: 2856 2857 asset.download().decode("utf-8") 2858 2859 Note 2: this method will store the entire file in memory. If you are 2860 downloading a large file, you might want to use download_to_file instead. 2861 """ 2862 2863 return self.allspice_client.requests.get( 2864 self.browser_download_url, 2865 headers=self.allspice_client.headers, 2866 ).content
Download the raw, binary data of this asset.
Note 1: if the file you are requesting is a text file, you might want to use .decode() on the result to get a string. For example:
asset.download().decode("utf-8")
Note 2: this method will store the entire file in memory. If you are downloading a large file, you might want to use download_to_file instead.
def
download_to_file(self, io: <class 'IO'>):
2868 def download_to_file(self, io: IO): 2869 """ 2870 Download the raw, binary data of this asset to a file-like object. 2871 2872 Example: 2873 2874 with open("my_file.zip", "wb") as f: 2875 asset.download_to_file(f) 2876 2877 :param io: The file-like object to write the data to. 2878 """ 2879 2880 response = self.allspice_client.requests.get( 2881 self.browser_download_url, 2882 headers=self.allspice_client.headers, 2883 stream=True, 2884 ) 2885 # 4kb chunks 2886 for chunk in response.iter_content(chunk_size=4096): 2887 if chunk: 2888 io.write(chunk)
Download the raw, binary data of this asset to a file-like object.
Example:
with open("my_file.zip", "wb") as f:
asset.download_to_file(f)
Parameters
- io: The file-like object to write the data to.
def
delete(self):
2890 def delete(self): 2891 if self.release is None or self.release.repo is None: 2892 raise ValueError("Cannot commit a release asset without a release or a repository.") 2893 2894 args = { 2895 "owner": self.release.repo.owner.username, 2896 "repo": self.release.repo.name, 2897 "release_id": self.release.id, 2898 "id": self.id, 2899 } 2900 self.allspice_client.requests_delete(self.API_OBJECT.format(**args)) 2901 self.deleted = True
2904class Content(ReadonlyApiObject): 2905 content: Any 2906 download_url: str 2907 encoding: Any 2908 git_url: str 2909 html_url: str 2910 last_commit_sha: str 2911 name: str 2912 path: str 2913 sha: str 2914 size: int 2915 submodule_git_url: Any 2916 target: Any 2917 type: str 2918 url: str 2919 2920 FILE = "file" 2921 2922 def __init__(self, allspice_client): 2923 super().__init__(allspice_client) 2924 2925 def __eq__(self, other): 2926 if not isinstance(other, Content): 2927 return False 2928 2929 return self.sha == other.sha and self.name == other.name 2930 2931 def __hash__(self): 2932 return hash(self.sha) ^ hash(self.name)
Inherited Members
class
Util:
2938class Util: 2939 @staticmethod 2940 def convert_time(time: str) -> datetime: 2941 """Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" 2942 try: 2943 return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") 2944 except ValueError: 2945 return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S") 2946 2947 @staticmethod 2948 def format_time(time: datetime) -> str: 2949 """ 2950 Format a datetime object to Gitea's time format. 2951 2952 :param time: The time to format 2953 :return: Formatted time 2954 """ 2955 2956 return time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + "Z" 2957 2958 @staticmethod 2959 def data_params_for_ref(ref: Optional[Ref]) -> Dict: 2960 """ 2961 Given a "ref", returns a dict with the ref parameter for the API call. 2962 2963 If the ref is None, returns an empty dict. You can pass this to the API 2964 directly. 2965 """ 2966 2967 if isinstance(ref, Branch): 2968 return {"ref": ref.name} 2969 elif isinstance(ref, Commit): 2970 return {"ref": ref.sha} 2971 elif ref: 2972 return {"ref": ref} 2973 else: 2974 return {}
@staticmethod
def
convert_time(time: str) -> datetime.datetime:
2939 @staticmethod 2940 def convert_time(time: str) -> datetime: 2941 """Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)""" 2942 try: 2943 return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S%z") 2944 except ValueError: 2945 return datetime.strptime(time[:-3] + "00", "%Y-%m-%dT%H:%M:%S")
Parsing of strange Gitea time format ("%Y-%m-%dT%H:%M:%S:%z" but with ":" in time zone notation)
@staticmethod
def
format_time(time: datetime.datetime) -> str:
2947 @staticmethod 2948 def format_time(time: datetime) -> str: 2949 """ 2950 Format a datetime object to Gitea's time format. 2951 2952 :param time: The time to format 2953 :return: Formatted time 2954 """ 2955 2956 return time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + "Z"
Format a datetime object to Gitea's time format.
Parameters
- time: The time to format
Returns
Formatted time
2958 @staticmethod 2959 def data_params_for_ref(ref: Optional[Ref]) -> Dict: 2960 """ 2961 Given a "ref", returns a dict with the ref parameter for the API call. 2962 2963 If the ref is None, returns an empty dict. You can pass this to the API 2964 directly. 2965 """ 2966 2967 if isinstance(ref, Branch): 2968 return {"ref": ref.name} 2969 elif isinstance(ref, Commit): 2970 return {"ref": ref.sha} 2971 elif ref: 2972 return {"ref": ref} 2973 else: 2974 return {}
Given a "ref", returns a dict with the ref parameter for the API call.
If the ref is None, returns an empty dict. You can pass this to the API directly.