allspice

A very simple API client for AllSpice Hub

Note that not the full Swagger-API is accessible. The whole implementation is focused on making access and working with Organizations, Teams, Repositories and Users as pain free as possible.

Forked from https://github.com/Langenfeld/py-gitea.

Usage

Docs

See the documentation site.

Examples

Check the examples directory for full, working example scripts that you can adapt or refer to for your own needs.

Quickstart

First get an allspice_client object wrapping access and authentication (via an api token) for your instance of AllSpice Hub.

from allspice import *

# By default, points to hub.allspice.io.
allspice_client = AllSpice(token_text=TOKEN)

# If you are self-hosting:
allspice_client = AllSpice(allspice_hub_url=URL, token_text=TOKEN)

Operations like requesting the AllSpice version or authentication user can be requested directly from the allspice_client object:

print("AllSpice Version: " + allspice_client.get_version())
print("API-Token belongs to user: " + allspice_client.get_user().username)

Adding entities like Users, Organizations, ... also is done via the allspice_client object.

user = allspice_client.create_user("Test Testson", "test@test.test", "password")

All operations on entities in allspice are then accomplished via the according wrapper objects for those entities. Each of those objects has a .request method that creates an entity according to your allspice_client instance.

other_user = User.request(allspice_client, "OtherUserName")
print(other_user.username)

Note that the fields of the User, Organization,... classes are dynamically created at runtime, and thus not visible during divelopment. Refer to the AllSpice API documentation for the fields names.

Fields that can not be altered via allspice-api, are read only. After altering a field, the .commit method of the according object must be called to synchronize the changed fields with your allspice_client instance.

org = Organization.request(allspice_client, test_org)
org.description = "some new description"
org.location = "some new location"
org.commit()

An entity in allspice can be deleted by calling delete.

org.delete()

All entity objects do have methods to execute some of the requests possible though the AllSpice api:

org = Organization.request(allspice_client, ORGNAME)
teams = org.get_teams()
for team in teams:
    repos = team.get_repos()
    for repo in repos:
        print(repo.name)

View Source

 1"""
 2.. include:: ../README.md
 3   :start-line: 1
 4   :end-before: Installation
 5"""
 6
 7from .allspice import (
 8    AllSpice,
 9)
10from .apiobject import (
11    Branch,
12    Comment,
13    Commit,
14    Content,
15    DesignReview,
16    DesignReviewReview,
17    Issue,
18    Milestone,
19    Organization,
20    Release,
21    Repository,
22    Team,
23    User,
24)
25from .exceptions import (
26    AlreadyExistsException,
27    APIError,
28    InternalServerException,
29    NotFoundException,
30    RenderException,
31)
32
33__version__ = "4.3.0"
34
35__all__ = [
36    "APIError",
37    "AllSpice",
38    "AlreadyExistsException",
39    "Branch",
40    "Comment",
41    "Commit",
42    "Content",
43    "DesignReview",
44    "DesignReviewReview",
45    "InternalServerException",
46    "Issue",
47    "Milestone",
48    "NotFoundException",
49    "Organization",
50    "Release",
51    "RenderException",
52    "Repository",
53    "Team",
54    "User",
55]

@dataclass(frozen=True)

class APIError: View Source

 7@dataclass(frozen=True)
 8class APIError:
 9    """The body of an AllSpice Hub error response"""
10
11    message: Optional[str]
12    url: Optional[str]
13    category: Optional[str]
14
15    @classmethod
16    def from_json(cls, text: str) -> Optional["APIError"]:
17        try:
18            data = json.loads(text)
19        except json.JSONDecodeError:
20            return None
21
22        if not isinstance(data, dict):
23            return None
24
25        message = data.get("message") or None
26        url = data.get("url") or None
27        category = data.get("category") or None
28
29        if message is None and url is None and category is None:
30            return None
31
32        return cls(message=message, url=url, category=category)

The body of an AllSpice Hub error response

APIError(message: Optional[str], url: Optional[str], category: Optional[str])

message: Optional[str]

url: Optional[str]

category: Optional[str]

@classmethod

def from_json(cls, text: str) -> Optional[APIError]: View Source

15    @classmethod
16    def from_json(cls, text: str) -> Optional["APIError"]:
17        try:
18            data = json.loads(text)
19        except json.JSONDecodeError:
20            return None
21
22        if not isinstance(data, dict):
23            return None
24
25        message = data.get("message") or None
26        url = data.get("url") or None
27        category = data.get("category") or None
28
29        if message is None and url is None and category is None:
30            return None
31
32        return cls(message=message, url=url, category=category)

