Update upstream racetrack libs (#27)

* Merge upstream changes from racetrack

* Switch async endpoints to non-blocking

* Fix Path parameters cannot have a default value

* Add type import compatible with python3.8

* Fix fastapi ENCODERS_BY_TYPE

Author: iszulcdeepsense

docs/compatibility.md

@@ -8,3 +8,4 @@ This document describes compatibility of the versions of this plugin with the Ra
 | 2.9.0          | `>= 2.18.0`                  |
 | 2.9.1          | `>= 2.19.0`                  |
 | 2.9.2          | `>= 2.19.0`                  |
+| 2.10.0         | `>= 2.19.0`                  |

python3-job-type/plugin-manifest.yaml

@@ -1,4 +1,4 @@
 name: python3-job-type
-version: 2.9.2
+version: 2.10.0
 url: 'https://github.com/TheRacetrack/plugin-python-job-type'
 category: 'job-type'

python3-job-type/python_wrapper/racetrack_client/log/context_error.py

@@ -40,3 +40,10 @@ def wrap_context(context_name: str, log_debug: bool = False):
         yield
     except Exception as e:
         raise ContextError(context_name) from e
+
+
+def unwrap(e: BaseException) -> BaseException:
+    """Find root cause of the error"""
+    while e.__cause__ is not None:
+        e = e.__cause__
+    return e

python3-job-type/python_wrapper/racetrack_client/log/errors.py

@@ -4,3 +4,7 @@ class EntityNotFound(RuntimeError):
 
 class AlreadyExists(RuntimeError):
     pass
+
+
+class ValidationError(RuntimeError):
+    pass

python3-job-type/python_wrapper/racetrack_client/log/logs.py

@@ -3,6 +3,7 @@
 from typing import Optional
 
 LOG_FORMAT = '\033[2m[%(asctime)s]\033[0m %(levelname)s %(message)s'
+LOG_FORMAT_DEBUG = '\033[2m[%(asctime)s]\033[0m %(name)s %(filename)s %(lineno)s %(levelname)s %(message)s'
 LOG_DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
 
 

python3-job-type/python_wrapper/racetrack_client/requirements.txt

@@ -1,4 +1,6 @@
-PyYAML >= 5.4
-pydantic >= 1.9, < 2
+PyYAML >= 6.0
+pydantic >= 1.10, < 2
 backoff >= 2.2
-python-socketio >= 5.7
+python-socketio[client] >= 5.8
+typer[all] >= 0.7.0
+jsonschema >= 4.17.3

python3-job-type/python_wrapper/racetrack_client/utils/datamodel.py

@@ -11,7 +11,7 @@
 
 
 def parse_dict_datamodel(
-    obj_dict: Dict, 
+    obj_dict: Dict,
     clazz: Type[T], 
 ) -> T:
     """
@@ -70,6 +70,12 @@ def convert_to_json(obj) -> str:
     return json.dumps(obj)
 
 
+def convert_to_yaml(obj) -> str:
+    obj = convert_to_json_serializable(obj)
+    obj = remove_none(obj)
+    return yaml.dump(obj, sort_keys=False)
+
+
 def datamodel_to_dict(dt: BaseModel) -> Dict:
     data_dict = dt.dict()
     data_dict = remove_none(data_dict)

python3-job-type/python_wrapper/racetrack_client/utils/quantity.py

@@ -43,7 +43,7 @@ def __init__(self, quantity_str: str):
         self.base_number: float = float(match.group('number'))
         self.suffix: str = match.group('suffix')
 
-        if not self.suffix in self._suffix_multipliers:
+        if self.suffix not in self._suffix_multipliers:
             raise ValueError(f'invalid suffix: {self.suffix}, use one of {self._suffix_multipliers.keys()}')
 
     def __str__(self) -> str:
@@ -81,3 +81,22 @@ def __truediv__(self, other) -> 'Quantity':
     def plain_number(self) -> float:
         """Convert quantity to a plain number without any suffixes"""
         return self.base_number * self._suffix_multipliers[self.suffix]
+
+    @classmethod
+    def __get_validators__(cls):
+        # Needed to make it compatible with pydantic's Custom Data Types
+        # https://docs.pydantic.dev/usage/types/#custom-data-types
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, v):
+        if v is None:
+            return None
+        return Quantity(str(v))
+
+    @classmethod
+    def __modify_schema__(cls, field_schema):
+        # Needed to make it compatible with pydantic's Custom Data Types
+        field_schema.update(
+            examples=["128974848", "129M", "128974848000m", "123Mi", "0.129G"],
+        )

python3-job-type/python_wrapper/racetrack_client/utils/request.py

@@ -3,14 +3,16 @@
 from email.message import Message
 from http.client import HTTPResponse
 from http.client import responses
-from typing import Any, Optional, Dict
+from typing import Any, List, Optional, Dict, Union
 from urllib import request
 from urllib.error import HTTPError, URLError
 import urllib.parse as urlparse
 from urllib.parse import urlencode
 
 from racetrack_client.log.context_error import ContextError
 
+DEBUG_MODE = False  # print the actual bytes sent in requests and responses
+
 
 class RequestError(ContextError):
     """HTTP Request error due to network error, bad URL, unreachable address"""
@@ -27,13 +29,16 @@ def __init__(self, error_context: str, status_code: int):
 
 class Response:
 
-    def __init__(self,
+    def __init__(
+        self,
         url: str,
+        method: str,
         status_code: int,
         content: bytes,
         headers: Message,
     ):
         self._url: str = url
+        self._method: str = method.upper()
         self._status_code: int = status_code
         self._content: bytes = content
         self._headers: Message = headers
@@ -62,7 +67,7 @@ def content(self) -> bytes:
     def text(self) -> str:
         return self._content.decode('utf8')
 
-    def json(self) -> Dict:
+    def json(self) -> Union[Dict, List]:
         response_data = self._content.decode('utf8')
         if not response_data:
             raise RuntimeError('no content in response to decode as JSON')
@@ -76,82 +81,96 @@ def is_json(self) -> bool:
     def url(self) -> str:
         return self._url
 
+    @property
+    def method(self) -> str:
+        return self._method
+    
     @property
     def headers(self) -> Message:
         return self._headers
 
     def header(self, name: str) -> Optional[str]:
         return self._headers[name]
 
+    @property
+    def error_details(self) -> str:
+        return f'{self.status_code} {self.status_reason} for url: {self.method} {self.url}'
+
 
 class Requests:
 
     insecure: bool = True  # whether to verify SSL certificates
 
     @classmethod
-    def get(cls, 
+    def get(
+            cls,
             url: str,
             params: Optional[Dict[str, Any]] = None,
             headers: Optional[Dict[str, str]] = None,
-            timeout: float = None,
-        ) -> Response:
+            timeout: Optional[float] = None,
+            ) -> Response:
         return cls._make_request('GET', url, None, None, params, headers, timeout)
 
     @classmethod
-    def post(cls,
+    def post(
+            cls,
             url: str,
             json: Optional[Any] = None,
             data: Optional[bytes] = None,
             params: Optional[Dict[str, Any]] = None,
             headers: Optional[Dict[str, str]] = None,
-            timeout: float = None,
-        ) -> Response:
+            timeout: Optional[float] = None,
+            ) -> Response:
         return cls._make_request('POST', url, json, data, params, headers, timeout)
 
     @classmethod
-    def put(cls,
+    def put(
+            cls,
             url: str,
             json: Optional[Any] = None,
             data: Optional[bytes] = None,
             params: Optional[Dict[str, Any]] = None,
             headers: Optional[Dict[str, str]] = None,
-            timeout: float = None,
-        ) -> Response:
+            timeout: Optional[float] = None,
+            ) -> Response:
         return cls._make_request('PUT', url, json, data, params, headers, timeout)
 
     @classmethod
-    def delete(cls,
+    def delete(
+            cls,
             url: str,
             json: Optional[Any] = None,
             data: Optional[bytes] = None,
             params: Optional[Dict[str, Any]] = None,
             headers: Optional[Dict[str, str]] = None,
-            timeout: float = None,
-        ) -> Response:
+            timeout: Optional[float] = None,
+            ) -> Response:
         return cls._make_request('DELETE', url, json, data, params, headers, timeout)
 
     @classmethod
-    def request(cls,
+    def request(
+            cls,
             method: str,
             url: str,
             json: Optional[Any] = None,
             data: Optional[bytes] = None,
             params: Optional[Dict[str, Any]] = None,
             headers: Optional[Dict[str, str]] = None,
-            timeout: float = None,
-        ) -> Response:
+            timeout: Optional[float] = None,
+            ) -> Response:
         return cls._make_request(method.upper(), url, json, data, params, headers, timeout)
 
     @classmethod
-    def _make_request(cls, 
+    def _make_request(
+            cls,
             method: str,
             url: str,
             jsondata: Optional[Any] = None,
             data: Optional[bytes] = None,
             params: Optional[Dict[str, Any]] = None,
             headers: Optional[Dict[str, str]] = None,
-            timeout: float = None,
-        ) -> Response:
+            timeout: Optional[float] = None,
+            ) -> Response:
         """
         Make HTTP request and return response object.
         :param method: HTTP method: GET, POST, PUT, DELETE
@@ -190,19 +209,33 @@ def _make_request(cls,
         if timeout is not None:
             kwargs['timeout'] = timeout
 
+        if not req.has_header('User-Agent'):
+            req.add_header('User-Agent', 'request')
+
         kwargs['context'] = cls._get_ssl_context()
 
+        if DEBUG_MODE:
+            http_handler = request.HTTPHandler(debuglevel=2)
+            https_handler = request.HTTPSHandler(context=kwargs['context'], debuglevel=2)
+            opener = request.build_opener(http_handler, https_handler)
+            request.install_opener(opener)
+            # Passing 'context' argument to urllib.request.urlopen rebuilds the URL opener,
+            # not giving a chance to turn the debug mode on.
+            del kwargs['context']
+
         try:
             http_response: HTTPResponse = request.urlopen(req, **kwargs)
             return Response(
                 url=url,
+                method=method,
                 status_code=http_response.status,
                 content=http_response.read(),
                 headers=http_response.headers,
             )
         except HTTPError as e:
             return Response(
                 url=url,
+                method=method,
                 status_code=e.code,
                 content=e.read(),
                 headers=e.headers,
@@ -232,7 +265,7 @@ def build_url_with_params(
     return urlparse.urlunparse(url_parts)
 
 
-def parse_response(response: Response, error_context: str) -> Optional[Dict]:
+def parse_response(response: Response, error_context: str) -> Optional[Union[Dict, List]]:
     """
     Ensure response was successful. If not, try to extract error message from it.
     :return: response parsed as JSON object
@@ -248,34 +281,53 @@ def parse_response(response: Response, error_context: str) -> Optional[Dict]:
     raise "Deployment error: 500 Internal Server Error: you have no power here"
     """
     try:
-        result: Optional[Dict] = None
+        result: Optional[Union[Dict, List]] = None
         if 'application/json' in response.headers['content-type']:
             result = response.json()
 
         if response.ok:
             return result
 
-        if result is not None and 'error' in result:
-            raise RuntimeError(f'{response.status_reason}: {result.get("error")}')
-        response.raise_for_status()
-        return result
+        if result is not None and isinstance(result, dict) and 'error' in result:
+            raise ContextError(response.status_reason, RuntimeError(result.get("error")))
+        raise ResponseError(response.error_details, response.status_code)
     except Exception as e:
         raise ResponseError(error_context, response.status_code) from e
 
 
 def parse_response_object(response: Response, error_context: str) -> Dict:
     try:
         if 'application/json' not in response.headers['content-type']:
-            raise RuntimeError('expected JSON response')
+            raise RuntimeError(f'expected JSON response, got "{response.headers["content-type"]}", '
+                               f'{response.error_details}')
+
+        result = response.json()
+
+        if response.ok:
+            assert isinstance(result, dict), f'response JSON expected to be a dictionary, got {type(result)}'
+            return result
+
+        if result is not None and isinstance(result, dict) and 'error' in result:
+            raise ContextError(response.status_reason, RuntimeError(result.get("error")))
+        raise ResponseError(response.error_details, response.status_code)
+    except Exception as e:
+        raise ResponseError(error_context, response.status_code) from e
+
+
+def parse_response_list(response: Response, error_context: str) -> List:
+    try:
+        if 'application/json' not in response.headers['content-type']:
+            raise RuntimeError(f'expected JSON response, got "{response.headers["content-type"]}", '
+                               f'{response.error_details}')
 
         result = response.json()
 
         if response.ok:
+            assert isinstance(result, list), f'response JSON expected to be a list, got {type(result)}'
             return result
 
-        if result is not None and 'error' in result:
-            raise RuntimeError(f'{response.status_reason}: {result.get("error")}')
-        response.raise_for_status()
-        return result
+        if result is not None and isinstance(result, dict) and 'error' in result:
+            raise ContextError(response.status_reason, RuntimeError(result.get("error")))
+        raise ResponseError(response.error_details, response.status_code)
     except Exception as e:
         raise ResponseError(error_context, response.status_code) from e

python3-job-type/python_wrapper/racetrack_commons/api/asgi/access_log.py

@@ -1,7 +1,10 @@
+import time
+
 from fastapi import FastAPI, Request, Response
 
 from racetrack_client.log.logs import get_logger
 from racetrack_commons.api.asgi.asgi_server import HIDDEN_ACCESS_LOGS
+from racetrack_commons.api.metrics import metric_request_duration, metric_requests_done, metric_requests_started
 from racetrack_commons.api.tracing import RequestTracingLogger, get_caller_header_name, get_tracing_header_name
 
 logger = get_logger(__name__)
@@ -45,6 +48,8 @@ def enable_response_access_log(fastapi_app: FastAPI):
 
     @fastapi_app.middleware('http')
     async def access_log(request: Request, call_next) -> Response:
+        metric_requests_started.inc()
+        start_time = time.time()
         try:
             response: Response = await call_next(request)
         except RuntimeError as exc:
@@ -59,6 +64,9 @@ async def access_log(request: Request, call_next) -> Response:
                         logger.error(f"Request cancelled by the client: {method} {uri}")
                     return Response(status_code=204)  # No Content
             raise
+        finally:
+            metric_request_duration.observe(time.time() - start_time)
+            metric_requests_done.inc()
 
         method = request.method
         uri = request.url.replace(scheme='', netloc='')