settings.py

import enum
import os
import re
from dataclasses import asdict, dataclass, field
from keyword import iskeyword
from pathlib import Path
from textwrap import dedent

from .client_generators.constants import (
    DEFAULT_ASYNC_BASE_CLIENT_NAME,
    DEFAULT_ASYNC_BASE_CLIENT_OPEN_TELEMETRY_NAME,
    DEFAULT_ASYNC_BASE_CLIENT_OPEN_TELEMETRY_PATH,
    DEFAULT_ASYNC_BASE_CLIENT_PATH,
    DEFAULT_BASE_CLIENT_NAME,
    DEFAULT_BASE_CLIENT_OPEN_TELEMETRY_NAME,
    DEFAULT_BASE_CLIENT_OPEN_TELEMETRY_PATH,
    DEFAULT_BASE_CLIENT_PATH,
)
from .client_generators.scalars import ScalarData
from .exceptions import InvalidConfiguration


class CommentsStrategy(str, enum.Enum):
    NONE = "none"
    STABLE = "stable"
    TIMESTAMP = "timestamp"


class Strategy(str, enum.Enum):
    CLIENT = "client"
    GRAPHQL_SCHEMA = "graphqlschema"


@dataclass
class IntrospectionSettings:
    """
    Introspection settings for schema generation.
    """

    descriptions: bool = False
    input_value_deprecation: bool = False
    specified_by_url: bool = False
    schema_description: bool = False
    directive_is_repeatable: bool = False
    # graphql-core will rename this to one_of in a future version (update when bumping)
    input_object_one_of: bool = False


@dataclass
class BaseSettings:
    schema_path: str = ""
    remote_schema_url: str = ""
    remote_schema_headers: dict = field(default_factory=dict)
    remote_schema_verify_ssl: bool = True
    remote_schema_timeout: float = 5
    enable_custom_operations: bool = False
    plugins: list[str] = field(default_factory=list)
    introspection_descriptions: bool = False
    introspection_input_value_deprecation: bool = False
    introspection_specified_by_url: bool = False
    introspection_schema_description: bool = False
    introspection_directive_is_repeatable: bool = False
    introspection_input_object_one_of: bool = False

    def __post_init__(self):
        if not self.schema_path and not self.remote_schema_url:
            raise InvalidConfiguration(
                "Schema source not provided. Use schema_path or remote_schema_url"
            )

        if self.schema_path:
            assert_path_exists(self.schema_path)

        self.remote_schema_headers = resolve_headers(self.remote_schema_headers)
        if self.remote_schema_url:
            self.remote_schema_url = resolve_env_vars_in_string(self.remote_schema_url)

    @property
    def using_remote_schema(self) -> bool:
        """
        Return true if remote schema is used as source, false otherwise.
        """
        return bool(self.remote_schema_url) and not bool(self.schema_path)

    @property
    def introspection_settings(self) -> IntrospectionSettings:
        """
        Return ``IntrospectionSettings`` instance build from provided configuration.
        """
        return IntrospectionSettings(
            descriptions=self.introspection_descriptions,
            input_value_deprecation=self.introspection_input_value_deprecation,
            specified_by_url=self.introspection_specified_by_url,
            schema_description=self.introspection_schema_description,
            directive_is_repeatable=self.introspection_directive_is_repeatable,
            input_object_one_of=self.introspection_input_object_one_of,
        )

    def _introspection_settings_message(self) -> str:
        """
        Return human readable message with introspection settings values.
        """
        formatted = ", ".join(
            f"{key}={str(value).lower()}"
            for key, value in asdict(self.introspection_settings).items()
        )
        return f"Introspection settings: {formatted}"