class AllSpice: View Source

 43class AllSpice:
 44    """Object to establish a session with AllSpice Hub."""
 45
 46    ADMIN_CREATE_USER = """/admin/users"""
 47    GET_USERS_ADMIN = """/admin/users"""
 48    ADMIN_REPO_CREATE = """/admin/users/%s/repos"""  # <ownername>
 49    ALLSPICE_HUB_VERSION = """/version"""
 50    GET_USER = """/user"""
 51    GET_REPOSITORY = """/repos/{owner}/{name}"""
 52    CREATE_ORG = """/admin/users/%s/orgs"""  # <username>
 53    CREATE_TEAM = """/orgs/%s/teams"""  # <orgname>
 54
 55    def __init__(
 56        self,
 57        allspice_hub_url="https://hub.allspice.io",
 58        token_text=None,
 59        auth=None,
 60        verify=True,
 61        log_level="INFO",
 62        ratelimiting=(100, 60),
 63        retry: Union[Retry, int, None] = DEFAULT_RETRY,
 64        use_new_schdoc_renderer: Optional[bool] = None,
 65    ):
 66        """Initializing an instance of the AllSpice Hub Client
 67
 68        Args:
 69            allspice_hub_url (str): The URL for the AllSpice Hub instance.
 70                Defaults to `https://hub.allspice.io`.
 71
 72            token_text (str, None): The access token, by default None.
 73
 74            auth (tuple, None): The user credentials
 75                `(username, password)`, by default None.
 76
 77            verify (bool): If True, allow insecure server connections
 78                when using SSL.
 79
 80            log_level (str): The log level, by default `INFO`.
 81
 82            ratelimiting (tuple[int, int], None): `(max_calls, period)`,
 83                If None, no rate limiting is applied. By default, 100 calls
 84                per minute are allowed.
 85
 86            retry (Retry, int, None): Set a retry policy for requests using a
 87                urllib3 Retry object, an integer for the number of retries, or None
 88                for no retries. This happens before py-allspice's own rate limiter.
 89                The default is to retry all methods (even mutating methods)
 90                only on 429s and 502s up to 6 times with exponential backoff
 91                and jitter.
 92
 93                NOTE: AllSpice Hub responds with a 503 and a Retry-After header
 94                when a generated file is not ready yet, and urllib3 retries any
 95                503 that carries Retry-After. So with retries enabled, fetching
 96                generated files transparently waits for generation instead of
 97                raising NotYetGeneratedException, until retries are exhausted.
 98
 99            use_new_schdoc_renderer (bool): Allows explicit override for using the new Altium schematic renderer. If set,
100            this will take precedence over the default behavior on the AllSpice Hub instance.
101        """
102
103        self.logger = logging.getLogger(__name__)
104        handler = logging.StreamHandler(sys.stderr)
105        handler.setFormatter(
106            logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
107        )
108        self.logger.addHandler(handler)
109        self.logger.setLevel(log_level)
110        self.headers = {
111            "Content-type": "application/json",
112        }
113        self.url = allspice_hub_url
114
115        if ratelimiting is None:
116            self.requests = requests.Session()
117        else:
118            (max_calls, period) = ratelimiting
119            self.requests = RateLimitedSession(max_calls=max_calls, period=period)
120
121        if retry is not None:
122            adapter = HTTPAdapter(max_retries=retry)
123            self.requests.mount("https://", adapter)
124            self.requests.mount("http://", adapter)
125
126        # Manage authentification
127        if not token_text and not auth:
128            raise ValueError("Please provide auth or token_text, but not both")
129        if token_text:
130            self.headers["Authorization"] = "token " + token_text
131        if auth:
132            self.logger.warning(
133                "Using basic auth is not recommended. Prefer using a token instead."
134            )
135            self.requests.auth = auth
136
137        # Manage SSL certification verification
138        self.requests.verify = verify
139        if not verify:
140            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
141
142        self.use_new_schdoc_renderer = use_new_schdoc_renderer
143
144    def __get_url(self, endpoint):
145        url = self.url + "/api/v1" + endpoint
146        self.logger.debug("Url: %s" % url)
147        return url
148
149    def __get(self, endpoint: str, params: Mapping = frozendict()) -> requests.Response:
150        response = self.requests.get(self.__get_url(endpoint), headers=self.headers, params=params)
151        if response.status_code not in [200, 201]:
152            message = f"Received status code: {response.status_code} ({response.url})"
153            if response.status_code in [404]:
154                raise NotFoundException(message)
155            if response.status_code in [403]:
156                raise Exception(
157                    f"Unauthorized: {response.url} - Check your permissions and try again! ({message})"
158                )
159            if response.status_code in [409]:
160                raise ConflictException(message)
161            if response.status_code in [503]:
162                raise NotYetGeneratedException(message)
163            if response.status_code in [500]:
164                raise InternalServerException(message, APIError.from_json(response.text))
165            raise Exception(message)
166        return response
167
168    @staticmethod
169    def parse_result(result) -> Dict:
170        """Parses the result-JSON to a dict."""
171        if result.text and len(result.text) > 3:
172            return json.loads(result.text)
173        return {}
174
175    def requests_get(self, endpoint: str, params: Mapping = frozendict(), sudo=None):
176        combined_params = {}
177        combined_params.update(params)
178        if sudo:
179            combined_params["sudo"] = sudo.username
180        return self.parse_result(self.__get(endpoint, combined_params))
181
182    def requests_get_raw(self, endpoint: str, params=frozendict(), sudo=None) -> bytes:
183        combined_params = {}
184        combined_params.update(params)
185        if sudo:
186            combined_params["sudo"] = sudo.username
187        return self.__get(endpoint, combined_params).content
188
189    def requests_get_paginated(
190        self,
191        endpoint: str,
192        params=frozendict(),
193        sudo=None,
194        page_key: str = "page",
195        first_page: int = 1,
196    ):
197        page = first_page
198        combined_params = {}
199        combined_params.update(params)
200        aggregated_result = []
201        while True:
202            combined_params[page_key] = page
203            result = self.requests_get(endpoint, combined_params, sudo)
204
205            if not result:
206                return aggregated_result
207
208            if isinstance(result, dict):
209                if "data" in result:
210                    data = result["data"]
211                    if len(data) == 0:
212                        return aggregated_result
213                    aggregated_result.extend(data)
214                elif "tree" in result:
215                    data = result["tree"]
216                    if data is None or len(data) == 0:
217                        return aggregated_result
218                    aggregated_result.extend(data)
219                else:
220                    raise NotImplementedError(
221                        "requests_get_paginated does not know how to handle responses of this type."
222                    )
223            else:
224                aggregated_result.extend(result)
225
226            page += 1
227
228    def requests_put(self, endpoint: str, data: Optional[dict] = None):
229        if not data:
230            data = {}
231        response = self.requests.put(
232            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
233        )
234        if response.status_code not in [200, 204]:
235            message = (
236                f"Received status code: {response.status_code} ({response.url}) {response.text}"
237            )
238            self.logger.error(message)
239            raise Exception(message)
240
241    def requests_delete(self, endpoint: str, data: Optional[dict] = None):
242        response = self.requests.delete(
243            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
244        )
245        if response.status_code not in [200, 204]:
246            message = f"Received status code: {response.status_code} ({response.url})"
247            self.logger.error(message)
248            raise Exception(message)
249
250    def requests_post(
251        self,
252        endpoint: str,
253        data: Optional[dict] = None,
254        params: Optional[dict] = None,
255        files: Optional[dict] = None,
256    ):
257        """
258        Make a POST call to the endpoint.
259
260        :param endpoint: The path to the endpoint
261        :param data: A dictionary for JSON data
262        :param params: A dictionary of query params
263        :param files: A dictionary of files, see requests.post. Using both files and data
264                      can lead to unexpected results!
265        :return: The JSON response parsed as a dict
266        """
267
268        # This should ideally be a TypedDict of the type of arguments taken by
269        # `requests.post`.
270        args: dict[str, Any] = {
271            "headers": self.headers.copy(),
272        }
273        if data is not None:
274            args["data"] = json.dumps(data)
275        if params is not None:
276            args["params"] = params
277        if files is not None:
278            args["headers"].pop("Content-type")
279            args["files"] = files
280
281        response = self.requests.post(self.__get_url(endpoint), **args)
282
283        if response.status_code not in [200, 201, 202]:
284            if "already exists" in response.text or "e-mail already in use" in response.text:
285                self.logger.warning(response.text)
286                raise AlreadyExistsException()
287            self.logger.error(f"Received status code: {response.status_code} ({response.url})")
288            self.logger.error(f"With info: {data} ({self.headers})")
289            self.logger.error(f"Answer: {response.text}")
290            raise Exception(
291                f"Received status code: {response.status_code} ({response.url}), {response.text}"
292            )
293        return self.parse_result(response)
294
295    def requests_patch(self, endpoint: str, data: dict):
296        response = self.requests.patch(
297            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
298        )
299        if response.status_code not in [200, 201]:
300            error_message = f"Received status code: {response.status_code} ({response.url}) {data}"
301            self.logger.error(error_message)
302            raise Exception(error_message)
303        return self.parse_result(response)
304
305    def get_orgs_public_members_all(self, orgname):
306        path = "/orgs/" + orgname + "/public_members"
307        return self.requests_get(path)
308
309    def get_orgs(self):
310        path = "/admin/orgs"
311        results = self.requests_get(path)
312        return [Organization.parse_response(self, result) for result in results]
313
314    def get_user(self):
315        result = self.requests_get(AllSpice.GET_USER)
316        return User.parse_response(self, result)
317
318    def get_version(self) -> str:
319        result = self.requests_get(AllSpice.ALLSPICE_HUB_VERSION)
320        return result["version"]
321
322    def get_users(self) -> List[User]:
323        results = self.requests_get(AllSpice.GET_USERS_ADMIN)
324        return [User.parse_response(self, result) for result in results]
325
326    def get_user_by_email(self, email: str) -> Optional[User]:
327        users = self.get_users()
328        for user in users:
329            if user.email == email or email in user.emails:
330                return user
331        return None
332
333    def get_user_by_name(self, username: str) -> Optional[User]:
334        users = self.get_users()
335        for user in users:
336            if user.username == username:
337                return user
338        return None
339
340    def get_repository(self, owner: str, name: str) -> Repository:
341        path = self.GET_REPOSITORY.format(owner=owner, name=name)
342        result = self.requests_get(path)
343        return Repository.parse_response(self, result)
344
345    def create_user(
346        self,
347        user_name: str,
348        email: str,
349        password: str,
350        full_name: Optional[str] = None,
351        login_name: Optional[str] = None,
352        change_pw=True,
353        send_notify=True,
354        source_id=0,
355    ):
356        """Create User.
357        Throws:
358            AlreadyExistsException, if the User exists already
359            Exception, if something else went wrong.
360        """
361        if not login_name:
362            login_name = user_name
363        if not full_name:
364            full_name = user_name
365        request_data = {
366            "source_id": source_id,
367            "login_name": login_name,
368            "full_name": full_name,
369            "username": user_name,
370            "email": email,
371            "password": password,
372            "send_notify": send_notify,
373            "must_change_password": change_pw,
374        }
375
376        self.logger.debug("Gitea post payload: %s", request_data)
377        result = self.requests_post(AllSpice.ADMIN_CREATE_USER, data=request_data)
378        if "id" in result:
379            self.logger.info(
380                "Successfully created User %s <%s> (id %s)",
381                result["login"],
382                result["email"],
383                result["id"],
384            )
385            self.logger.debug("Gitea response: %s", result)
386        else:
387            self.logger.error(result["message"])
388            raise Exception("User not created... (gitea: %s)" % result["message"])
389        user = User.parse_response(self, result)
390        return user
391
392    def create_repo(
393        self,
394        repoOwner: Union[User, Organization],
395        repoName: str,
396        description: str = "",
397        private: bool = False,
398        autoInit=True,
399        gitignores: Optional[str] = None,
400        license: Optional[str] = None,
401        readme: str = "Default",
402        issue_labels: Optional[str] = None,
403        default_branch="master",
404    ):
405        """Create a Repository as the administrator
406
407        Throws:
408            AlreadyExistsException: If the Repository exists already.
409            Exception: If something else went wrong.
410
411        Note:
412            Non-admin users can not use this method. Please use instead
413            `allspice.User.create_repo` or `allspice.Organization.create_repo`.
414        """
415        # although this only says user in the api, this also works for
416        # organizations
417        assert isinstance(repoOwner, User) or isinstance(repoOwner, Organization)
418        result = self.requests_post(
419            AllSpice.ADMIN_REPO_CREATE % repoOwner.username,
420            data={
421                "name": repoName,
422                "description": description,
423                "private": private,
424                "auto_init": autoInit,
425                "gitignores": gitignores,
426                "license": license,
427                "issue_labels": issue_labels,
428                "readme": readme,
429                "default_branch": default_branch,
430            },
431        )
432        if "id" in result:
433            self.logger.info("Successfully created Repository %s " % result["name"])
434        else:
435            self.logger.error(result["message"])
436            raise Exception("Repository not created... (gitea: %s)" % result["message"])
437        return Repository.parse_response(self, result)
438
439    def create_org(
440        self,
441        owner: User,
442        orgName: str,
443        description: str,
444        location="",
445        website="",
446        full_name="",
447    ):
448        assert isinstance(owner, User)
449        result = self.requests_post(
450            AllSpice.CREATE_ORG % owner.username,
451            data={
452                "username": orgName,
453                "description": description,
454                "location": location,
455                "website": website,
456                "full_name": full_name,
457            },
458        )
459        if "id" in result:
460            self.logger.info("Successfully created Organization %s" % result["username"])
461        else:
462            self.logger.error("Organization not created... (gitea: %s)" % result["message"])
463            self.logger.error(result["message"])
464            raise Exception("Organization not created... (gitea: %s)" % result["message"])
465        return Organization.parse_response(self, result)
466
467    def create_team(
468        self,
469        org: Organization,
470        name: str,
471        description: str = "",
472        permission: str = "read",
473        can_create_org_repo: bool = False,
474        includes_all_repositories: bool = False,
475        units=(
476            "repo.code",
477            "repo.issues",
478            "repo.ext_issues",
479            "repo.wiki",
480            "repo.pulls",
481            "repo.releases",
482            "repo.ext_wiki",
483        ),
484        units_map={},
485    ):
486        """Creates a Team.
487
488        Args:
489            org (Organization): Organization the Team will be part of.
490            name (str): The Name of the Team to be created.
491            description (str): Optional, None, short description of the new Team.
492            permission (str): Optional, 'read', What permissions the members
493            units_map (dict): Optional, {}, a mapping of units to their
494                permissions. If None or empty, the `permission` permission will
495                be applied to all units. Note: When both `units` and `units_map`
496                are given, `units_map` will be preferred.
497        """
498
499        result = self.requests_post(
500            AllSpice.CREATE_TEAM % org.username,
501            data={
502                "name": name,
503                "description": description,
504                "permission": permission,
505                "can_create_org_repo": can_create_org_repo,
506                "includes_all_repositories": includes_all_repositories,
507                "units": units,
508                "units_map": units_map,
509            },
510        )
511
512        if "id" in result:
513            self.logger.info("Successfully created Team %s" % result["name"])
514        else:
515            self.logger.error("Team not created... (gitea: %s)" % result["message"])
516            self.logger.error(result["message"])
517            raise Exception("Team not created... (gitea: %s)" % result["message"])
518        api_object = Team.parse_response(self, result)
519        setattr(
520            api_object, "_organization", org
521        )  # fixes strange behaviour of gitea not returning a valid organization here.
522        return api_object

