Type annotations and models

osctiny/extensions/bs_requests.py

@@ -2,10 +2,13 @@
 Requests extension
 ------------------
 """
+import typing
 from urllib.parse import urljoin
 
 from lxml.etree import XMLSyntaxError
+from lxml.objectify import ObjectifiedElement
 
+from ..models import IntOrString, ParamsType
 from ..utils.base import ExtensionBase
 
 
@@ -16,15 +19,15 @@ class Request(ExtensionBase):
     base_path = "/request/"
 
     @staticmethod
-    def _validate_id(request_id):
+    def _validate_id(request_id: IntOrString) -> str:
         request_id = str(request_id)
         if not request_id.isnumeric():
             raise ValueError(
                 "Request ID must be numeric! Got instead: {}".format(request_id)
             )
         return request_id
 
-    def get_list(self, **params):
+    def get_list(self, **params: ParamsType) -> ObjectifiedElement:
         """
         Get a list or request objects
 
@@ -40,7 +43,8 @@ def get_list(self, **params):
 
         return self.osc.get_objectified_xml(response)
 
-    def get(self, request_id, withhistory=False, withfullhistory=False):
+    def get(self, request_id: IntOrString, withhistory: bool = False,
+            withfullhistory: bool = False) -> ObjectifiedElement:
         """
         Get one request object
 
@@ -65,7 +69,8 @@ def get(self, request_id, withhistory=False, withfullhistory=False):
 
         return self.osc.get_objectified_xml(response)
 
-    def update(self, request_id, **kwargs):
+    def update(self, request_id: IntOrString, **kwargs: ParamsType) \
+            -> typing.Union[ObjectifiedElement, str]:
         """
         Update request or execute command
 
@@ -96,7 +101,8 @@ def update(self, request_id, **kwargs):
         except XMLSyntaxError:
             return response.text
 
-    def cmd(self, request_id, cmd="diff", **kwargs):
+    def cmd(self, request_id: IntOrString, cmd: str = "diff", **kwargs: ParamsType) \
+            -> typing.Union[ObjectifiedElement, str]:
         """
         Get the result of the specified command
 
@@ -130,7 +136,8 @@ def cmd(self, request_id, cmd="diff", **kwargs):
         request_id = self._validate_id(request_id)
         return self.update(request_id=request_id, **kwargs)
 
-    def add_comment(self, request_id, comment, parent_id=None):
+    def add_comment(self, request_id: IntOrString, comment: str,
+                    parent_id: typing.Optional[str] = None) -> bool:
         """
         Add a comment to a request
 
@@ -152,7 +159,7 @@ def add_comment(self, request_id, comment, parent_id=None):
             parent_id=parent_id
         )
 
-    def get_comments(self, request_id):
+    def get_comments(self, request_id: IntOrString) -> ObjectifiedElement:
         """
         Get a list of comments for request
 

osctiny/extensions/staging.py

@@ -17,7 +17,7 @@
 
 from lxml.objectify import ObjectifiedElement, Element, SubElement
 
-from ..models.staging import ExcludedRequest, CheckReport
+from ..models.staging import E, ExcludedRequest, CheckReport
 from ..utils.base import ExtensionBase
 
 
@@ -69,15 +69,11 @@ def set_excluded_requests(self, project: str, *requests: ExcludedRequest) -> boo
         :raises HTTPError: if comment was not saved correctly. The raised exception contains the
                            full response object and API response.
         """
-        excluded_requests = Element("excluded_requests")
-        for request in requests:
-            SubElement(excluded_requests, "request", **request.asdict())
-
         response = self.osc.request(
             method="POST",
             url=urljoin(self.osc.url, "{}/{}/excluded_requests".format(self.base_path_staging,
                                                                        project)),
-            data=excluded_requests
+            data=E.excluded_requests(*(request.asxml() for request in requests))
         )
         parsed = self.osc.get_objectified_xml(response)
         if response.status_code == 200 and parsed.get("code") == "ok":
