path: root/openbb_platform/core/openbb_core/app/command_runner.py

"""Command runner module."""

from copy import deepcopy
from dataclasses import asdict, is_dataclass
from datetime import datetime
from inspect import Parameter, signature
from sys import exc_info
from time import perf_counter_ns
from typing import Any, Callable, Dict, List, Optional, Tuple
from warnings import catch_warnings, showwarning, warn

from fastapi.params import Query
from pydantic import BaseModel, ConfigDict, create_model

from openbb_core.app.logs.logging_service import LoggingService
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_core.app.model.abstract.warning import OpenBBWarning, cast_warning
from openbb_core.app.model.command_context import CommandContext
from openbb_core.app.model.metadata import Metadata
from openbb_core.app.model.obbject import OBBject
from openbb_core.app.model.system_settings import SystemSettings
from openbb_core.app.model.user_settings import UserSettings
from openbb_core.app.provider_interface import ProviderInterface
from openbb_core.app.router import CommandMap
from openbb_core.app.service.system_service import SystemService
from openbb_core.app.service.user_service import UserService
from openbb_core.env import Env
from openbb_core.provider.utils.helpers import maybe_coroutine, run_async


class ExecutionContext:
    """Execution context."""

    def __init__(
        self,
        command_map: CommandMap,
        route: str,
        system_settings: SystemSettings,
        user_settings: UserSettings,
    ) -> None:
        self.command_map = command_map
        self.route = route
        self.system_settings = system_settings
        self.user_settings = user_settings


class ParametersBuilder:
    """Build parameters for a function."""

    @staticmethod
    def get_polished_parameter_list(func: Callable) -> List[Parameter]:
        """Get the signature parameters values as a list."""
        sig = signature(func)
        parameter_list = list(sig.parameters.values())

        return parameter_list

    @staticmethod
    def get_polished_func(func: Callable) -> Callable:
        """Remove __authenticated_user_settings from the function signature and annotations."""
        func = deepcopy(func)
        sig = signature(func)
        parameter_map = dict(sig.parameters)

        if "__authenticated_user_settings" in parameter_map:
            parameter_map.pop("__authenticated_user_settings")

        parameter_list = list(parameter_map.values())
        new_signature = signature(func).replace(parameters=parameter_list)

        func.__signature__ = new_signature  # type: ignore
        func.__annotations__ = parameter_map

        return func

    @classmethod
    def merge_args_and_kwargs(
        cls,
        func: Callable,
        args: Tuple[Any],
        kwargs: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Merge args and kwargs into a single dict."""
        args = deepcopy(args)
        kwargs = deepcopy(kwargs)
        parameter_list = cls.get_polished_parameter_list(func=func)
        parameter_map = {}

        for index, parameter in enumerate(parameter_list):
            if index < len(args):
                parameter_map[parameter.name] = args[index]
            elif parameter.name in kwargs:
                parameter_map[parameter.name] = kwargs[parameter.name]
            elif parameter.default is not parameter.empty:
                parameter_map[parameter.name] = parameter.default
            else:
                parameter_map[parameter.name] = None

        return parameter_map

    @staticmethod
    def update_command_context(
        func: Callable,
        kwargs: Dict[str, Any],
        system_settings: SystemSettings,
        user_settings: UserSettings,
    ) -> Dict[str, Any]:
        """Update the command context with the available user and system settings."""
        argcount = func.__code__.co_argcount
        if "cc" in func.__code__.co_varnames[:argcount]:
            kwargs["cc"] = CommandContext(
                user_settings=user_settings,
                system_settings=system_settings,
            )

        return kwargs

    @staticmethod
    def update_provider_choices(
        func: Callable,
        command_coverage: Dict[str, List[str]],
        route: str,
        kwargs: Dict[str, Any],
        route_default: Optional[Dict[str, Optional[str]]],
    ) -> Dict[str, Any]:
        """Update the provider choices with the available providers and set default provider."""

        def _needs_provider(func: Callable) -> bool:
            """Check if the function needs a provider."""
            parameters = signature(func).parameters.keys()
            return "provider_choices" in parameters

        def _has_provider(kwargs: Dict[str, Any]) -> bool:
            """Check if the kwargs already have a provider."""
            provider_choices = kwargs.get("provider_choices", None)

            if isinstance(provider_choices, dict):  # when in python
                return provider_choices.get("provider", None) is not None
            if isinstance(provider_choices, object):  # when running as fastapi
                return getattr(provider_choices, "provider", None) is not None
            return False

        def _get_first_provider() -> Optional[str]:
            """Get the first available provider."""
            available_providers = ProviderInterface().available_providers
            return available_providers[0] if available_providers else None

        def _get_default_provider(
            command_coverage: Dict[str, List[str]],
            route_default: Optional[Dict[str, Optional[str]]],
        ) -> Optional[str]:
            """
            Get the default provider for the given route.

            Either pick it from the user defaults or from the command coverage.
            """
            cmd_cov_given_route = command_coverage.get(route, None)
            command_cov_provider = (
                cmd_cov_given_route[0] if cmd_cov_given_route else None
            )

            if route_default:
                return route_default.get("provider", None) or command_cov_provider  # type: ignore

            return command_cov_provider

        if not _has_provider(kwargs) and _needs_provider(func):
            provider = (
                _get_default_provider(
                    command_coverage,
                    route_default,
                )
                if route in command_coverage
                else _get_first_provider()
            )
            kwargs["provider_choices"] = {"provider": provider}

        return kwargs

    @staticmethod
    def _warn_kwargs(
        provider_choices: Dict[str, Any],
        extra_params: Dict[str, Any],
        model: BaseModel,
    ) -> None:
        """Warn if kwargs received and ignored by the validation model."""
        # We only check the extra_params annotation because ignored fields
        # will always be kwargs
        annotation = getattr(
            model.model_fields.get("extra_params", None), "annotation", None
        )
        if annotation:
            # When there is no annotation there is nothing to warn
            valid = asdict(annotation())