Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion services/observability/oas_commit
Original file line number Diff line number Diff line change
@@ -1 +1 @@
10774896634990c655a523f5f0fabd6e2be9e216
0e64886dd0847341800d7191ed193b75413be998
31 changes: 22 additions & 9 deletions services/observability/src/stackit/observability/api_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,13 @@
""" # noqa: E501

import datetime
import decimal
import json
import mimetypes
import os
import re
import tempfile
import uuid
from enum import Enum
from typing import Dict, List, Optional, Tuple, Union
from urllib.parse import quote
Expand Down Expand Up @@ -64,8 +66,10 @@ class ApiClient:
"bool": bool,
"date": datetime.date,
"datetime": datetime.datetime,
"decimal": decimal.Decimal,
"object": object,
}
_pool = None

def __init__(self, configuration, header_name=None, header_value=None, cookie=None) -> None:
self.config: Configuration = configuration
Expand Down Expand Up @@ -268,7 +272,7 @@ def response_deserialize(
return_data = self.__deserialize_file(response_data)
elif response_type is not None:
match = None
content_type = response_data.getheader("content-type")
content_type = response_data.headers.get("content-type")
if content_type is not None:
match = re.search(r"charset=([a-zA-Z\-\d]+)[\s;]?", content_type)
encoding = match.group(1) if match else "utf-8"
Expand All @@ -285,7 +289,7 @@ def response_deserialize(
return ApiResponse(
status_code=response_data.status,
data=return_data,
headers=response_data.getheaders(),
headers=response_data.headers,
raw_data=response_data.data,
)

Expand All @@ -297,6 +301,7 @@ def sanitize_for_serialization(self, obj):
If obj is str, int, long, float, bool, return directly.
If obj is datetime.datetime, datetime.date
convert to string in iso8601 format.
If obj is decimal.Decimal return string representation.
If obj is list, sanitize each element in the list.
If obj is dict, return the dict.
If obj is OpenAPI model, return the properties dict.
Expand All @@ -312,12 +317,16 @@ def sanitize_for_serialization(self, obj):
return obj.get_secret_value()
elif isinstance(obj, self.PRIMITIVE_TYPES):
return obj
elif isinstance(obj, uuid.UUID):
return str(obj)
elif isinstance(obj, list):
return [self.sanitize_for_serialization(sub_obj) for sub_obj in obj]
elif isinstance(obj, tuple):
return tuple(self.sanitize_for_serialization(sub_obj) for sub_obj in obj)
elif isinstance(obj, (datetime.datetime, datetime.date)):
return obj.isoformat()
elif isinstance(obj, decimal.Decimal):
return str(obj)

elif isinstance(obj, dict):
obj_dict = obj
Expand All @@ -327,7 +336,7 @@ def sanitize_for_serialization(self, obj):
# and attributes which value is not None.
# Convert attribute name to json key in
# model definition for request.
if hasattr(obj, "to_dict") and callable(obj.to_dict):
if hasattr(obj, "to_dict") and callable(getattr(obj, "to_dict")):
obj_dict = obj.to_dict()
else:
obj_dict = obj.__dict__
Expand Down Expand Up @@ -355,7 +364,7 @@ def deserialize(self, response_text: str, response_type: str, content_type: Opti
data = json.loads(response_text)
except ValueError:
data = response_text
elif re.match(r"^application/(json|[\w!#$&.+-^_]+\+json)\s*(;|$)", content_type, re.IGNORECASE):
elif re.match(r"^application/(json|[\w!#$&.+\-^_]+\+json)\s*(;|$)", content_type, re.IGNORECASE):
if response_text == "":
data = ""
else:
Expand Down Expand Up @@ -401,12 +410,14 @@ def __deserialize(self, data, klass):

if klass in self.PRIMITIVE_TYPES:
return self.__deserialize_primitive(data, klass)
elif klass == object:
elif klass is object:
return self.__deserialize_object(data)
elif klass == datetime.date:
elif klass is datetime.date:
return self.__deserialize_date(data)
elif klass == datetime.datetime:
elif klass is datetime.datetime:
return self.__deserialize_datetime(data)
elif klass is decimal.Decimal:
return decimal.Decimal(data)
elif issubclass(klass, Enum):
return self.__deserialize_enum(data, klass)
else:
Expand Down Expand Up @@ -554,12 +565,14 @@ def __deserialize_file(self, response):
os.close(fd)
os.remove(path)

content_disposition = response.getheader("Content-Disposition")
content_disposition = response.headers.get("Content-Disposition")
if content_disposition:
m = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition)
if m is None:
raise ValueError("Unexpected 'content-disposition' header value")
filename = m.group(1)
filename = os.path.basename(m.group(1)) # Strip any directory traversal
if filename in ("", ".", ".."): # fall back to tmp filename
filename = os.path.basename(path)
path = os.path.join(os.path.dirname(path), filename)

with open(path, "wb") as f:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ def __init__(
self.body = http_resp.data.decode("utf-8")
except Exception: # noqa: S110
pass
self.headers = http_resp.getheaders()
self.headers = http_resp.headers

@classmethod
def from_response(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
Do not edit the class manually.
""" # noqa: E501