Object to establish a session with AllSpice Hub.

AllSpice( allspice_hub_url='https://hub.allspice.io', token_text=None, auth=None, verify=True, log_level='INFO', ratelimiting=(100, 60), retry: Union[urllib3.util.retry.Retry, int, NoneType] = Retry(total=6, connect=None, read=None, redirect=None, status=None), use_new_schdoc_renderer: Optional[bool] = None) View Source

 55    def __init__(
 56        self,
 57        allspice_hub_url="https://hub.allspice.io",
 58        token_text=None,
 59        auth=None,
 60        verify=True,
 61        log_level="INFO",
 62        ratelimiting=(100, 60),
 63        retry: Union[Retry, int, None] = DEFAULT_RETRY,
 64        use_new_schdoc_renderer: Optional[bool] = None,
 65    ):
 66        """Initializing an instance of the AllSpice Hub Client
 67
 68        Args:
 69            allspice_hub_url (str): The URL for the AllSpice Hub instance.
 70                Defaults to `https://hub.allspice.io`.
 71
 72            token_text (str, None): The access token, by default None.
 73
 74            auth (tuple, None): The user credentials
 75                `(username, password)`, by default None.
 76
 77            verify (bool): If True, allow insecure server connections
 78                when using SSL.
 79
 80            log_level (str): The log level, by default `INFO`.
 81
 82            ratelimiting (tuple[int, int], None): `(max_calls, period)`,
 83                If None, no rate limiting is applied. By default, 100 calls
 84                per minute are allowed.
 85
 86            retry (Retry, int, None): Set a retry policy for requests using a
 87                urllib3 Retry object, an integer for the number of retries, or None
 88                for no retries. This happens before py-allspice's own rate limiter.
 89                The default is to retry all methods (even mutating methods)
 90                only on 429s and 502s up to 6 times with exponential backoff
 91                and jitter.
 92
 93                NOTE: AllSpice Hub responds with a 503 and a Retry-After header
 94                when a generated file is not ready yet, and urllib3 retries any
 95                503 that carries Retry-After. So with retries enabled, fetching
 96                generated files transparently waits for generation instead of
 97                raising NotYetGeneratedException, until retries are exhausted.
 98
 99            use_new_schdoc_renderer (bool): Allows explicit override for using the new Altium schematic renderer. If set,
100            this will take precedence over the default behavior on the AllSpice Hub instance.
101        """
102
103        self.logger = logging.getLogger(__name__)
104        handler = logging.StreamHandler(sys.stderr)
105        handler.setFormatter(
106            logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
107        )
108        self.logger.addHandler(handler)
109        self.logger.setLevel(log_level)
110        self.headers = {
111            "Content-type": "application/json",
112        }
113        self.url = allspice_hub_url
114
115        if ratelimiting is None:
116            self.requests = requests.Session()
117        else:
118            (max_calls, period) = ratelimiting
119            self.requests = RateLimitedSession(max_calls=max_calls, period=period)
120
121        if retry is not None:
122            adapter = HTTPAdapter(max_retries=retry)
123            self.requests.mount("https://", adapter)
124            self.requests.mount("http://", adapter)
125
126        # Manage authentification
127        if not token_text and not auth:
128            raise ValueError("Please provide auth or token_text, but not both")
129        if token_text:
130            self.headers["Authorization"] = "token " + token_text
131        if auth:
132            self.logger.warning(
133                "Using basic auth is not recommended. Prefer using a token instead."
134            )
135            self.requests.auth = auth
136
137        # Manage SSL certification verification
138        self.requests.verify = verify
139        if not verify:
140            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
141
142        self.use_new_schdoc_renderer = use_new_schdoc_renderer

