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