Add issues API

Author: dmelikyan

graphsignal/client/models/issue.py

@@ -0,0 +1,107 @@
+# coding: utf-8
+
+"""
+    Graphsignal API
+
+    API for uploading and querying spans, issues, metrics, and logs.
+
+    The version of the OpenAPI document: 1.0.0
+    Generated by OpenAPI Generator (https://openapi-generator.tech)
+
+    Do not edit the class manually.
+"""  # noqa: E501
+
+
+from __future__ import annotations
+import pprint
+import re  # noqa: F401
+import json
+
+from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr
+from typing import Any, ClassVar, Dict, List, Optional
+from graphsignal.client.models.tag import Tag
+from typing import Optional, Set
+from typing_extensions import Self
+
+class Issue(BaseModel):
+    """
+    Issue
+    """ # noqa: E501
+    issue_id: Optional[StrictStr] = Field(default=None, description="Unique identifier for the issue.")
+    span_id: Optional[StrictStr] = Field(default=None, description="The associated span identifier, if the issue is being associated with a span.")
+    tags: Optional[List[Tag]] = Field(default=None, description="Tags associated with the issue.")
+    name: StrictStr = Field(description="The name of the issue.")
+    severity: Optional[StrictInt] = Field(default=None, description="Severity of the issue, 1-5 (info, low, medium, high, critical).")
+    description: Optional[StrictStr] = Field(default=None, description="The description of the issue.")
+    create_ts: StrictInt = Field(description="Unix timestamp (seconds) when the issue was created.")
+    __properties: ClassVar[List[str]] = ["issue_id", "span_id", "tags", "name", "severity", "description", "create_ts"]
+
+    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 Issue 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 tags (list)
+        _items = []
+        if self.tags:
+            for _item_tags in self.tags:
+                if _item_tags:
+                    _items.append(_item_tags.to_dict())
+            _dict['tags'] = _items
+        return _dict
+
+    @classmethod
+    def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
+        """Create an instance of Issue from a dict"""
+        if obj is None:
+            return None
+
+        if not isinstance(obj, dict):
+            return cls.model_validate(obj)
+
+        _obj = cls.model_validate({
+            "issue_id": obj.get("issue_id"),
+            "span_id": obj.get("span_id"),
+            "tags": [Tag.from_dict(_item) for _item in obj["tags"]] if obj.get("tags") is not None else None,
+            "name": obj.get("name"),
+            "severity": obj.get("severity"),
+            "description": obj.get("description"),
+            "create_ts": obj.get("create_ts")
+        })
+        return _obj
+
+

graphsignal/client/models/issue_query_result.py

@@ -0,0 +1,95 @@
+# coding: utf-8
+
+"""
+    Graphsignal API
+
+    API for uploading and querying spans, issues, metrics, and logs.
+
+    The version of the OpenAPI document: 1.0.0
+    Generated by OpenAPI Generator (https://openapi-generator.tech)
+
+    Do not edit the class manually.
+"""  # noqa: E501
+
+
+from __future__ import annotations
+import pprint
+import re  # noqa: F401
+import json
+
+from pydantic import BaseModel, ConfigDict, Field
+from typing import Any, ClassVar, Dict, List, Optional
+from graphsignal.client.models.issue import Issue
+from typing import Optional, Set
+from typing_extensions import Self
+
+class IssueQueryResult(BaseModel):
+    """
+    IssueQueryResult
+    """ # noqa: E501
+    data: Optional[List[Issue]] = Field(default=None, description="List of issues resulting from the query.")
+    __properties: ClassVar[List[str]] = ["data"]
+
+    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 IssueQueryResult 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 data (list)
+        _items = []
+        if self.data:
+            for _item_data in self.data:
+                if _item_data:
+                    _items.append(_item_data.to_dict())
+            _dict['data'] = _items
+        return _dict
+
+    @classmethod
+    def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
+        """Create an instance of IssueQueryResult from a dict"""
+        if obj is None:
+            return None
+
+        if not isinstance(obj, dict):
+            return cls.model_validate(obj)
+
+        _obj = cls.model_validate({
+            "data": [Issue.from_dict(_item) for _item in obj["data"]] if obj.get("data") is not None else None
+        })
+        return _obj
+
+

graphsignal/env_vars.py

@@ -0,0 +1,50 @@
+from typing import Dict, Any, Union, Optional, Type
+import os
+
+
+def parse_env_param(name: str, value: Any, expected_type: Type) -> Any:
+    if value is None:
+        return None
+
+    try:
+        if expected_type == bool:
+            return value if isinstance(value, bool) else str(value).lower() in ("true", "1", "yes")
+        elif expected_type == int:
+            return int(value)
+        elif expected_type == float:
+            return float(value)
+        elif expected_type == str:
+            return str(value)
+        elif expected_type == list:
+            return [item.strip() for item in str(value).split(',') if item.strip()]
+    except (ValueError, TypeError):
+        pass
+
+    raise ValueError(f"Invalid type for {name}: expected {expected_type.__name__}, got {type(value).__name__}")
+
+
+def read_config_param(name: str, expected_type: Type, provided_value: Optional[Any] = None, default_value: Optional[Any] = None, required: bool = False) -> Any:
+    # Check if the value was provided as an argument
+    if provided_value is not None:
+        return provided_value
+
+    # Check if the value was provided as an environment variable
+    env_value = os.getenv(f'GRAPHSIGNAL_{name.upper()}')
+    if env_value is not None:
+        parsed_env_value = parse_env_param(name, env_value, expected_type)
+        if parsed_env_value is not None:
+            return parsed_env_value
+
+    if required:
+        raise ValueError(f"Missing required argument: {name}")
+
+    return default_value
+
+
+def read_config_tags(provided_value: Optional[dict] = None, prefix: str = "GRAPHSIGNAL_TAG_") -> Dict[str, str]:
+    # Check if the value was provided as an argument
+    if provided_value is not None:
+        return provided_value
+
+    # Check if the value was provided as an environment variable
+    return {key[len(prefix):].lower(): value for key, value in os.environ.items() if key.startswith(prefix)}

graphsignal/profiles.py

@@ -0,0 +1,45 @@
+from typing import Union, Any, Optional, Dict
+import logging
+import json
+
+logger = logging.getLogger('graphsignal')
+
+
+class EventAverages():
+    __slots__ = [
+        'events'
+    ]
+
+    def __init__(self):
+        self.events = {}
+
+    def is_empty(self) -> bool:
+        return len(self.events) == 0
+
+    def inc_counters(self, event_name, stats):
+        if event_name not in self.events:
+            event_counters = self.events[event_name] = dict()
+        else:
+            event_counters = self.events[event_name]
+
+        for key, value in stats.items():
+            if key not in event_counters:
+                event_counters[key] = value
+            else:
+                event_counters[key] += value
+
+        if 'count' not in stats:
+            if 'count' not in event_counters:
+                event_counters['count'] = 1
+            else:
+                event_counters['count'] += 1
+
+    def dumps(self, max_events: int = 250, limit_by: str = 'count') -> str:
+        # sort by limit_by key of event counters in descending order and return first max_events
+        sorted_events = sorted(self.events.items(), key=lambda x: x[1].get(limit_by, 0), reverse=True)
+        limited_events = dict(sorted_events[:max_events])
+        
+        return json.dumps(limited_events)
+
+    def clear(self):
+        self.events.clear()