# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
from typing import Any, TYPE_CHECKING, TypeVar, Generic, Union, Optional

from .base import BaseModel
from .parameter_list import ClientGlobalParameterList, ConfigGlobalParameterList
from .imports import FileImport, ImportType, TypingSection, MsrestImportType
from .utils import add_to_pylint_disable
from .operation_group import OperationGroup
from .request_builder import (
    RequestBuilder,
    OverloadedRequestBuilder,
    get_request_builder,
)
from .parameter import Parameter, ParameterMethodLocation, ParameterLocation
from .lro_operation import LROOperation
from .lro_paging_operation import LROPagingOperation
from ...utils import extract_original_name, NAME_LENGTH_LIMIT

ParameterListType = TypeVar(
    "ParameterListType",
    bound=Union[ClientGlobalParameterList, ConfigGlobalParameterList],
)

if TYPE_CHECKING:
    from .code_model import CodeModel
    from . import OperationType


class _ClientConfigBase(Generic[ParameterListType], BaseModel):
    """The service client base. Shared across our Client and Config type"""

    def __init__(
        self,
        yaml_data: dict[str, Any],
        code_model: "CodeModel",
        parameters: ParameterListType,
    ):
        super().__init__(yaml_data, code_model)
        self.parameters = parameters
        self.url: str = self.yaml_data["url"]  # the base endpoint of the client. Can be parameterized or not
        self.legacy_filename: str = self.yaml_data.get("legacyFilename", "client")
        self.client_namespace: str = self.yaml_data.get("clientNamespace", code_model.namespace)

    @property
    def description(self) -> str:
        return self.yaml_data["description"]

    @property
    def name(self) -> str:
        return self.yaml_data["name"]