@dataclass
class ClientSettings(BaseSettings):
    queries_path: str = ""
    target_package_name: str = "graphql_client"
    target_package_path: str = field(default_factory=lambda: Path.cwd().as_posix())
    client_name: str = "Client"
    client_file_name: str = "client"
    base_client_name: str = ""
    base_client_file_path: str = ""
    enums_module_name: str = "enums"
    input_types_module_name: str = "input_types"
    fragments_module_name: str = "fragments"
    include_comments: CommentsStrategy = field(default=CommentsStrategy.STABLE)
    convert_to_snake_case: bool = True
    include_all_inputs: bool = True
    include_all_enums: bool = True
    async_client: bool = True
    opentelemetry_client: bool = False
    files_to_include: list[str] = field(default_factory=list)
    scalars: dict[str, ScalarData] = field(default_factory=dict)
    default_optional_fields_to_none: bool = False
    include_typename: bool = True
    ignore_extra_fields: bool = True

    def __post_init__(self):
        if not self.queries_path and not self.enable_custom_operations:
            raise TypeError("__init__ missing 1 required argument: 'queries_path'")
        super().__post_init__()

        try:
            self.include_comments = CommentsStrategy(self.include_comments)
        except ValueError as exc:
            valid_options = ", ".join(strategy.value for strategy in CommentsStrategy)
            raise InvalidConfiguration(
                f"'{self.include_comments}' is not a valid choice. "
                f"Valid options are: {valid_options}"
            ) from exc

        self._set_default_base_client_data()

        for name, data in self.scalars.items():
            data.graphql_name = name

        assert_path_exists(self.queries_path)

        assert_string_is_valid_python_identifier(self.target_package_name)
        assert_path_is_valid_directory(self.target_package_path)

        assert_string_is_valid_python_identifier(self.client_name)
        assert_string_is_valid_python_identifier(self.client_file_name)
        assert_string_is_valid_python_identifier(self.base_client_name)
        assert_path_exists(self.base_client_file_path)
        assert_path_is_valid_file(self.base_client_file_path)
        assert_class_is_defined_in_file(
            Path(self.base_client_file_path), self.base_client_name
        )

        assert_string_is_valid_python_identifier(self.enums_module_name)
        assert_string_is_valid_python_identifier(self.input_types_module_name)

        for file_path in self.files_to_include:
            assert_path_is_valid_file(file_path)

    def _set_default_base_client_data(self):
        default_clients_map = {
            (True, True): (
                DEFAULT_ASYNC_BASE_CLIENT_OPEN_TELEMETRY_PATH,
                DEFAULT_ASYNC_BASE_CLIENT_OPEN_TELEMETRY_NAME,
            ),
            (True, False): (
                DEFAULT_ASYNC_BASE_CLIENT_PATH,
                DEFAULT_ASYNC_BASE_CLIENT_NAME,
            ),
            (False, True): (
                DEFAULT_BASE_CLIENT_OPEN_TELEMETRY_PATH,
                DEFAULT_BASE_CLIENT_OPEN_TELEMETRY_NAME,
            ),
            (False, False): (
                DEFAULT_BASE_CLIENT_PATH,
                DEFAULT_BASE_CLIENT_NAME,
            ),
        }
        if not self.base_client_name and not self.base_client_file_path:
            path, name = default_clients_map[
                (self.async_client, self.opentelemetry_client)
            ]
            self.base_client_name = name
            self.base_client_file_path = path.as_posix()

    @property
    def schema_source(self) -> str:
        return self.schema_path if self.schema_path else self.remote_schema_url

    @property
    def used_settings_message(self) -> str:
        snake_case_msg = (
            "Converting fields and arguments name to snake case."
            if self.convert_to_snake_case
            else "Not converting fields and arguments name to snake case."
        )
        async_client_msg = (
            "Generating async client."
            if self.async_client
            else "Generating not async client."
        )
        files_to_include_list = ",".join(self.files_to_include)
        files_to_include_msg = (
            f"Copying the following files into the package: {files_to_include_list}"
            if self.files_to_include
            else "No files to copy."
        )
        plugins_list = ",".join(self.plugins)
        plugins_msg = (
            f"Plugins to use: {plugins_list}"
            if self.plugins
            else "No plugin is being used."
        )
        include_typename_msg = (
            "Including __typename fields in generated queries."
            if self.include_typename
            else "Not including __typename fields in generated queries."
        )
        introspection_msg = (
            self._introspection_settings_message() if self.using_remote_schema else ""
        )
        return dedent(
            f"""\
            Selected strategy: {Strategy.CLIENT}
            Using schema from '{self.schema_path or self.remote_schema_url}'.
            {introspection_msg}
            Reading queries from '{self.queries_path}'.
            Using '{self.target_package_name}' as package name.
            Generating package into '{self.target_package_path}'.
            Using '{self.client_name}' as client name.
            Using '{self.base_client_name}' as base client class.
            Copying base client class from '{self.base_client_file_path}'.
            Generating enums into '{self.enums_module_name}.py'.
            Generating inputs into '{self.input_types_module_name}.py'.
            Generating fragments into '{self.fragments_module_name}.py'.
            Comments type: {self.include_comments.value}
            {snake_case_msg}
            {async_client_msg}
            {include_typename_msg}
            {files_to_include_msg}
            {plugins_msg}
            """
        )


