✨ feat: implement RFC 9457 error handling

.github/workflows/Pyinstaller-windows.yml

@@ -11,41 +11,28 @@ jobs:
     runs-on: windows-latest
     steps:
     - uses: actions/checkout@v5
+      with:
+        submodules: true
 
-    - uses: msys2/setup-msys2@v2
+    - uses: actions/setup-python@v6
       with:
-        msystem: MINGW64
-        install: >-
-          git
-          base-devel
-          zip
-        pacboy:
-          openssl
-          python-pip
+        python-version: '3.13'
 
-    - shell: msys2 {0}
+    - name: build and package
       run: |
-        git submodule update --init
         python --version
         python -m pip install pyinstaller
         python -m pip install .
-        pyinstaller --distpath ./pkg --clean --name osc-cli osc_sdk/sdk.py --add-data /mingw64/bin/libcrypto-3-x64.dll:. --add-data /mingw64/bin/libssl-3-x64.dll:.
-        zip -r osc-cli-x86_64.zip pkg/osc-cli
+        pyinstaller --distpath ./pkg --clean --name osc-cli osc_sdk/sdk.py
+        Compress-Archive -Path pkg/osc-cli -DestinationPath osc-cli-x86_64.zip
+
+    - name: smoke test
+      run: |
         ./pkg/osc-cli/osc-cli.exe api ReadRegions | grep api.eu-west-2.outscale.com
-    - name: Upload artifacts
-      uses: actions/upload-artifact@v5
-      if: ${{ github.event_name != 'push' }}
-      with:
-        name: osc-cli-win
-        path: |
-          osc-cli-x86_64.zip
-    - name: upload nightly
-      uses: "marvinpinto/action-automatic-releases@latest"
-      if: ${{ github.event_name == 'push' }}
+
+    - name: Upload artifact
+      if: github.event_name == 'push'
+      uses: actions/upload-artifact@v4
       with:
-        repo_token: "${{ secrets.GITHUB_TOKEN }}"
-        prerelease: true
-        automatic_release_tag: "nightly-windows"
-        title: "Windows Development Build"
-        files: |
-          osc-cli-x86_64.zip
+        name: osc-cli-win64
+        path: osc-cli-x86_64.zip

osc_sdk/__init__.py

@@ -1 +1,11 @@
-from .sdk import DirectLinkCall, EimCall, FcuCall, IcuCall, LbuCall, OKMSCall, OSCCall
+from .sdk import (
+    DirectLinkCall,
+    EimCall,
+    FcuCall,
+    IcuCall,
+    LbuCall,
+    LegacyProblem,
+    OKMSCall,
+    OSCCall,
+    Problem,
+)

osc_sdk/sdk.py

@@ -19,6 +19,145 @@
 from requests.models import Response
 from typing_extensions import TypedDict
 