class Client(_ClientConfigBase[ClientGlobalParameterList]):
    """Model representing our service client"""

    def __init__(
        self,
        yaml_data: dict[str, Any],
        code_model: "CodeModel",
        parameters: ClientGlobalParameterList,
        *,
        is_subclient: bool = False,
    ):
        super().__init__(yaml_data, code_model, parameters)
        self.operation_groups: list[OperationGroup] = []
        self.config = Config.from_yaml(yaml_data, self.code_model)
        self.is_subclient = is_subclient
        self.request_builders = self._build_request_builders()
        if self.code_model.options["show-operations"]:
            self.operation_groups = [
                OperationGroup.from_yaml(op_group, code_model, self)
                for op_group in self.yaml_data.get("operationGroups", [])
            ]
            self.link_lro_initial_operations()
        self.request_id_header_name = self.yaml_data.get("requestIdHeaderName", None)
        self.has_etag: bool = yaml_data.get("hasEtag", False)

        # update the host parameter value. In later logic, SDK will overwrite it
        # with value from cloud_setting if users don't provide it.
        if self.code_model.options.get("azure-arm"):
            if self.need_cloud_setting:
                for p in self.parameters.parameters:
                    if p.location == ParameterLocation.ENDPOINT_PATH:
                        p.client_default_value = None
                        p.optional = True
                        break
            else:
                for idx, p in enumerate(self.parameters.parameters):
                    if p.client_name == "cloud_setting":
                        self.parameters.parameters.pop(idx)
                        break

    @property
    def need_cloud_setting(self) -> bool:
        return bool(
            self.code_model.options.get("azure-arm", False)
            and self.credential_scopes is not None
            and self.endpoint_parameter is not None
        )

    @property
    def endpoint_parameter(self) -> Optional[Parameter]:
        return next((p for p in self.parameters.parameters if p.location == ParameterLocation.ENDPOINT_PATH), None)

    def _build_request_builders(
        self,
    ) -> list[Union[RequestBuilder, OverloadedRequestBuilder]]:
        request_builders: list[Union[RequestBuilder, OverloadedRequestBuilder]] = []

        def add_og_request_builder(og: dict[str, Any]):
            for operation_yaml in og["operations"]:
                request_builder = get_request_builder(
                    operation_yaml,
                    code_model=self.code_model,
                    client=self,
                )
                if operation_yaml.get("isLroInitialOperation"):
                    # we want to change the name
                    request_builder.name = request_builder.get_name(
                        extract_original_name(request_builder.yaml_data["name"]),
                        request_builder.yaml_data,
                        request_builder.code_model,
                        request_builder.client,
                    )
                if request_builder.overloads:
                    request_builders.extend(request_builder.overloads)
                request_builders.append(request_builder)
                if operation_yaml.get("nextOperation"):
                    # i am a paging operation and i have a next operation.
                    # Make sure to include my next operation
                    request_builders.append(
                        get_request_builder(
                            operation_yaml["nextOperation"],
                            code_model=self.code_model,
                            client=self,
                        )
                    )

        queue = self.yaml_data.get("operationGroups", []).copy()
        while queue:
            now = queue.pop(0)
            add_og_request_builder(now)
            if now.get("operationGroups"):
                queue.extend(now["operationGroups"])

        return request_builders

    def pipeline_class(self, async_mode: bool) -> str:
        if self.code_model.options["azure-arm"]:
            if async_mode:
                return "AsyncARMPipelineClient"
            return "ARMPipelineClient"
        if async_mode:
            return "AsyncPipelineClient"
        return "PipelineClient"

    @property
    def credential(self) -> Optional[Parameter]:
        """The credential param, if one exists"""
        return self.parameters.credential

    @property
    def send_request_name(self) -> str:
        """Name of the send request function"""
        return "send_request" if self.code_model.options["show-send-request"] else "_send_request"

    @property
    def has_parameterized_host(self) -> bool:
        """Whether the base url is parameterized or not"""
        return not any(p for p in self.parameters if p.is_host)

    def pylint_disable(self) -> str:
        retval = ""
        if not any(
            p
            for p in self.parameters.parameters
            if p.is_api_version
            and p.client_name == "api_version"
            and p.method_location in [ParameterMethodLocation.KEYWORD_ONLY, ParameterMethodLocation.KWARG]
        ):
            retval = add_to_pylint_disable(retval, "client-accepts-api-version-keyword")
        if len(self.operation_groups) > 6:
            retval = add_to_pylint_disable(retval, "too-many-instance-attributes")
        if len(self.name) > NAME_LENGTH_LIMIT:
            retval = add_to_pylint_disable(retval, "name-too-long")
        return retval

    @property
    def filename(self) -> str:
        """Name of the file for the client"""
        if self.code_model.options["version-tolerant"] or self.code_model.options["low-level-client"]:
            return "_client"
        return f"_{self.legacy_filename}"

    def lookup_request_builder(self, request_builder_id: int) -> Union[RequestBuilder, OverloadedRequestBuilder]:
        """Find the request builder based off of id"""
        try:
            return next(rb for rb in self.request_builders if id(rb.yaml_data) == request_builder_id)
        except StopIteration as exc:
            raise KeyError(f"No request builder with id {request_builder_id} found.") from exc

    def lookup_operation(self, operation_id: int) -> "OperationType":
        try:
            return next(o for og in self.operation_groups for o in og.operations if id(o.yaml_data) == operation_id)
        except StopIteration as exc:
            raise KeyError(f"No operation with id {operation_id} found.") from exc

    @property
    def has_mixin(self) -> bool:
        """Do we want a mixin ABC class for typing purposes?"""
        return any(og for og in self.operation_groups if og.is_mixin)

    @property
    def lro_operations(self) -> list["OperationType"]:
        """all LRO operations in this SDK?"""
        return [operation for operation_group in self.operation_groups for operation in operation_group.lro_operations]

    @property
    def has_public_lro_operations(self) -> bool:
        """Are there any public LRO operations in this SDK?"""
        return any(not operation.internal for operation in self.lro_operations)

    @property
    def has_operations(self) -> bool:
        return any(operation_group.has_operations for operation_group in self.operation_groups)

    def link_lro_initial_operations(self) -> None:
        """Link each LRO operation to its initial operation"""
        for operation_group in self.operation_groups:
            for operation in operation_group.operations:
                if isinstance(operation, (LROOperation, LROPagingOperation)):
                    operation.initial_operation = self.lookup_operation(id(operation.yaml_data["initialOperation"]))

    @property
    def has_abstract_operations(self) -> bool:
        """Whether there is abstract operation in any operation group."""
        return any(og.has_abstract_operations for og in self.operation_groups)

    @property
    def has_non_abstract_operations(self) -> bool:
        """Whether there is non-abstract operation in any operation group."""
        return any(og.has_non_abstract_operations for og in self.operation_groups)

    def imports(self, async_mode: bool, **kwargs) -> FileImport:
        file_import = FileImport(self.code_model)
        file_import.add_submodule_import("typing", "Any", ImportType.STDLIB, TypingSection.REGULAR)
        if self.code_model.options["azure-arm"]:
            file_import.add_submodule_import("azure.mgmt.core", self.pipeline_class(async_mode), ImportType.SDKCORE)
        else:
            file_import.add_submodule_import(
                "" if self.code_model.is_azure_flavor else "runtime",
                self.pipeline_class(async_mode),
                ImportType.SDKCORE,
            )

        for gp in self.parameters:
            if gp.method_location == ParameterMethodLocation.KWARG:
                continue
            file_import.merge(
                gp.imports(
                    async_mode,
                    is_operation_file=True,
                    **kwargs,
                )
            )
        file_import.add_submodule_import(
            "._configuration",
            f"{self.name}Configuration",
            ImportType.LOCAL,
        )
        serialize_namespace = kwargs.get("serialize_namespace", self.code_model.namespace)
        file_import.add_msrest_import(
            serialize_namespace=serialize_namespace,
            msrest_import_type=MsrestImportType.SerializerDeserializer,
            typing_section=TypingSection.REGULAR,
        )
        file_import.add_submodule_import(
            "pipeline" if self.code_model.is_azure_flavor else "runtime",
            "policies",
            ImportType.SDKCORE,
        )
        if self.code_model.options["azure-arm"]:
            async_prefix = "Async" if async_mode else ""
            file_import.add_submodule_import(
                "azure.mgmt.core.policies",
                f"{async_prefix}ARMAutoResourceProviderRegistrationPolicy",
                ImportType.SDKCORE,
            )

        # import for "Self" (typing.Self available from Python 3.11+)
        file_import.add_submodule_import(
            "typing_extensions",
            "Self",
            ImportType.STDLIB,
            version_modules=(((3, 11), "typing", None),),
        )
        if self.need_cloud_setting:
            file_import.add_submodule_import("typing", "cast", ImportType.STDLIB)
            file_import.add_submodule_import("azure.core.settings", "settings", ImportType.SDKCORE)
            file_import.add_submodule_import("azure.mgmt.core.tools", "get_arm_endpoints", ImportType.SDKCORE)
        if async_mode:
            file_import.add_submodule_import("typing", "Awaitable", ImportType.STDLIB)
            file_import.add_submodule_import(
                "rest",
                "AsyncHttpResponse",
                ImportType.SDKCORE,
                TypingSection.REGULAR,
            )
        else:
            file_import.add_submodule_import(
                "rest",
                "HttpResponse",
                ImportType.SDKCORE,
                TypingSection.REGULAR,
            )
        file_import.add_submodule_import(
            "rest",
            "HttpRequest",
            ImportType.SDKCORE,
            TypingSection.REGULAR,
        )
        serialize_namespace = kwargs.get("serialize_namespace", self.code_model.namespace)
        for og in self.operation_groups:
            file_import.add_submodule_import(
                self.code_model.get_relative_import_path(
                    serialize_namespace,
                    self.code_model.get_imported_namespace_for_operation(og.client_namespace, async_mode),
                ),
                og.class_name,
                ImportType.LOCAL,
            )

        if self.code_model.model_types and self.code_model.options["models-mode"] == "msrest":
            path_to_models = ".." if async_mode else "."
            file_import.add_submodule_import(path_to_models, "models", ImportType.LOCAL, alias="_models")
        file_import.add_submodule_import("copy", "deepcopy", ImportType.STDLIB)
        return file_import

    @property
    def credential_scopes(self) -> Optional[list[str]]:
        """Credential scopes for this client"""

        if self.credential:
            if hasattr(getattr(self.credential.type, "policy", None), "credential_scopes"):
                return self.credential.type.policy.credential_scopes  # type: ignore
            for t in getattr(self.credential.type, "types", []):
                if hasattr(getattr(t, "policy", None), "credential_scopes"):
                    return t.policy.credential_scopes
        return None

    @classmethod
    def from_yaml(
        cls,
        yaml_data: dict[str, Any],
        code_model: "CodeModel",
        *,
        is_subclient: bool = False,
    ) -> "Client":
        return cls(
            yaml_data=yaml_data,
            code_model=code_model,
            parameters=ClientGlobalParameterList.from_yaml(yaml_data, code_model),
            is_subclient=is_subclient,
        )


