allspice.allspice

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