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
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,9 @@
from stackit.observability.models.alert_config_route_response import (
AlertConfigRouteResponse,
)
from stackit.observability.models.alert_group import AlertGroup
from stackit.observability.models.alert_groups_response import AlertGroupsResponse
from stackit.observability.models.alert_rule_record import AlertRuleRecord
from stackit.observability.models.basic_auth import BasicAuth
from stackit.observability.models.create_alert_config_receiver_payload import (
CreateAlertConfigReceiverPayload,
Expand Down
248 changes: 248 additions & 0 deletions services/observability/src/stackit/observability/api/default_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from stackit.observability.models.alert_config_route_response import (
AlertConfigRouteResponse,
)
from stackit.observability.models.alert_groups_response import AlertGroupsResponse
from stackit.observability.models.create_alert_config_receiver_payload import (
CreateAlertConfigReceiverPayload,
)
Expand Down Expand Up @@ -9453,3 +9454,250 @@ def _update_scrape_config_serialize(
_host=_host,
_request_auth=_request_auth,
)

@validate_call
def v1_projects_instances_alertgroups_list(
self,
instance_id: StrictStr,
project_id: StrictStr,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)]],
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> AlertGroupsResponse:
"""v1_projects_instances_alertgroups_list

Get alert groups.

:param instance_id: (required)
:type instance_id: str
:param project_id: (required)
:type project_id: str
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:type _request_timeout: int, tuple(int, int), optional
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501 docstring might be too long

_param = self._v1_projects_instances_alertgroups_list_serialize(
instance_id=instance_id,
project_id=project_id,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index,
)

_response_types_map: Dict[str, Optional[str]] = {
"200": "AlertGroupsResponse",
"400": "Error",
"403": "PermissionDenied",
}
response_data = self.api_client.call_api(*_param, _request_timeout=_request_timeout)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
).data

@validate_call
def v1_projects_instances_alertgroups_list_with_http_info(
self,
instance_id: StrictStr,
project_id: StrictStr,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)]],
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> ApiResponse[AlertGroupsResponse]:
"""v1_projects_instances_alertgroups_list

Get alert groups.

:param instance_id: (required)
:type instance_id: str
:param project_id: (required)
:type project_id: str
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:type _request_timeout: int, tuple(int, int), optional
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501 docstring might be too long

_param = self._v1_projects_instances_alertgroups_list_serialize(
instance_id=instance_id,
project_id=project_id,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index,
)

_response_types_map: Dict[str, Optional[str]] = {
"200": "AlertGroupsResponse",
"400": "Error",
"403": "PermissionDenied",
}
response_data = self.api_client.call_api(*_param, _request_timeout=_request_timeout)
response_data.read()
return self.api_client.response_deserialize(
response_data=response_data,
response_types_map=_response_types_map,
)

@validate_call
def v1_projects_instances_alertgroups_list_without_preload_content(
self,
instance_id: StrictStr,
project_id: StrictStr,
_request_timeout: Union[
None,
Annotated[StrictFloat, Field(gt=0)],
Tuple[Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)]],
] = None,
_request_auth: Optional[Dict[StrictStr, Any]] = None,
_content_type: Optional[StrictStr] = None,
_headers: Optional[Dict[StrictStr, Any]] = None,
_host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
) -> RESTResponseType:
"""v1_projects_instances_alertgroups_list

Get alert groups.

:param instance_id: (required)
:type instance_id: str
:param project_id: (required)
:type project_id: str
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
:type _request_timeout: int, tuple(int, int), optional
:param _request_auth: set to override the auth_settings for an a single
request; this effectively ignores the
authentication in the spec for a single request.
:type _request_auth: dict, optional
:param _content_type: force content-type for the request.
:type _content_type: str, Optional
:param _headers: set to override the headers for a single
request; this effectively ignores the headers
in the spec for a single request.
:type _headers: dict, optional
:param _host_index: set to override the host_index for a single
request; this effectively ignores the host_index
in the spec for a single request.
:type _host_index: int, optional
:return: Returns the result object.
""" # noqa: E501 docstring might be too long

_param = self._v1_projects_instances_alertgroups_list_serialize(
instance_id=instance_id,
project_id=project_id,
_request_auth=_request_auth,
_content_type=_content_type,
_headers=_headers,
_host_index=_host_index,
)

