🐛 fix: sonar issues

.github/workflows/pythonpackage.yml

@@ -2,23 +2,25 @@ name: Python package
 
 on:
   push:
+    branches:
+      - "main"
+  pull_request:
+    branches:
+      - "**"
+    types: [opened, synchronize, reopened]
+  create:
     branches:
       - "**"
 
 jobs:
   python-lint:
-    strategy:
-      matrix:
-        python-version: ["3.10"]
-        platform: [ubuntu-latest]
-      fail-fast: false
-    runs-on: ${{ matrix.platform }}
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
       - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: ${{ matrix.python-version }}
+          python-version: "3.10"
       - uses: actions/cache@v4
         id: cache
         with:
@@ -33,6 +35,11 @@ jobs:
         run: python -m pdm install -G lint -G dev
       - name: Lint
         run: python -m pdm run lint
+      - name: Archive lint reports
+        uses: actions/upload-artifact@v4
+        with:
+          name: lint-reports
+          path: reports
 
   python-test:
     strategy:
@@ -71,6 +78,7 @@ jobs:
         with:
           name: coverage-${{ matrix.platform }}-${{ matrix.python-version }}
           path: reports/.coverage
+          include-hidden-files: true
 
   coveralls-finish:
     needs: [python-test]
@@ -85,13 +93,27 @@ jobs:
 
   sonarcloud:
     needs: [python-test]
-    if: ${{ always() }}
     name: SonarCloud
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+      - name: Install coverage
+        run: pip install coverage[toml]
+      - name: Download artifacts
+        uses: actions/download-artifact@v4
+        with:
+          name: coverage-ubuntu-latest-3.12
+          path: reports
+      - name: Create coverage XML
+        run: |
+          ls -al reports
+          coverage xml
       - name: Generate sonar properties
         run: |
           cat << EOF > sonar-project.properties

pyproject.toml

@@ -59,16 +59,19 @@ warn_unreachable = true
 branch = true
 command_line = "--module pytest"
 data_file = "reports/.coverage"
-source = ["src"]
+include = ["src/*"]
+omit = ["tests/*"]
 
 [tool.coverage.paths]
-source = ["src/", "/home/runner/**/src", "D:\\**\\src"]
+source = ["src/"]
 
 [tool.coverage.report]
 fail_under = 50
 precision = 1
 show_missing = true
 skip_covered = true
+include = ["src/*"]
+omit = ["tests/*"]
 
 [tool.coverage.xml]
 output = "reports/coverage.xml"

src/coinapi/_hooks/hooks.py

@@ -24,9 +24,6 @@ class SDKHooks(Hooks):
     after_success_hooks: ClassVar[list[AfterSuccessHook]] = []
     after_error_hooks: ClassVar[list[AfterErrorHook]] = []
 
-    def __init__(self) -> None:
-        pass
-
     def register_sdk_init_hook(self, hook: SDKInitHook) -> None:
         """Register an SDK init hook."""
         self.sdk_init_hooks.append(hook)

src/coinapi/base.py

@@ -1,14 +1,14 @@
 """Base class for operation collections."""
 
 import enum
-from typing import TypeVar
+from typing import Any, TypeVar
 
 import httpx
 import msgspec
 from httpx import codes
 
 from coinapi import utils
-from coinapi._hooks import HookContext
+from coinapi._hooks import BeforeRequestContext, HookContext
 from coinapi.config import CoinAPIConfig
 from coinapi.models import errors
 from coinapi.models.operations.base import CoinAPIRequest, CoinAPIResponse
@@ -34,119 +34,161 @@ class Base:
     def __init__(self, sdk_config: CoinAPIConfig) -> None:
         self.sdk_configuration = sdk_config
 
