import logging
from dataclasses import asdict, dataclass, field
from typing import List, Optional, Union

from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook


@dataclass
class Dimension:
    Name: str
    Value: str


@dataclass
class Metric:
    MetricName: str
    Value: Union[int, float, bool]
    Dimensions: List[Dimension] = field(default_factory=list)
    Unit: str = "None"


def _metric_to_dict(metric: Metric):
    metric_dict = asdict(metric)
    metric_dict["Value"] = float(metric_dict["Value"])
    return metric_dict


class MetricsHook(AwsBaseHook):
    def __init__(self, default_namespace: Optional[str] = None, **kwargs):
        super().__init__(client_type="cloudwatch", aws_conn_id=None, **kwargs)
        self.default_namespace = default_namespace

    def put_metrics(self, payloads: List[Metric], namespace: Optional[str] = None):
        logging.info(
            f"Sending payload: {payloads} to cloudwatch with namespace: {namespace}."
        )
        if namespace is None:
            if self.default_namespace is None:
                raise ValueError("Both namespace and default_namespace are None")
            namespace = self.default_namespace
        dict_payloads = [_metric_to_dict(metric) for metric in payloads]
        resp = self.conn.put_metric_data(MetricData=dict_payloads, Namespace=namespace)
        if resp["ResponseMetadata"]["HTTPStatusCode"] != 200:
            raise RuntimeError("Error putting metric: \n%s" % resp)
        return resp