+
+class ProblemDecoder(json.JSONDecoder):
+    def decode(self, s):
+        data = super().decode(s)
+        if isinstance(data, dict):
+            return self._make_problem(data)
+        return data
+
+    def _make_problem(self, data):
+        type_ = data.pop("type", None)
+        status = data.pop("status", None)
+        title = data.pop("title", None)
+        detail = data.pop("detail", None)
+        instance = data.pop("instance", None)
+        return Problem(type_, status, title, detail, instance, **data)
+
+
+class Problem(Exception):
+    def __init__(self, type_, status, title, detail, instance, **kwargs):
+        self._type = type_ or "about:blank"
+        self.status = status
+        self.title = title
+        self.detail = detail
+        self.instance = instance
+        self.extras = kwargs
+
+        for k in self.extras:
+            if k in ["type", "status", "title", "detail", "instance"]:
+                raise ValueError(f"Reserved key '{k}' used in Problem extra arguments.")
+
+    def __str__(self):
+        return self.title
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}<type={self._type}; status={self.status}; title={self.title}>"
+
+    def msg(self):
+        msg = (
+            f"type = {self._type}, "
+            f"status = {self.status}, "
+            f"title = {self.title}, "
+            f"detail = {self.detail}, "
+            f"instance = {self.instance}, "
+            f"extras = {self.extras}"
+        )
+        return msg
+
+    @property
+    def type(self):
+        return self._type
+
+
+class LegacyProblemDecoder(json.JSONDecoder):
+    def decode(self, s):
+        data = super().decode(s)
+        if isinstance(data, dict):
+            return self._make_legacy_problem(data)
+        return data
+
+    def _make_legacy_problem(self, data):
+        request_id = None
+        error_code = None
+        code_type = None
+
+        if "__type" in data:
+            error_code = data.get("__type")
+        else:
+            request_id = (data.get("ResponseContext") or {}).get("RequestId")
+            errors = data.get("Errors")
+            if errors:
+                error = errors[0]
+                error_code = error.get("Code")
+                reason = error.get("Type")
+                if error.get("Details"):
+                    code_type = reason
+                else:
+                    code_type = None
+        return LegacyProblem(None, error_code, code_type, request_id, None)
+
+
+class LegacyProblem(Exception):
+    def __init__(self, status, error_code, code_type, request_id, url):
+        self.status = status
+        self.error_code = error_code
+        self.code_type = code_type
+        self.request_id = request_id
+        self.url = url
+
+    def msg(self):
+        msg = (
+            f"status = {self.status}, "
+            f"code = {self.error_code}, "
+            f"{'code_type = ' if self.code_type is not None else ''}"
+            f"{self.code_type + ', ' if self.code_type is not None else ''}"
+            f"request_id = {self.request_id}, "
+            f"url = {self.url}"
+        )
+        return msg
+
+
+def api_error(response):
+    try:
+        problem = None
+        ct = response.headers.get("content-type") or ""
+        if "application/json" in ct:
+            problem: LegacyProblem = response.json(cls=LegacyProblemDecoder)
+            problem.status = problem.status or str(response.status_code)
+            problem.url = response.url
+        elif "application/problem+json" in ct:
+            problem: Problem = response.json(cls=ProblemDecoder)
+            problem.status = problem.status or str(response.status_code)
+
+        if problem:
+            return problem
+    except json.JSONDecodeError:
+        # If it is not JSON, pass
+        pass
+
+    try:
+        error = ET.fromstring(response.text)
+
+        err = dict()
+        for key, attr in [
+            ("Code", "error_code"),
+            ("Message", "status"),
+            ("RequestId", "request_id"),
+            ("RequestID", "request_id"),
+        ]:
+            value = next((x.text for x in error.iter() if x.tag.endswith(key)), None)
+            if value:
+                err[attr] = value
+
+        return LegacyProblem(**err)
+    except:
+        raise Exception(
+            f"Could not decode error response from {response.url} with status code {response.status_code}"
+        )
+
+
 CANONICAL_URI = "/"
 CONFIGURATION_FILE = "config.json"
 CONFIGURATION_FOLDER = ".osc"
@@ -121,77 +260,6 @@ class Tag(TypedDict):
     Values: List[str]
 
 