@@ -95,15 +91,11 @@ def delete_excluded_requests(self, project: str, *requests: ExcludedRequest) ->
         :raises HTTPError: if comment was not saved correctly. The raised exception contains the
                            full response object and API response.
         """
-        excluded_requests = Element("excluded_requests")
-        for request in requests:
-            SubElement(excluded_requests, "request", **request.asdict())
-
         response = self.osc.request(
             method="DELETE",
             url=urljoin(self.osc.url, "{}/{}/excluded_requests".format(self.base_path_staging,
                                                                        project)),
-            data=excluded_requests
+            data=E.excluded_requests(*(request.asxml() for request in requests))
         )
         parsed = self.osc.get_objectified_xml(response)
         if response.status_code == 200 and parsed.get("code") == "ok":
@@ -295,7 +287,6 @@ def set_required_checks(self, project: str, checks: typing.List[str],
         :raises HTTPError: if comment was not saved correctly. The raised exception contains the
                            full response object and API response.
         """
-        # pylint: disable=protected-access
         kwargs = {"project": project}
         if repo and arch:
             url_path = "{}/built_repositories/{}/{}/{}/required_checks".format(
@@ -310,15 +301,10 @@ def set_required_checks(self, project: str, checks: typing.List[str],
         else:
             url_path = "{}/projects/{}/required_checks".format(self.base_path_status, project)
 
-        required_checks = Element("required_checks")
-        for check in checks:
-            elem = SubElement(required_checks, "name")
-            elem._setText(check)
-
         response = self.osc.request(
             method="POST",
             url=urljoin(self.osc.url, url_path),
-            data=required_checks
+            data=E.required_checks(*(E.name(check) for check in checks))
         )
         parsed = self.osc.get_objectified_xml(response)
         if response.status_code == 200 and parsed.get("code") == "ok":
@@ -376,19 +362,6 @@ def set_status_report(self, project: str, repo: str, build_id: str, report: Chec
         :raises HTTPError: if comment was not saved correctly. The raised exception contains the
                            full response object and API response.
         """
-        # pylint: disable=protected-access
-        check = Element("check",
-                        name=report.name,
-                        required="true" if report.required else "false")
-        state = SubElement(check, "state")
-        state._setText(report.state.value)
-        if report.short_description:
-            short_desc = SubElement(check, "short_description")
-            short_desc._setText(report.short_description)
-        if report.url:
-            url = SubElement(check, "url")
-            url._setText(report.url)
-
         if arch:
             url_path = "{}/built/{}/{}/{}/reports/{}".format(
                 self.base_path_status, project, repo, arch, build_id
@@ -401,7 +374,7 @@ def set_status_report(self, project: str, repo: str, build_id: str, report: Chec
         response = self.osc.request(
             method="POST",
             url=urljoin(self.osc.url, url_path),
-            data=check
+            data=report.asxml()
         )
 
         parsed = self.osc.get_objectified_xml(response)

osctiny/models/__init__.py

@@ -9,3 +9,4 @@
 
 
 ParamsType = typing.Union[bytes, str, StringIO, BytesIO, BufferedReader, dict, ObjectifiedElement]
+IntOrString = typing.Union[int, str]

osctiny/models/staging.py

@@ -6,6 +6,11 @@
 import enum
 import typing
 
+from lxml.objectify import ObjectifiedElement, ElementMaker
+
+
+E = ElementMaker(annotate=False)
+
 
 class ExcludedRequest(typing.NamedTuple):
     id: int
@@ -18,6 +23,9 @@ def asdict(self) -> typing.Dict[str, str]:
 
         return d
 
+    def asxml(self) -> ObjectifiedElement:
+        return E.request(**self.asdict())
+
 
 class CheckState(enum.Enum):
     PENDING = "pending"
@@ -32,3 +40,30 @@ class CheckReport(typing.NamedTuple):
     state: CheckState
     short_description: typing.Optional[str] = None
     url: typing.Optional[str] = None
+
+    @property
+    def required_str(self) -> str:
+        return "true" if self.required else "false"
+
+    def _optional_fields(self) -> typing.Generator[typing.Tuple[str, str], None, None]:
+        for key in ('url', 'short_description'):
+            value = getattr(self, key, None)
+            if value:
+                yield key, str(value)
+
+    def asdict(self) -> typing.Dict[str, str]:
+        d = {"name": self.name, "required": self.required_str,
+             "state": self.state.value}
+
+        for key, value in self._optional_fields():
+            d[key] = value
+
+        return d
+
+    def asxml(self) -> ObjectifiedElement:
+        sub_elems = [E.state(self.state.value)]
+        for key, value in self._optional_fields():
+            _E = getattr(E, key)
+            sub_elems.append(_E(value))
+
+        return E.check(*sub_elems, name=self.name, required=self.required_str)

osctiny/osc.py

@@ -22,7 +22,7 @@
 
 from lxml.etree import tostring
 from lxml.objectify import ObjectifiedElement
-from requests import Session, Request
+from requests import Session, Request, Response
 from requests.auth import HTTPBasicAuth
 from requests.cookies import RequestsCookieJar, cookiejar_from_dict
 from requests.exceptions import ConnectionError as _ConnectionError
@@ -236,8 +236,11 @@ def parser(self):
         """
         return get_xml_parser()
 
-    def request(self, url, method="GET", stream=False, data=None, params=None,
-                raise_for_status=True, timeout=None):
+    def request(self, url: str, method: str = "GET", stream: bool = False,
+                data: typing.Optional[ParamsType] = None,
+                params: typing.Optional[ParamsType] = None,
+                raise_for_status: bool = True, timeout: typing.Optional[int] = None) \
+            -> typing.Optional[Response]:
         """
         Perform HTTP(S) request
 
@@ -437,7 +440,8 @@ def handle_params(self, url: str, method: str, params: ParamsType) \
             if value is not None
         ).encode()
 
-    def download(self, url, destdir, destfile=None, overwrite=False, **params):
+    def download(self, url: str, destdir: Path, destfile: typing.Optional[str] = None,
+                 overwrite: bool = False, **params: ParamsType) -> Path:
         """
         Shortcut for a streaming GET request
 
@@ -472,7 +476,7 @@ def download(self, url, destdir, destfile=None, overwrite=False, **params):
 
         return target
 
-    def get_objectified_xml(self, response):
+    def get_objectified_xml(self, response: Response) -> ObjectifiedElement:
         """
         Return API response as an XML object
 

osctiny/utils/base.py

@@ -4,6 +4,8 @@
 """
 # pylint: disable=too-few-public-methods,
 import os
+import typing
+from pathlib import Path
 
 from lxml.etree import tounicode
 
@@ -12,7 +14,7 @@ class ExtensionBase:
     """
     Base class for extensions of the :py:class:`Ocs` entry point.
     """
-    def __init__(self, osc_obj):
+    def __init__(self, osc_obj: "Osc"):
         self.osc = osc_obj
 
 
@@ -25,7 +27,8 @@ class DataDir:
     osclib_version_string = "1.0"
 
     # pylint: disable=too-many-arguments
-    def __init__(self, osc, path, project, package=None, overwrite=False):
+    def __init__(self, osc: "Osc", path: Path, project: str, package: typing.Optional[str] = None,
+                 overwrite: bool = False):
         self.osc = osc
         self.path = os.path.join(path, self.data_dir)
         self.project = project
@@ -42,7 +45,7 @@ def __init__(self, osc, path, project, package=None, overwrite=False):
         if overwrite:
             self.write_dir_contents()
 
-    def write_dir_contents(self):
+    def write_dir_contents(self) -> None:
         """
         Create files with default content in ``.osc`` sub-directory
         """