Initializing an instance of the AllSpice Hub Client

Args: allspice_hub_url (str): The URL for the AllSpice Hub instance. Defaults to https://hub.allspice.io.

token_text (str, None): The access token, by default None.

auth (tuple, None): The user credentials
    `(username, password)`, by default None.

verify (bool): If True, allow insecure server connections
    when using SSL.

log_level (str): The log level, by default `INFO`.

ratelimiting (tuple[int, int], None): `(max_calls, period)`,
    If None, no rate limiting is applied. By default, 100 calls
    per minute are allowed.

retry (Retry, int, None): Set a retry policy for requests using a
    urllib3 Retry object, an integer for the number of retries, or None
    for no retries. This happens before py-allspice's own rate limiter.
    The default is to retry all methods (even mutating methods)
    only on 429s and 502s up to 6 times with exponential backoff
    and jitter.

    NOTE: AllSpice Hub responds with a 503 and a Retry-After header
    when a generated file is not ready yet, and urllib3 retries any
    503 that carries Retry-After. So with retries enabled, fetching
    generated files transparently waits for generation instead of
    raising NotYetGeneratedException, until retries are exhausted.

use_new_schdoc_renderer (bool): Allows explicit override for using the new Altium schematic renderer. If set,
this will take precedence over the default behavior on the AllSpice Hub instance.

ADMIN_CREATE_USER = '/admin/users'

GET_USERS_ADMIN = '/admin/users'

ADMIN_REPO_CREATE = '/admin/users/%s/repos'

ALLSPICE_HUB_VERSION = '/version'

GET_USER = '/user'

GET_REPOSITORY = '/repos/{owner}/{name}'

CREATE_ORG = '/admin/users/%s/orgs'

CREATE_TEAM = '/orgs/%s/teams'

logger

headers

url

use_new_schdoc_renderer

@staticmethod

def parse_result(result) -> Dict: View Source

168    @staticmethod
169    def parse_result(result) -> Dict:
170        """Parses the result-JSON to a dict."""
171        if result.text and len(result.text) > 3:
172            return json.loads(result.text)
173        return {}

Parses the result-JSON to a dict.

def requests_get( self, endpoint: str, params: Mapping = frozendict.frozendict({}), sudo=None): View Source

175    def requests_get(self, endpoint: str, params: Mapping = frozendict(), sudo=None):
176        combined_params = {}
177        combined_params.update(params)
178        if sudo:
179            combined_params["sudo"] = sudo.username
180        return self.parse_result(self.__get(endpoint, combined_params))

def requests_get_raw( self, endpoint: str, params=frozendict.frozendict({}), sudo=None) -> bytes: View Source

182    def requests_get_raw(self, endpoint: str, params=frozendict(), sudo=None) -> bytes:
183        combined_params = {}
184        combined_params.update(params)
185        if sudo:
186            combined_params["sudo"] = sudo.username
187        return self.__get(endpoint, combined_params).content

def requests_get_paginated( self, endpoint: str, params=frozendict.frozendict({}), sudo=None, page_key: str = 'page', first_page: int = 1): View Source

189    def requests_get_paginated(
190        self,
191        endpoint: str,
192        params=frozendict(),
193        sudo=None,
194        page_key: str = "page",
195        first_page: int = 1,
196    ):
197        page = first_page
198        combined_params = {}
199        combined_params.update(params)
200        aggregated_result = []
201        while True:
202            combined_params[page_key] = page
203            result = self.requests_get(endpoint, combined_params, sudo)
204
205            if not result:
206                return aggregated_result
207
208            if isinstance(result, dict):
209                if "data" in result:
210                    data = result["data"]
211                    if len(data) == 0:
212                        return aggregated_result
213                    aggregated_result.extend(data)
214                elif "tree" in result:
215                    data = result["tree"]
216                    if data is None or len(data) == 0:
217                        return aggregated_result
218                    aggregated_result.extend(data)
219                else:
220                    raise NotImplementedError(
221                        "requests_get_paginated does not know how to handle responses of this type."
222                    )
223            else:
224                aggregated_result.extend(result)
225
226            page += 1

def requests_put(self, endpoint: str, data: Optional[dict] = None): View Source

228    def requests_put(self, endpoint: str, data: Optional[dict] = None):
229        if not data:
230            data = {}
231        response = self.requests.put(
232            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
233        )
234        if response.status_code not in [200, 204]:
235            message = (
236                f"Received status code: {response.status_code} ({response.url}) {response.text}"
237            )
238            self.logger.error(message)
239            raise Exception(message)

def requests_delete(self, endpoint: str, data: Optional[dict] = None): View Source

241    def requests_delete(self, endpoint: str, data: Optional[dict] = None):
242        response = self.requests.delete(
243            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
244        )
245        if response.status_code not in [200, 204]:
246            message = f"Received status code: {response.status_code} ({response.url})"
247            self.logger.error(message)
248            raise Exception(message)

def requests_post( self, endpoint: str, data: Optional[dict] = None, params: Optional[dict] = None, files: Optional[dict] = None): View Source