-    def _make_request(  # noqa: PLR0912, C901
+    def _make_request(  # type: ignore[return]
         self,
         operation_id: str,
         request: RequestT,
         response_cls: type[ResponseT],
         accept_header_override: AcceptEnum | None = None,
     ) -> ResponseT:
-        """Send a request."""
-        hook_ctx = HookContext(
+        """Send an HTTP request."""
+        hook_ctx = self._create_hook_context(operation_id)
+        prepared_request = self._prepare_request(request, accept_header_override)
+        client = self._configure_security_client()
+
+        try:
+            http_res = self._execute_request(hook_ctx, prepared_request, client)
+            return self._process_response(http_res, response_cls)
+        except Exception as e:  # noqa: BLE001
+            self._handle_request_error(hook_ctx, e)
+
+    def _create_hook_context(self, operation_id: str) -> BeforeRequestContext:
+        """Create a hook context."""
+        return BeforeRequestContext(
             operation_id=operation_id,
             oauth2_scopes=[],
             security_source=self.sdk_configuration.security,
         )
+
+    def _prepare_request(
+        self,
+        request: RequestT,
+        accept_header_override: AcceptEnum | None,
+    ) -> httpx.Request:
+        """Prepare an HTTP request."""
         base_url = utils.template_url(*self.sdk_configuration.get_server_details())
-        url = utils.generate_url(type(request), base_url, request.endpoint, request)  # type: ignore[arg-type]
+        url = utils.generate_url(type(request), base_url, request.endpoint, request)
+        headers = self._prepare_headers(request, accept_header_override)
+        data, form = self._prepare_body(request)
+        query_params = utils.get_query_params(type(request), request) or None
+
+        return httpx.Request(
+            request.method,
+            url,
+            params=query_params,
+            data=data,
+            files=form,
+            headers=headers,
+        )
+
+    def _prepare_headers(
+        self,
+        request: RequestT,
+        accept_header_override: AcceptEnum | None,
+    ) -> dict[str, str]:
+        """Prepare request headers."""
         headers = {}
-        data, form = None, None
         if request.method in {"POST", "PUT", "PATCH"}:
-            req_content_type, data, form = utils.serialize_request_body(
-                request,
-                "body",
-            )
+            req_content_type, _, _ = utils.serialize_request_body(request, "body")
             if req_content_type is not None and req_content_type not in (
                 "multipart/form-data",
                 "multipart/mixed",
             ):
                 headers["content-type"] = req_content_type
-        query_params = utils.get_query_params(type(request), request) or None  # type: ignore[arg-type]
-        if accept_header_override is not None:
-            headers["Accept"] = accept_header_override.value
-        else:
-            headers["Accept"] = (
-                "application/json;q=1, text/json;q=0.8, text/plain;q=0.5, application/x-msgpack;q=0"
-            )
+
+        headers["Accept"] = (
+            accept_header_override.value
+            if accept_header_override is not None
+            else "application/json;q=1, text/json;q=0.8, text/plain;q=0.5, application/x-msgpack;q=0"
+        )
         headers["user-agent"] = self.sdk_configuration.user_agent
+        return headers
 
+    def _prepare_body(self, request: RequestT) -> tuple[Any, Any]:
+        """Prepare request body."""
+        if request.method in {"POST", "PUT", "PATCH"}:
+            _, data, form = utils.serialize_request_body(request, "body")
+            return data, form
+        return None, None
+
+    def _configure_security_client(self) -> utils.SecurityClient:
+        """Configure the security client."""
         security = (
             self.sdk_configuration.security()
             if callable(self.sdk_configuration.security)
             else self.sdk_configuration.security
         )
-        client = utils.configure_security_client(
-            self.sdk_configuration.client,
-            security,
-        )
+        return utils.configure_security_client(self.sdk_configuration.client, security)
 
-        try:
-            req = self.sdk_configuration.get_hooks().before_request(
-                hook_ctx,  # type: ignore[arg-type]
-                httpx.Request(
-                    request.method,
-                    url,
-                    params=query_params,
-                    data=data,
-                    files=form,
-                    headers=headers,
-                ),
-            )
-            http_res = client.send(req)
-        except Exception as e:
-            _, exc = self.sdk_configuration.get_hooks().after_error(hook_ctx, None, e)  # type: ignore[arg-type]
-            raise exc from e  # type: ignore[misc]
+    def _execute_request(
+        self,
+        hook_ctx: BeforeRequestContext,
+        prepared_request: httpx.Request,
+        client: utils.SecurityClient,
+    ) -> httpx.Response:
+        """Execute an HTTP request."""
+        req = self.sdk_configuration.get_hooks().before_request(
+            hook_ctx,
+            prepared_request,
+        )
+        return client.send(req)
 
+    def _process_response(
+        self,
+        http_res: httpx.Response,
+        response_cls: type[ResponseT],
+    ) -> ResponseT:
+        """Process an HTTP response."""
         if utils.match_status_codes(["4XX", "5XX"], http_res.status_code):
-            http_res, exc = self.sdk_configuration.get_hooks().after_error(  # type: ignore[assignment]
-                hook_ctx,  # type: ignore[arg-type]
-                http_res,
-                None,
-            )
-            if exc:
-                raise exc
-        else:
-            result = self.sdk_configuration.get_hooks().after_success(
-                hook_ctx,  # type: ignore[arg-type]
-                http_res,
-            )
-            if isinstance(result, Exception):
-                raise result
-            http_res = result
+            self._handle_error_response(http_res)
 
         content_type = http_res.headers.get("Content-Type", "")
-
         res = response_cls(
             status_code=http_res.status_code,
             content_type=content_type,
             raw_response=http_res,
         )
 
         if httpx.codes.is_success(http_res.status_code):
-            if utils.match_content_type(content_type, "text/plain"):
-                res.content_plain = http_res.text
-            elif utils.match_content_type(
-                content_type,
-                "application/json",
-            ) or utils.match_content_type(content_type, "text/json"):
-                content_cls = next(
-                    field
-                    for field in msgspec.structs.fields(response_cls)
-                    if field.name == "content"
-                ).type
-                out = msgspec.json.decode(http_res.content, type=content_cls)
-                res.content = out
-            elif utils.match_content_type(content_type, "application/x-msgpack"):
-                res.body = http_res.content
-            else:
-                msg = f"unknown content-type received: {content_type}"
-                raise errors.CoinAPIError(
-                    msg,
-                    http_res.status_code,
-                    http_res.text,
-                    http_res,
-                )
-        elif codes.is_client_error(http_res.status_code) or codes.is_server_error(
+            self._set_response_content(res, http_res, content_type, response_cls)
+
+        return res
+
+    def _set_response_content(
+        self,
+        res: ResponseT,
+        http_res: httpx.Response,
+        content_type: str,
+        response_cls: type[ResponseT],
+    ) -> None:
+        """Set the response content."""
+        if utils.match_content_type(content_type, "text/plain"):
+            res.content_plain = http_res.text
+        elif utils.match_content_type(
+            content_type,
+            "application/json",
+        ) or utils.match_content_type(content_type, "text/json"):
+            content_cls = next(
+                field
+                for field in msgspec.structs.fields(response_cls)
+                if field.name == "content"
+            ).type
+            out = msgspec.json.decode(http_res.content, type=content_cls)
+            res.content = out
+        elif utils.match_content_type(content_type, "application/x-msgpack"):
+            res.body = http_res.content
+        else:
+            msg = f"unknown content-type received: {content_type}"
+            raise errors.CoinAPIError(
+                msg,
+                http_res.status_code,
+                http_res.text,
+                http_res,
+            )
+
+    def _handle_error_response(self, http_res: httpx.Response) -> None:
+        """Handle an error response."""
+        if codes.is_client_error(http_res.status_code) or codes.is_server_error(
             http_res.status_code,
         ):
             raise errors.CoinAPIError(
@@ -156,4 +198,7 @@ def _make_request(  # noqa: PLR0912, C901
                 http_res,
             )
 
-        return res
+    def _handle_request_error(self, hook_ctx: HookContext, error: Exception) -> None:
+        """Handle a request error."""
+        _, exc = self.sdk_configuration.get_hooks().after_error(hook_ctx, None, error)  # type: ignore[arg-type]
+        raise exc from error  # type: ignore[misc]