Dateien nach "docs/Zierl" hochladen

2026-02-09 11:37:32 +00:00
parent 84dd542f84
commit 63de2d2dcf
5 changed files with 629 additions and 0 deletions
--- a/docs/Zierl/Ziel_3.txt
+++ b/docs/Zierl/Ziel_3.txt
@@ -0,0 +1,112 @@
+from typing import Dict, Optional, final
+
+import requests
+import xmltodict
+from rest_framework import status
+
+
+class BaseAPI:
+    __api_name__ = "BaseAPI"
+
+    def __init__(self):
+        pass
+
+    def get_host(self) -> str:
+        raise NotImplementedError("Subclasses must implement _get_host() method")
+
+    def build_headers(self) -> Dict[str, str]:
+        return {}
+
+    def build_params(self) -> Dict[str, str]:
+        return {}
+
+    def repair_authentication(self) -> bool:
+        """Callback to repair authentication, e.g. refresh token or re-login."""
+        return False
+
+    @final
+    def request(
+        self,
+        method: str,
+        path: str,
+        data: Optional[Dict | str | bytes] = None,
+        json: Optional[Dict] = None,
+        headers: Optional[Dict[str, str]] = None,
+        params: Optional[Dict[str, str]] = None,
+        custom_host: Optional[str] = None,
+        timeout: int = 30,
+        use_build_headers: bool = True,
+    ) -> Dict:
+        host = custom_host or self.get_host()
+        url = f"{host}{path}"
+        combined_headers: Optional[Dict[str, str]] = {
+            **(headers or {}),
+        }
+        if use_build_headers:
+            combined_headers = {
+                **self.build_headers(),
+                **(headers or {}),
+            }
+        combined_params: Optional[Dict[str, str]] = {
+            **self.build_params(),
+            **(params or {}),
+        }
+
+        if combined_headers == {}:
+            combined_headers = None
+        if combined_params == {}:
+            combined_params = None
+
+        try:
+            response = requests.request(
+                method=method,
+                url=url,
+                headers=combined_headers,
+                params=combined_params,
+                data=data,
+                json=json,
+                timeout=timeout,
+            )
+            if (
+                response.status_code == status.HTTP_401_UNAUTHORIZED
+                or response.status_code == status.HTTP_403_FORBIDDEN
+            ):
+                if self.repair_authentication():
+                    host = custom_host or self.get_host()
+                    url = f"{host}{path}"
+                    combined_headers: Optional[Dict[str, str]] = {
+                        **self.build_headers(),
+                        **(headers or {}),
+                    }
+                    combined_params: Optional[Dict[str, str]] = {
+                        **self.build_params(),
+                        **(params or {}),
+                    }
+                    response = requests.request(
+                        method=method,
+                        url=url,
+                        headers=combined_headers,
+                        params=combined_params,
+                        data=data,
+                        timeout=timeout,
+                    )
+            response.raise_for_status()
+            if response.status_code == status.HTTP_204_NO_CONTENT:
+                return {}
+
+            content_type = response.headers.get("Content-Type", "")
+
+            if "xml" in content_type:
+                return xmltodict.parse(response.text)
+            if "application/json" in content_type:
+                return response.json()
+            else:
+                raise ValueError(
+                    "Unsupported Content-Type: {content_type} must be application/json"
+                )
+
+        except requests.exceptions.RequestException as e:
+            raise e
+        except ValueError as e:
+            raise e
+