@dataclass
class GraphQLSchemaSettings(BaseSettings):
    target_file_path: str = "schema.py"
    schema_variable_name: str = "schema"
    type_map_variable_name: str = "type_map"

    def __post_init__(self):
        super().__post_init__()

        assert_string_is_valid_schema_target_filename(self.target_file_path)
        assert_string_is_valid_python_identifier(self.schema_variable_name)
        assert_string_is_valid_python_identifier(self.type_map_variable_name)

    @property
    def used_settings_message(self):
        plugins_list = ",".join(self.plugins)
        plugins_msg = (
            f"Plugins to use: {plugins_list}"
            if self.plugins
            else "No plugin is being used."
        )
        introspection_msg = (
            self._introspection_settings_message() if self.using_remote_schema else ""
        )

        if self.target_file_format == "py":
            return dedent(
                f"""\
                Selected strategy: {Strategy.GRAPHQL_SCHEMA}
                Using schema from {self.schema_path or self.remote_schema_url}
                {introspection_msg}
                Saving graphql schema to: {self.target_file_path}
                Using {self.schema_variable_name} as variable name for schema.
                Using {self.type_map_variable_name} as variable name for type map.
                {plugins_msg}
                """
            )

        return dedent(
            f"""\
            Selected strategy: {Strategy.GRAPHQL_SCHEMA}
            Using schema from {self.schema_path or self.remote_schema_url}
            {introspection_msg}
            Saving graphql schema to: {self.target_file_path}
            {plugins_msg}
            """
        )

    @property
    def target_file_format(self):
        return Path(self.target_file_path).suffix[1:].lower()


def assert_path_exists(path: str):
    if not Path(path).exists():
        raise InvalidConfiguration(f"Provided path {path} doesn't exist.")


def assert_path_is_valid_directory(path: str):
    if not Path(path).is_dir():
        raise InvalidConfiguration(f"Provided path {path} isn't a directory.")


def assert_path_is_valid_file(path: str):
    if not Path(path).is_file():
        raise InvalidConfiguration(f"Provided path {path} isn't a file.")


def assert_string_is_valid_schema_target_filename(filename: str):
    file_type = Path(filename).suffix
    if not file_type:
        raise InvalidConfiguration(
            f"Provided file name {filename} is missing a file type."
        )

    file_type = file_type[1:].lower()
    if file_type not in ("py", "graphql", "gql"):
        raise InvalidConfiguration(
            f"Provided file name {filename} has an invalid type {file_type}."
            " Valid types are py, graphql and gql."
        )


def assert_string_is_valid_python_identifier(name: str):
    if not name.isidentifier() and not iskeyword(name):
        raise InvalidConfiguration(
            f"Provided name {name} cannot be used as python identifier."
        )


def resolve_headers(headers: dict) -> dict:
    return {key: get_header_value(value) for key, value in headers.items()}


def get_header_value(value: str) -> str:
    return resolve_env_vars_in_string(value)


def resolve_env_vars_in_string(value: str) -> str:
    """Replace $VAR and ${VAR} with values from the environment (any position).

    Only matches well-formed placeholders: ${VAR} or $VAR (variable name must
    start with a letter or underscore, then alphanumeric/underscore).
    """
    # Two explicit patterns so we never match malformed ${VAR or $VAR}
    ident = r"[A-Za-z_][A-Za-z0-9_]*"
    pattern = re.compile(rf"\$\{{({ident})\}}|\$({ident})")

    def replacer(match):
        env_var_name = match.group(1) or match.group(2)
        var_value = os.environ.get(env_var_name)
        if var_value is None or var_value == "":
            raise InvalidConfiguration(
                f"Environment variable {env_var_name} not found."
            )
        return var_value

    return pattern.sub(replacer, value)


def assert_class_is_defined_in_file(file_path: Path, class_name: str):
    file_content = file_path.read_text()
    if f"class {class_name}" not in file_content:
        raise InvalidConfiguration(f"Cannot import {class_name} from {file_path}")