250    def requests_post(
251        self,
252        endpoint: str,
253        data: Optional[dict] = None,
254        params: Optional[dict] = None,
255        files: Optional[dict] = None,
256    ):
257        """
258        Make a POST call to the endpoint.
259
260        :param endpoint: The path to the endpoint
261        :param data: A dictionary for JSON data
262        :param params: A dictionary of query params
263        :param files: A dictionary of files, see requests.post. Using both files and data
264                      can lead to unexpected results!
265        :return: The JSON response parsed as a dict
266        """
267
268        # This should ideally be a TypedDict of the type of arguments taken by
269        # `requests.post`.
270        args: dict[str, Any] = {
271            "headers": self.headers.copy(),
272        }
273        if data is not None:
274            args["data"] = json.dumps(data)
275        if params is not None:
276            args["params"] = params
277        if files is not None:
278            args["headers"].pop("Content-type")
279            args["files"] = files
280
281        response = self.requests.post(self.__get_url(endpoint), **args)
282
283        if response.status_code not in [200, 201, 202]:
284            if "already exists" in response.text or "e-mail already in use" in response.text:
285                self.logger.warning(response.text)
286                raise AlreadyExistsException()
287            self.logger.error(f"Received status code: {response.status_code} ({response.url})")
288            self.logger.error(f"With info: {data} ({self.headers})")
289            self.logger.error(f"Answer: {response.text}")
290            raise Exception(
291                f"Received status code: {response.status_code} ({response.url}), {response.text}"
292            )
293        return self.parse_result(response)

Make a POST call to the endpoint.

Parameters

endpoint: The path to the endpoint
data: A dictionary for JSON data
params: A dictionary of query params
files: A dictionary of files, see requests.post. Using both files and data can lead to unexpected results!

Returns

The JSON response parsed as a dict

def requests_patch(self, endpoint: str, data: dict): View Source

295    def requests_patch(self, endpoint: str, data: dict):
296        response = self.requests.patch(
297            self.__get_url(endpoint), headers=self.headers, data=json.dumps(data)
298        )
299        if response.status_code not in [200, 201]:
300            error_message = f"Received status code: {response.status_code} ({response.url}) {data}"
301            self.logger.error(error_message)
302            raise Exception(error_message)
303        return self.parse_result(response)

def get_orgs_public_members_all(self, orgname): View Source

305    def get_orgs_public_members_all(self, orgname):
306        path = "/orgs/" + orgname + "/public_members"
307        return self.requests_get(path)

def get_orgs(self): View Source

309    def get_orgs(self):
310        path = "/admin/orgs"
311        results = self.requests_get(path)
312        return [Organization.parse_response(self, result) for result in results]

def get_user(self): View Source

314    def get_user(self):
315        result = self.requests_get(AllSpice.GET_USER)
316        return User.parse_response(self, result)

def get_version(self) -> str: View Source

318    def get_version(self) -> str:
319        result = self.requests_get(AllSpice.ALLSPICE_HUB_VERSION)
320        return result["version"]

def get_users(self) -> List[User]: View Source

322    def get_users(self) -> List[User]:
323        results = self.requests_get(AllSpice.GET_USERS_ADMIN)
324        return [User.parse_response(self, result) for result in results]

def get_user_by_email(self, email: str) -> Optional[User]: View Source

326    def get_user_by_email(self, email: str) -> Optional[User]:
327        users = self.get_users()
328        for user in users:
329            if user.email == email or email in user.emails:
330                return user
331        return None

def get_user_by_name(self, username: str) -> Optional[User]: View Source

333    def get_user_by_name(self, username: str) -> Optional[User]:
334        users = self.get_users()
335        for user in users:
336            if user.username == username:
337                return user
338        return None

def get_repository(self, owner: str, name: str) -> Repository: View Source

340    def get_repository(self, owner: str, name: str) -> Repository:
341        path = self.GET_REPOSITORY.format(owner=owner, name=name)
342        result = self.requests_get(path)
343        return Repository.parse_response(self, result)

def create_user( self, user_name: str, email: str, password: str, full_name: Optional[str] = None, login_name: Optional[str] = None, change_pw=True, send_notify=True, source_id=0): View Source

345    def create_user(
346        self,
347        user_name: str,
348        email: str,
349        password: str,
350        full_name: Optional[str] = None,
351        login_name: Optional[str] = None,
352        change_pw=True,
353        send_notify=True,
354        source_id=0,
355    ):
356        """Create User.
357        Throws:
358            AlreadyExistsException, if the User exists already
359            Exception, if something else went wrong.
360        """
361        if not login_name:
362            login_name = user_name
363        if not full_name:
364            full_name = user_name
365        request_data = {
366            "source_id": source_id,
367            "login_name": login_name,
368            "full_name": full_name,
369            "username": user_name,
370            "email": email,
371            "password": password,
372            "send_notify": send_notify,
373            "must_change_password": change_pw,
374        }
375
376        self.logger.debug("Gitea post payload: %s", request_data)
377        result = self.requests_post(AllSpice.ADMIN_CREATE_USER, data=request_data)
378        if "id" in result:
379            self.logger.info(
380                "Successfully created User %s <%s> (id %s)",
381                result["login"],
382                result["email"],
383                result["id"],
384            )
385            self.logger.debug("Gitea response: %s", result)
386        else:
387            self.logger.error(result["message"])
388            raise Exception("User not created... (gitea: %s)" % result["message"])
389        user = User.parse_response(self, result)
390        return user

Create User. Throws: AlreadyExistsException, if the User exists already Exception, if something else went wrong.

