import json
from typing import Any, Dict

from airflow.models.connection import Connection
from twitch_airflow_components.hooks.postgres import PostgresHook

from conductor.hooks.hook_iface import IConfiguredHook


class ConfiguredRedshiftHook(IConfiguredHook):
    def generate_hook(self, *_args: Any, **kwargs: Any) -> PostgresHook:
        config = self.project_resources.env.redshift
        if config is None:
            raise ValueError(
                "Cannot use RedshiftOperator when no Redshift config is set."
            )
        extra: Dict[str, Any] = {"iam": True, "aws_conn_id": None, "redshift": True}
        if config.cluster_identifier is not None:
            extra["cluster-identifier"] = config.cluster_identifier
        connection = Connection(
            host=config.host,
            login=config.db_user,
            schema=config.db_name,
            extra=json.dumps(extra),
        )
        if config.port is not None:
            connection.port = int(config.port)
        return PostgresHook(connection=connection, schema=config.db_name, **kwargs)