_response_types_map: Dict[str, Optional[str]] = {
"200": "AlertGroupsResponse",
"400": "Error",
"403": "PermissionDenied",
}
response_data = self.api_client.call_api(*_param, _request_timeout=_request_timeout)
return response_data.response

def _v1_projects_instances_alertgroups_list_serialize(
self,
instance_id,
project_id,
_request_auth,
_content_type,
_headers,
_host_index,
) -> RequestSerialized:

_host = None

_collection_formats: Dict[str, str] = {}

_path_params: Dict[str, str] = {}
_query_params: List[Tuple[str, str]] = []
_header_params: Dict[str, Optional[str]] = _headers or {}
_form_params: List[Tuple[str, str]] = []
_files: Dict[str, Union[str, bytes]] = {}
_body_params: Optional[bytes] = None

# process the path parameters
if instance_id is not None:
_path_params["instanceId"] = instance_id
if project_id is not None:
_path_params["projectId"] = project_id
# process the query parameters
# process the header parameters
# process the form parameters
# process the body parameter

# set the HTTP header `Accept`
if "Accept" not in _header_params:
_header_params["Accept"] = self.api_client.select_header_accept(["application/json"])

# authentication setting
_auth_settings: List[str] = []

return self.api_client.param_serialize(
method="GET",
resource_path="/v1/projects/{projectId}/instances/{instanceId}/alertgroups",
path_params=_path_params,
query_params=_query_params,
header_params=_header_params,
body=_body_params,
post_params=_form_params,
files=_files,
auth_settings=_auth_settings,
collection_formats=_collection_formats,
_host=_host,
_request_auth=_request_auth,
)
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
from stackit.observability.models.alert_config_route_response import (
AlertConfigRouteResponse,
)
from stackit.observability.models.alert_group import AlertGroup
from stackit.observability.models.alert_groups_response import AlertGroupsResponse
from stackit.observability.models.alert_rule_record import AlertRuleRecord
from stackit.observability.models.basic_auth import BasicAuth
from stackit.observability.models.create_alert_config_receiver_payload import (
CreateAlertConfigReceiverPayload,
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
# coding: utf-8

"""
STACKIT Observability API

API endpoints for Observability on STACKIT

The version of the OpenAPI document: 1.1.1
Contact: [email protected]
Generated by OpenAPI Generator (https://openapi-generator.tech)

Do not edit the class manually.
""" # noqa: E501 docstring might be too long

from __future__ import annotations

import json
import pprint
from typing import Any, ClassVar, Dict, List, Optional, Set

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

from stackit.observability.models.alert_rule_record import AlertRuleRecord


class AlertGroup(BaseModel):
"""
AlertGroup
"""

interval: Optional[Annotated[str, Field(min_length=2, strict=True, max_length=8)]] = "60s"
name: Annotated[str, Field(min_length=1, strict=True, max_length=200)]
rules: List[AlertRuleRecord]
__properties: ClassVar[List[str]] = ["interval", "name", "rules"]

model_config = ConfigDict(
populate_by_name=True,
validate_assignment=True,
protected_namespaces=(),
)

def to_str(self) -> str:
"""Returns the string representation of the model using alias"""
return pprint.pformat(self.model_dump(by_alias=True))

def to_json(self) -> str:
"""Returns the JSON representation of the model using alias"""
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
return json.dumps(self.to_dict())

@classmethod
def from_json(cls, json_str: str) -> Optional[Self]:
"""Create an instance of AlertGroup from a JSON string"""
return cls.from_dict(json.loads(json_str))

def to_dict(self) -> Dict[str, Any]:
"""Return the dictionary representation of the model using alias.

This has the following differences from calling pydantic's
`self.model_dump(by_alias=True)`:

* `None` is only added to the output dict for nullable fields that
were set at model initialization. Other fields with value `None`
are ignored.
"""
excluded_fields: Set[str] = set([])

_dict = self.model_dump(
by_alias=True,
exclude=excluded_fields,
exclude_none=True,
)
# 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())
_dict["rules"] = _items
return _dict

@classmethod
def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
"""Create an instance of AlertGroup from a dict"""
if obj is None:
return None

if not isinstance(obj, dict):
return cls.model_validate(obj)

_obj = cls.model_validate(
{
"interval": obj.get("interval") if obj.get("interval") is not None else "60s",
"name": obj.get("name"),
"rules": (
[AlertRuleRecord.from_dict(_item) for _item in obj["rules"]]
if obj.get("rules") is not None
else None
),
}
)
return _obj
Loading
Loading