class Config(_ClientConfigBase[ConfigGlobalParameterList]):
    """Model representing our Config type."""

    def pylint_disable(self) -> str:
        retval = add_to_pylint_disable("", "too-many-instance-attributes") if self.code_model.is_azure_flavor else ""
        if len(self.name) > NAME_LENGTH_LIMIT:
            retval = add_to_pylint_disable(retval, "name-too-long")
        return retval

    @property
    def description(self) -> str:
        return (
            f"Configuration for {self.yaml_data['name']}.\n\n."
            "Note that all parameters used to create this instance are saved as instance attributes."
        )

    @property
    def sdk_moniker(self) -> str:
        package_name = self.code_model.options.get("package-name")
        if package_name and package_name.startswith("azure-"):
            package_name = package_name[len("azure-") :]
        return package_name if package_name else self.yaml_data["name"].lower()

    @property
    def name(self) -> str:
        return f"{super().name}Configuration"

    def imports(self, async_mode: bool, **kwargs) -> FileImport:
        file_import = FileImport(self.code_model)
        file_import.add_submodule_import(
            "pipeline" if self.code_model.is_azure_flavor else "runtime",
            "policies",
            ImportType.SDKCORE,
        )
        file_import.add_submodule_import("typing", "Any", ImportType.STDLIB, TypingSection.REGULAR)
        if self.code_model.options.get("package-version"):
            serialize_namespace = kwargs.get("serialize_namespace", self.code_model.namespace)
            file_import.add_submodule_import(
                self.code_model.get_relative_import_path(serialize_namespace, module_name="_version"),
                "VERSION",
                ImportType.LOCAL,
            )
        if self.code_model.options["azure-arm"]:
            policy = "AsyncARMChallengeAuthenticationPolicy" if async_mode else "ARMChallengeAuthenticationPolicy"
            file_import.add_submodule_import("azure.mgmt.core.policies", "ARMHttpLoggingPolicy", ImportType.SDKCORE)
            file_import.add_submodule_import("azure.mgmt.core.policies", policy, ImportType.SDKCORE)
        for gp in self.parameters:
            if gp.method_location == ParameterMethodLocation.KWARG and gp not in self.parameters.kwargs_to_pop:
                continue
            file_import.merge(
                gp.imports(
                    async_mode=async_mode,
                    **kwargs,
                )
            )
        return file_import

    @classmethod
    def from_yaml(cls, yaml_data: dict[str, Any], code_model: "CodeModel") -> "Config":
        return cls(
            yaml_data=yaml_data,
            code_model=code_model,
            parameters=ConfigGlobalParameterList.from_yaml(yaml_data, code_model),
        )