-@dataclass
-class OscApiException(Exception):
-    http_response: InitVar[Response]
-
-    status_code: int = field(init=False)
-    error_code: Optional[str] = field(default=None, init=False)
-    message: Optional[str] = field(default=None, init=False)
-    code_type: Optional[str] = field(default=None, init=False)
-    request_id: Optional[str] = field(default=None, init=False)
-
-    def __post_init__(self, http_response: Response):
-        super().__init__()
-        self.status_code = http_response.status_code
-        # Set error details
-        self._set(http_response)
-
-    def __str__(self) -> str:
-        return (
-            f"Error --> status = {self.status_code}, "
-            f"code = {self.error_code}, "
-            f"{'code_type = ' if self.code_type is not None else ''}"
-            f"{self.code_type + ', ' if self.code_type is not None else ''}"
-            f"Reason = {self.message}, "
-            f"request_id = {self.request_id}"
-        )
-
-    def _set(self, http_response: Response):
-        content = http_response.content.decode()
-        # In case it is JSON error format
-        try:
-            error = json.loads(content)
-        except json.JSONDecodeError:
-            pass
-        else:
-            if "__type" in error:
-                self.error_code = error.get("__type")
-                self.message = error.get("message")
-                self.request_id = http_response.headers.get("x-amz-requestid")
-            else:
-                self.request_id = (error.get("ResponseContext") or {}).get("RequestId")
-                errors = error.get("Errors")
-                if errors:
-                    error = errors[0]
-                    self.error_code = error.get("Code")
-                    self.message = error.get("Type")
-                    if error.get("Details"):
-                        self.code_type = self.message
-                        self.message = error.get("Details")
-                    else:
-                        self.code_type = None
-            return
-
-        # In case it is XML format
-        try:
-            error = ET.fromstring(content)
-        except ET.ParseError:
-            return
-        else:
-            for key, attr in [
-                ("Code", "error_code"),
-                ("Message", "message"),
-                ("RequestId", "request_id"),
-                ("RequestID", "request_id"),
-            ]:
-                value = next(
-                    (x.text for x in error.iter() if x.tag.endswith(key)), None
-                )
-                if value:
-                    setattr(self, attr, value)
-
-
 @dataclass
 class ApiCall:
     profile: str = DEFAULT_PROFILE
@@ -437,7 +505,7 @@ def make_request(self, call: str, **kwargs: CallParameters):
 class XmlApiCall(ApiCall):
     def get_response(self, http_response: Response) -> Union[str, ResponseContent]:
         if http_response.status_code not in SUCCESS_CODES:
-            raise OscApiException(http_response)
+            raise api_error(http_response)
         try:
             return cast(ResponseContent, xmltodict.parse(http_response.content))
         except Exception:
@@ -502,7 +570,7 @@ def get_parameters(
 
     def get_response(self, http_response: Response) -> ResponseContent:
         if http_response.status_code not in SUCCESS_CODES:
-            raise OscApiException(http_response)
+            raise api_error(http_response)
 
         return json.loads(http_response.text)
 
@@ -641,7 +709,7 @@ class DirectLinkCall(JsonApiCall):
 
     def get_response(self, http_response: Response) -> ResponseContent:
         if http_response.status_code not in SUCCESS_CODES:
-            raise OscApiException(http_response)
+            raise api_error(http_response)
 
         res = json.loads(http_response.text)
         res["requestid"] = http_response.headers["x-amz-requestid"]

osc_sdk/test_auth.py

@@ -10,7 +10,7 @@
 class Env(object):
     access_key: str
     secret_key: str
-    endpoint_icu: str
+    endpoint_api: str
     region: str
 
 
@@ -19,17 +19,17 @@ def env() -> Env:
     return Env(
         access_key=os.getenv("OSC_TEST_ACCESS_KEY", ""),
         secret_key=os.getenv("OSC_TEST_SECRET_KEY", ""),
-        endpoint_icu=os.getenv("OSC_TEST_ENDPOINT_ICU", ""),
+        endpoint_api=os.getenv("OSC_TEST_ENDPOINT_API", ""),
         region=os.getenv("OSC_TEST_REGION", ""),
     )
 
 
-def test_icu_auth_ak_sk(env):
-    icu = sdk.IcuCall(
+def test_api_auth_ak_sk(env):
+    api = sdk.OSCCall(
         access_key=env.access_key,
         secret_key=env.secret_key,
-        endpoint=env.endpoint_icu,
+        endpoint=env.endpoint_api,
         region_name=env.region,
     )
-    icu.make_request("GetAccount")
-    assert len(icu.response) > 0
+    api.make_request("ReadAccounts")
+    assert len(api.response) > 0

osc_sdk/test_errors.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from . import sdk
+from . import LegacyProblem, sdk
 
 
 @dataclass
@@ -32,6 +32,6 @@ def test_bad_filter(env):
         endpoint=env.endpoint_api,
         region_name=env.region,
     )
-    with pytest.raises(sdk.OscApiException) as e:
+    with pytest.raises(LegacyProblem) as e:
         oapi.make_request("ReadImages", Filters='"bad_filter"')
-    assert e.value.status_code == 400
+    assert e.value.status == "400"