# import models into model package
from stackit.observability.models.alert import Alert
from stackit.observability.models.alert_config_receivers_response import (
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -81,16 +81,16 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in inhibit_rules (list)
_items = []
if self.inhibit_rules:
for _item in self.inhibit_rules:
if _item:
_items.append(_item.to_dict())
for _item_inhibit_rules in self.inhibit_rules:
if _item_inhibit_rules:
_items.append(_item_inhibit_rules.to_dict())
_dict["inhibitRules"] = _items
# override the default output from pydantic by calling `to_dict()` of each item in receivers (list)
_items = []
if self.receivers:
for _item in self.receivers:
if _item:
_items.append(_item.to_dict())
for _item_receivers in self.receivers:
if _item_receivers:
_items.append(_item_receivers.to_dict())
_dict["receivers"] = _items
# override the default output from pydantic by calling `to_dict()` of route
if self.route:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in data (list)
_items = []
if self.data:
for _item in self.data:
if _item:
_items.append(_item.to_dict())
for _item_data in self.data:
if _item_data:
_items.append(_item_data.to_dict())
_dict["data"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in rules (list)
_items = []
if self.rules:
for _item in self.rules:
if _item:
_items.append(_item.to_dict())
for _item_rules in self.rules:
if _item_rules:
_items.append(_item_rules.to_dict())
_dict["rules"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in data (list)
_items = []
if self.data:
for _item in self.data:
if _item:
_items.append(_item.to_dict())
for _item_data in self.data:
if _item_data:
_items.append(_item_data.to_dict())
_dict["data"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in data (list)
_items = []
if self.data:
for _item in self.data:
if _item:
_items.append(_item.to_dict())
for _item_data in self.data:
if _item_data:
_items.append(_item_data.to_dict())
_dict["data"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in data (list)
_items = []
if self.data:
for _item in self.data:
if _item:
_items.append(_item.to_dict())
for _item_data in self.data:
if _item_data:
_items.append(_item_data.to_dict())
_dict["data"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,9 @@
import json
import pprint
from typing import Any, ClassVar, Dict, List, Optional, Set
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field, StrictStr
from pydantic import BaseModel, ConfigDict, Field
from typing_extensions import Annotated, Self


Expand All @@ -28,7 +29,7 @@ class BackupSchedule(BaseModel):
""" # noqa: E501

schedule: Annotated[str, Field(min_length=1, strict=True, max_length=100)]
schedule_id: StrictStr = Field(alias="scheduleId")
schedule_id: UUID = Field(alias="scheduleId")
__properties: ClassVar[List[str]] = ["schedule", "scheduleId"]

model_config = ConfigDict(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -88,30 +88,30 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in alert_config_backup_schedules (list)
_items = []
if self.alert_config_backup_schedules:
for _item in self.alert_config_backup_schedules:
if _item:
_items.append(_item.to_dict())
for _item_alert_config_backup_schedules in self.alert_config_backup_schedules:
if _item_alert_config_backup_schedules:
_items.append(_item_alert_config_backup_schedules.to_dict())
_dict["alertConfigBackupSchedules"] = _items
# override the default output from pydantic by calling `to_dict()` of each item in alert_rules_backup_schedules (list)
_items = []
if self.alert_rules_backup_schedules:
for _item in self.alert_rules_backup_schedules:
if _item:
_items.append(_item.to_dict())
for _item_alert_rules_backup_schedules in self.alert_rules_backup_schedules:
if _item_alert_rules_backup_schedules:
_items.append(_item_alert_rules_backup_schedules.to_dict())
_dict["alertRulesBackupSchedules"] = _items
# override the default output from pydantic by calling `to_dict()` of each item in grafana_backup_schedules (list)
_items = []
if self.grafana_backup_schedules:
for _item in self.grafana_backup_schedules:
if _item:
_items.append(_item.to_dict())
for _item_grafana_backup_schedules in self.grafana_backup_schedules:
if _item_grafana_backup_schedules:
_items.append(_item_grafana_backup_schedules.to_dict())
_dict["grafanaBackupSchedules"] = _items
# override the default output from pydantic by calling `to_dict()` of each item in scrape_config_backup_schedules (list)
_items = []
if self.scrape_config_backup_schedules:
for _item in self.scrape_config_backup_schedules:
if _item:
_items.append(_item.to_dict())
for _item_scrape_config_backup_schedules in self.scrape_config_backup_schedules:
if _item_scrape_config_backup_schedules:
_items.append(_item_scrape_config_backup_schedules.to_dict())
_dict["scrapeConfigBackupSchedules"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -79,9 +79,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in cert_checks (list)
_items = []
if self.cert_checks:
for _item in self.cert_checks:
if _item:
_items.append(_item.to_dict())
for _item_cert_checks in self.cert_checks:
if _item_cert_checks:
_items.append(_item_cert_checks.to_dict())
_dict["certChecks"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -91,23 +91,23 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in email_configs (list)
_items = []
if self.email_configs:
for _item in self.email_configs:
if _item:
_items.append(_item.to_dict())
for _item_email_configs in self.email_configs:
if _item_email_configs:
_items.append(_item_email_configs.to_dict())
_dict["emailConfigs"] = _items
# override the default output from pydantic by calling `to_dict()` of each item in opsgenie_configs (list)
_items = []
if self.opsgenie_configs:
for _item in self.opsgenie_configs:
if _item:
_items.append(_item.to_dict())
for _item_opsgenie_configs in self.opsgenie_configs:
if _item_opsgenie_configs:
_items.append(_item_opsgenie_configs.to_dict())
_dict["opsgenieConfigs"] = _items
# override the default output from pydantic by calling `to_dict()` of each item in web_hook_configs (list)
_items = []
if self.web_hook_configs:
for _item in self.web_hook_configs:
if _item:
_items.append(_item.to_dict())
for _item_web_hook_configs in self.web_hook_configs:
if _item_web_hook_configs:
_items.append(_item_web_hook_configs.to_dict())
_dict["webHookConfigs"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -128,9 +128,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in routes (list)
_items = []
if self.routes:
for _item in self.routes:
if _item:
_items.append(_item.to_dict())
for _item_routes in self.routes:
if _item_routes:
_items.append(_item_routes.to_dict())
_dict["routes"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -81,9 +81,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in rules (list)
_items = []
if self.rules:
for _item in self.rules:
if _item:
_items.append(_item.to_dict())
for _item_rules in self.rules:
if _item_rules:
_items.append(_item_rules.to_dict())
_dict["rules"] = _items
return _dict

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -81,9 +81,9 @@ def to_dict(self) -> Dict[str, Any]:
# override the default output from pydantic by calling `to_dict()` of each item in rules (list)
_items = []
if self.rules:
for _item in self.rules:
if _item:
_items.append(_item.to_dict())
for _item_rules in self.rules:
if _item_rules:
_items.append(_item_rules.to_dict())
_dict["rules"] = _items
return _dict

Expand Down
Loading
Loading