def create_repo( self, repoOwner: Union[User, Organization], 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'): View Source

392    def create_repo(
393        self,
394        repoOwner: Union[User, Organization],
395        repoName: str,
396        description: str = "",
397        private: bool = False,
398        autoInit=True,
399        gitignores: Optional[str] = None,
400        license: Optional[str] = None,
401        readme: str = "Default",
402        issue_labels: Optional[str] = None,
403        default_branch="master",
404    ):
405        """Create a Repository as the administrator
406
407        Throws:
408            AlreadyExistsException: If the Repository exists already.
409            Exception: If something else went wrong.
410
411        Note:
412            Non-admin users can not use this method. Please use instead
413            `allspice.User.create_repo` or `allspice.Organization.create_repo`.
414        """
415        # although this only says user in the api, this also works for
416        # organizations
417        assert isinstance(repoOwner, User) or isinstance(repoOwner, Organization)
418        result = self.requests_post(
419            AllSpice.ADMIN_REPO_CREATE % repoOwner.username,
420            data={
421                "name": repoName,
422                "description": description,
423                "private": private,
424                "auto_init": autoInit,
425                "gitignores": gitignores,
426                "license": license,
427                "issue_labels": issue_labels,
428                "readme": readme,
429                "default_branch": default_branch,
430            },
431        )
432        if "id" in result:
433            self.logger.info("Successfully created Repository %s " % result["name"])
434        else:
435            self.logger.error(result["message"])
436            raise Exception("Repository not created... (gitea: %s)" % result["message"])
437        return Repository.parse_response(self, result)

Create a Repository as the administrator

Throws: AlreadyExistsException: If the Repository exists already. Exception: If something else went wrong.

Note: Non-admin users can not use this method. Please use instead allspice.User.create_repo or allspice.Organization.create_repo.

def create_org( self, owner: User, orgName: str, description: str, location='', website='', full_name=''): View Source

439    def create_org(
440        self,
441        owner: User,
442        orgName: str,
443        description: str,
444        location="",
445        website="",
446        full_name="",
447    ):
448        assert isinstance(owner, User)
449        result = self.requests_post(
450            AllSpice.CREATE_ORG % owner.username,
451            data={
452                "username": orgName,
453                "description": description,
454                "location": location,
455                "website": website,
456                "full_name": full_name,
457            },
458        )
459        if "id" in result:
460            self.logger.info("Successfully created Organization %s" % result["username"])
461        else:
462            self.logger.error("Organization not created... (gitea: %s)" % result["message"])
463            self.logger.error(result["message"])
464            raise Exception("Organization not created... (gitea: %s)" % result["message"])
465        return Organization.parse_response(self, result)

def create_team( self, org: Organization, 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={}): View Source

467    def create_team(
468        self,
469        org: Organization,
470        name: str,
471        description: str = "",
472        permission: str = "read",
473        can_create_org_repo: bool = False,
474        includes_all_repositories: bool = False,
475        units=(
476            "repo.code",
477            "repo.issues",
478            "repo.ext_issues",
479            "repo.wiki",
480            "repo.pulls",
481            "repo.releases",
482            "repo.ext_wiki",
483        ),
484        units_map={},
485    ):
486        """Creates a Team.
487
488        Args:
489            org (Organization): Organization the Team will be part of.
490            name (str): The Name of the Team to be created.
491            description (str): Optional, None, short description of the new Team.
492            permission (str): Optional, 'read', What permissions the members
493            units_map (dict): Optional, {}, a mapping of units to their
494                permissions. If None or empty, the `permission` permission will
495                be applied to all units. Note: When both `units` and `units_map`
496                are given, `units_map` will be preferred.
497        """
498
499        result = self.requests_post(
500            AllSpice.CREATE_TEAM % org.username,
501            data={
502                "name": name,
503                "description": description,
504                "permission": permission,
505                "can_create_org_repo": can_create_org_repo,
506                "includes_all_repositories": includes_all_repositories,
507                "units": units,
508                "units_map": units_map,
509            },
510        )
511
512        if "id" in result:
513            self.logger.info("Successfully created Team %s" % result["name"])
514        else:
515            self.logger.error("Team not created... (gitea: %s)" % result["message"])
516            self.logger.error(result["message"])
517            raise Exception("Team not created... (gitea: %s)" % result["message"])
518        api_object = Team.parse_response(self, result)
519        setattr(
520            api_object, "_organization", org
521        )  # fixes strange behaviour of gitea not returning a valid organization here.
522        return api_object

Creates a Team.

Args: org (Organization): Organization the Team will be part of. name (str): The Name of the Team to be created. description (str): Optional, None, short description of the new Team. permission (str): Optional, 'read', What permissions the members units_map (dict): Optional, {}, a mapping of units to their permissions. If None or empty, the permission permission will be applied to all units. Note: When both units and units_map are given, units_map will be preferred.

class AlreadyExistsException(builtins.Exception): View Source

70class AlreadyExistsException(Exception):
71    pass

Common base class for all non-exit exceptions.

class Branch(allspice.baseapiobject.ReadonlyApiObject): View Source

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})

Branch(allspice_client) View Source

436    def __init__(self, allspice_client):
437        super().__init__(allspice_client)

commit: Dict[str, Union[str, Dict[str, str], Dict[str, Union[bool, str, NoneType]], NoneType]]

effective_branch_protection_name: str

enable_status_check: bool

protected: bool

required_approvals: int

status_check_contexts: List[Any]

user_can_merge: bool

user_can_push: bool

API_OBJECT = '/repos/{owner}/{repo}/branches/{branch}'

@classmethod

def request(cls, allspice_client, owner: str, repo: str, branch: str): View Source

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

allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; parse_response

class Comment(allspice.baseapiobject.ApiObject): View Source

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

Comment(allspice_client) View Source

1640    def __init__(self, allspice_client):
1641        super().__init__(allspice_client)

assets: List[Union[Any, Dict[str, Union[int, str]]]]

body: str

created_at: datetime.datetime

html_url: str

id: int

issue_url: str

original_author: str

original_author_id: int

pull_request_url: str

updated_at: datetime.datetime

user: User

API_OBJECT = '/repos/{owner}/{repo}/issues/comments/{id}'

GET_ATTACHMENTS_PATH = '/repos/{owner}/{repo}/issues/comments/{id}/assets'

ATTACHMENT_PATH = '/repos/{owner}/{repo}/issues/comments/{id}/assets/{attachment_id}'

@classmethod

def request( cls, allspice_client, owner: str, repo: str, id: str) -> Comment: View Source

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})

parent_url: str View Source

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 View Source

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.

def commit(self): View Source

1686    def commit(self):
1687        self._commit(self.__fields_for_path())

def delete(self): View Source

1689    def delete(self):
1690        self.allspice_client.requests_delete(self.API_OBJECT.format(**self.__fields_for_path()))
1691        self.deleted = True

def get_attachments(self) -> List[allspice.apiobject.Attachment]: View Source

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

def create_attachment( self, file: <class 'IO'>, name: Optional[str] = None) -> allspice.apiobject.Attachment: View Source

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.

def edit_attachment( self, attachment: allspice.apiobject.Attachment, data: dict) -> allspice.apiobject.Attachment: View Source

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

def delete_attachment(self, attachment: allspice.apiobject.Attachment): View Source

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

Inherited Members

allspice.baseapiobject.ApiObject: get_dirty_fields
allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; parse_response

class Commit(allspice.baseapiobject.ReadonlyApiObject): View Source

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        }

Commit(allspice_client) View Source

1783    def __init__(self, allspice_client):
1784        super().__init__(allspice_client)

author: User

commit: Dict[str, Union[str, Dict[str, str], Dict[str, Union[bool, str, NoneType]]]]

committer: Dict[str, Union[int, str, bool]]

created: str

files: List[Dict[str, str]]

html_url: str

inner_commit: Dict[str, Union[str, Dict[str, str], Dict[str, Union[bool, str, NoneType]]]]

parents: List[Union[Dict[str, str], Any]]

sha: str

stats: Dict[str, int]

url: str

API_OBJECT = '/repos/{owner}/{repo}/commits/{sha}'

COMMIT_GET_STATUS = '/repos/{owner}/{repo}/commits/{sha}/status'

COMMIT_GET_STATUSES = '/repos/{owner}/{repo}/commits/{sha}/statuses'

URL_REGEXP = re.compile('/repos/([^/]+)/([^/]+)/git/commits')

@classmethod

def parse_response(cls, allspice_client, result) -> Commit: View Source

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

def get_status(self) -> allspice.apiobject.CommitCombinedStatus: View Source

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

def get_statuses(self) -> List[allspice.apiobject.CommitStatus]: View Source

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

allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; request

class Content(allspice.baseapiobject.ReadonlyApiObject): View Source

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)

Content(allspice_client) View Source

2964    def __init__(self, allspice_client):
2965        super().__init__(allspice_client)

content: Any

download_url: str

encoding: Any

git_url: str

html_url: str

last_author_date: str

last_commit_sha: str

last_committer_date: str

path: str

sha: str

size: int

submodule_git_url: Any

