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