target: Any

type: str

url: str

FILE = 'file'

Inherited Members

allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; request; parse_response

class DesignReview.MergeType(enum.Enum): View Source

2379    class MergeType(Enum):
2380        MERGE = "merge"
2381        REBASE = "rebase"
2382        REBASE_MERGE = "rebase-merge"
2383        SQUASH = "squash"
2384        MANUALLY_MERGED = "manually-merged"

MERGE = <MergeType.MERGE: 'merge'>

REBASE = <MergeType.REBASE: 'rebase'>

REBASE_MERGE = <MergeType.REBASE_MERGE: 'rebase-merge'>

SQUASH = <MergeType.SQUASH: 'squash'>

MANUALLY_MERGED = <MergeType.MANUALLY_MERGED: 'manually-merged'>

class DesignReviewReview(allspice.baseapiobject.ReadonlyApiObject): View Source

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.

DesignReviewReview(allspice_client) View Source

2237    def __init__(self, allspice_client):
2238        super().__init__(allspice_client)

body: str

comments_count: int

commit_id: str

dismissed: bool

html_url: str

id: int

official: bool

pull_request_url: str

stale: bool

state: DesignReviewReview.ReviewEvent

submitted_at: str

team: Any

updated_at: str

user: User

API_OBJECT = '/repos/{owner}/{repo}/pulls/{index}/reviews/{id}'

GET_COMMENTS = '/repos/{owner}/{repo}/pulls/{index}/reviews/{id}/comments'

owner_name: str View Source

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 View Source

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 View Source

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): View Source

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.

def get_comments(self) -> List[allspice.apiobject.DesignReviewReviewComment]: View Source

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

allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; request; parse_response

class DesignReviewReview.ReviewEvent(enum.Enum): View Source

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 = ""

APPROVED = <ReviewEvent.APPROVED: 'APPROVED'>

PENDING = <ReviewEvent.PENDING: 'PENDING'>

COMMENT = <ReviewEvent.COMMENT: 'COMMENT'>

REQUEST_CHANGES = <ReviewEvent.REQUEST_CHANGES: 'REQUEST_CHANGES'>

REQUEST_REVIEW = <ReviewEvent.REQUEST_REVIEW: 'REQUEST_REVIEW'>

UNKNOWN = <ReviewEvent.UNKNOWN: ''>

@dataclass

class DesignReviewReview.ReviewComment: View Source

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 Content object, get the path using the path property.
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.

DesignReviewReview.ReviewComment( body: str, path: str, sub_path: Optional[str] = None, new_position: Optional[int] = None, old_position: Optional[int] = None, severity: Optional[Literal['error', 'warning', 'recommendation']] = None)

body: str

path: str

sub_path: Optional[str] = None

new_position: Optional[int] = None

old_position: Optional[int] = None

severity: Optional[Literal['error', 'warning', 'recommendation']] = None

class InternalServerException(builtins.Exception): View Source

35class InternalServerException(Exception):
36    def __init__(self, message: str, body: Optional[APIError] = None):
37        super().__init__(message)
38        self.body = body

Common base class for all non-exit exceptions.

InternalServerException(message: str, body: Optional[APIError] = None) View Source

36    def __init__(self, message: str, body: Optional[APIError] = None):
37        super().__init__(message)
38        self.body = body

body

class Milestone(allspice.baseapiobject.ApiObject): View Source

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})

Milestone(allspice_client) View Source

1534    def __init__(self, allspice_client):
1535        super().__init__(allspice_client)

allow_merge_commits: Any

allow_rebase: Any

allow_rebase_explicit: Any

allow_squash_merge: Any

archived: Any

closed_at: Any

closed_issues: int

created_at: str

default_branch: Any

description: str

due_on: Any

has_issues: Any

has_pull_requests: Any

has_wiki: Any

id: int

ignore_whitespace_conflicts: Any

open_issues: int

private: Any

state: str

title: str

updated_at: str

website: Any

API_OBJECT = '/repos/{owner}/{repo}/milestones/{number}'

@classmethod

def request(cls, allspice_client, owner: str, repo: str, number: str): View Source

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})

Inherited Members

allspice.baseapiobject.ApiObject: commit; get_dirty_fields
allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; parse_response

class NotFoundException(builtins.Exception): View Source

74class NotFoundException(Exception):
75    pass

Common base class for all non-exit exceptions.

class Release(allspice.baseapiobject.ApiObject): View Source

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.

Release(allspice_client) View Source

2732    def __init__(self, allspice_client):
2733        super().__init__(allspice_client)

assets: List[Union[Any, Dict[str, Union[int, str]], allspice.apiobject.ReleaseAsset]]

author: User

body: str

created_at: str

draft: bool

html_url: str

id: int

prerelease: bool

published_at: str

repo: Optional[Repository]

repository: Optional[Repository]

tag_name: str

tarball_url: str

target_commitish: str

upload_url: str

url: str

zipball_url: str

API_OBJECT = '/repos/{owner}/{repo}/releases/{id}'

RELEASE_CREATE_ASSET = '/repos/{owner}/{repo}/releases/{id}/assets'

@classmethod

def parse_response(cls, allspice_client, result, repo) -> Release: View Source

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: View Source

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

def commit(self): View Source

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)

def create_asset( self, file: <class 'IO'>, name: Optional[str] = None) -> allspice.apiobject.ReleaseAsset: View Source

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): View Source

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

Inherited Members

allspice.baseapiobject.ApiObject: get_dirty_fields
allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted

class RenderException(builtins.Exception): View Source

41class RenderException(Exception):
42    """Raised when a server-side render fails; not retried."""
43
44    @classmethod
45    def from_internal(
46        cls, internal: InternalServerException, file_path: str, ref: Optional[str]
47    ) -> Optional["RenderException"]:
48        body = internal.body
49
50        if body is None or body.category != "render":
51            return None
52
53        if body.message:
54            message = body.message
55        else:
56            message = f"Render failed for {file_path}"
57            if ref is not None:
58                message += f" at ref {ref}"
59
60        # Include a diagnostic url if present (a Hub link to the admin panel, don't include other links like swagger docs),
61        # otherwise just ask them to check Hub logs
62        if body.url and "-/admin/" in body.url:
63            message += f"; A diagnostic report for this failure can be downloaded by a site admin at {body.url}."
64        else:
65            message += "; Check AllSpice Hub logs for more information."
66
67        return cls(message)

Raised when a server-side render fails; not retried.

@classmethod

def from_internal( cls, internal: InternalServerException, file_path: str, ref: Optional[str]) -> Optional[RenderException]: View Source

44    @classmethod
45    def from_internal(
46        cls, internal: InternalServerException, file_path: str, ref: Optional[str]
47    ) -> Optional["RenderException"]:
48        body = internal.body
49
50        if body is None or body.category != "render":
51            return None
52
53        if body.message:
54            message = body.message
55        else:
56            message = f"Render failed for {file_path}"
57            if ref is not None:
58                message += f" at ref {ref}"
59
60        # Include a diagnostic url if present (a Hub link to the admin panel, don't include other links like swagger docs),
61        # otherwise just ask them to check Hub logs
62        if body.url and "-/admin/" in body.url:
63            message += f"; A diagnostic report for this failure can be downloaded by a site admin at {body.url}."
64        else:
65            message += "; Check AllSpice Hub logs for more information."
66
67        return cls(message)

class Repository(allspice.baseapiobject.ApiObject): View Source

 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

Repository(allspice_client) View Source

593    def __init__(self, allspice_client):
594        super().__init__(allspice_client)

allow_fast_forward_only_merge: bool

allow_manual_merge: Any

allow_merge_commits: bool

allow_rebase: bool

allow_rebase_explicit: bool

allow_rebase_update: bool

allow_squash_merge: bool

archived: bool

archived_at: str

autodetect_manual_merge: Any

avatar_url: str

clone_url: str

created_at: str

default_allow_maintainer_edit: bool

default_branch: str

default_delete_branch_after_merge: bool

default_merge_style: str

description: str

empty: bool

enable_prune: Any

external_tracker: Any

external_wiki: Any

fork: bool

forks_count: int

full_name: str

has_actions: bool

has_issues: bool

has_packages: bool

has_projects: bool

has_pull_requests: bool

has_releases: bool

has_wiki: bool

html_url: str

id: int

ignore_whitespace_conflicts: bool

internal: bool

internal_tracker: Dict[str, bool]

language: str

languages_url: str

licenses: List[Any]

link: str

mirror: bool

mirror_interval: str

mirror_updated: str

object_format_name: str

open_issues_count: int

open_pr_counter: int

original_url: str

owner: Union[User, Organization]

parent: Any

permissions: Dict[str, bool]

private: bool

projects_mode: str

release_counter: int

repo_transfer: Any

size: int

ssh_url: str

stars_count: int

topics: List[Union[Any, str]]

updated_at: datetime.datetime

url: str

watchers_count: int

website: str

API_OBJECT = '/repos/{owner}/{name}'

REPO_IS_COLLABORATOR = '/repos/%s/%s/collaborators/%s'

REPO_SEARCH = '/repos/search/'

REPO_BRANCHES = '/repos/%s/%s/branches'

REPO_BRANCH = '/repos/{owner}/{repo}/branches/{branch}'

REPO_ISSUES = '/repos/{owner}/{repo}/issues'

REPO_DESIGN_REVIEWS = '/repos/{owner}/{repo}/pulls'

REPO_DELETE = '/repos/%s/%s'

REPO_TIMES = '/repos/%s/%s/times'

REPO_USER_TIME = '/repos/%s/%s/times/%s'

REPO_COMMITS = '/repos/%s/%s/commits'

REPO_TRANSFER = '/repos/{owner}/{repo}/transfer'

REPO_MILESTONES = '/repos/{owner}/{repo}/milestones'

REPO_GET_ARCHIVE = '/repos/{owner}/{repo}/archive/{ref}.{format}'

REPO_GET_ALLSPICE_JSON = '/repos/{owner}/{repo}/allspice_generated/json/{content}'

REPO_GET_ALLSPICE_SVG = '/repos/{owner}/{repo}/allspice_generated/svg/{content}'

REPO_GET_ALLSPICE_PROJECT = '/repos/{owner}/{repo}/allspice_generated/project/{content}'

REPO_GET_TOPICS = '/repos/{owner}/{repo}/topics'

REPO_ADD_TOPIC = '/repos/{owner}/{repo}/topics/{topic}'

REPO_GET_RELEASES = '/repos/{owner}/{repo}/releases'

REPO_GET_LATEST_RELEASE = '/repos/{owner}/{repo}/releases/latest'

REPO_GET_RELEASE_BY_TAG = '/repos/{owner}/{repo}/releases/tags/{tag}'

REPO_GET_COMMIT_STATUS = '/repos/{owner}/{repo}/statuses/{sha}'

REPO_GET_MEDIA = '/repos/{owner}/{repo}/media/{path}'

REPO_GET_TREE = '/repos/{owner}/{repo}/git/trees/{ref}'

@classmethod

def request( cls, allspice_client, owner: str, name: str) -> Repository: View Source

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})

@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]: View Source

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.

def commit(self): View Source

698    def commit(self):
699        args = {"owner": self.owner.username, "name": self.name}
700        self._commit(args)

def get_branches(self) -> List[Branch]: View Source

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.

def get_branch(self, name: str) -> Branch: View Source

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.

def add_branch( self, create_from: Union[Branch, Commit, str], newname: str) -> Branch: View Source

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]: View Source

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_query in 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]: View Source

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]: View Source

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.

def get_issues_state(self, state) -> List[Issue]: View Source

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_times(self): View Source

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

def get_user_time(self, username) -> float: View Source

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 get_full_name(self) -> str: View Source

912    def get_full_name(self) -> str:
913        return self.owner.username + "/" + self.name

def create_issue( self, title, assignees=frozenset(), description='') -> allspice.baseapiobject.ApiObject: View Source

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: View Source

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: View Source

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]): View Source

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 list_hooks(self): View Source

1010    def list_hooks(self):
1011        url = f"/repos/{self.owner.username}/{self.name}/hooks"
1012        return self.allspice_client.requests_get(url)

def delete_hook(self, id: str): View Source

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)

def is_collaborator(self, username) -> bool: View Source

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

def get_users_with_access(self) -> Sequence[User]: View Source

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 remove_collaborator(self, user_name: str): View Source

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)

def transfer_ownership( self, new_owner: Union[User, Organization], new_teams: Union[Set[Team], FrozenSet[Team]] = frozenset()): View Source

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]: View Source

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[allspice.apiobject.GitEntry]: View Source

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]]: View Source

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

def get_raw_file( self, file_path: str, ref: Union[Branch, Commit, str, NoneType] = None) -> bytes: View Source

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: View Source

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: View Source

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: View Source

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: View Source

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): View Source

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): View Source

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): View Source

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: View Source

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]: View Source

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): View Source

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): View Source

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]: View Source

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

def get_latest_release(self) -> Release: View Source

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

def get_release_by_tag(self, tag: str) -> Release: View Source

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[allspice.apiobject.CommitStatusState] = None) -> List[allspice.apiobject.CommitStatus]: View Source

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[allspice.apiobject.CommitStatusState] = None, target_url: Optional[str] = None) -> allspice.apiobject.CommitStatus: View Source

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

def delete(self): View Source

1500    def delete(self):
1501        self.allspice_client.requests_delete(
1502            Repository.REPO_DELETE % (self.owner.username, self.name)
1503        )
1504        self.deleted = True

Inherited Members

allspice.baseapiobject.ApiObject: get_dirty_fields
allspice.baseapiobject.ReadonlyApiObject: allspice_client; deleted; parse_response

class Repository.ArchiveFormat(enum.Enum): View Source

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

TAR = <ArchiveFormat.TAR: 'tar.gz'>

ZIP = <ArchiveFormat.ZIP: 'zip'>

class Repository.CommitStatusSort(enum.Enum): View Source

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

OLDEST = <CommitStatusSort.OLDEST: 'oldest'>

RECENT_UPDATE = <CommitStatusSort.RECENT_UPDATE: 'recentupdate'>

LEAST_UPDATE = <CommitStatusSort.LEAST_UPDATE: 'leastupdate'>

LEAST_INDEX = <CommitStatusSort.LEAST_INDEX: 'leastindex'>

HIGHEST_INDEX = <CommitStatusSort.HIGHEST_INDEX: 'highestindex'>