Source code for onetick.py.configuration

import os
from datetime import datetime
from typing import Iterable, Type, Union
from contextlib import suppress
from textwrap import dedent

import dotenv
import onetick.query as otq
from .utils import default_concurrency, default_license_dir, default_license_file

DEFAULT_CONCURRENCY = default_concurrency()
DEFAULT_LICENSE_DIR = default_license_dir()
DEFAULT_LICENSE_FILE = default_license_file()

DATETIME_FORMATS = (
    '%Y/%m/%d %H:%M:%S.%f',
    '%Y/%m/%d %H:%M:%S',
)
DATETIME_DESCRIPTION = (
    "Format of the env variable: "
    f"{', '.join(':code:`{}`'.format(fmt) for fmt in DATETIME_FORMATS)}."
)


def parse_datetime(s):
    for fmt in DATETIME_FORMATS:
        with suppress(ValueError):
            return datetime.strptime(s, fmt)
    raise ValueError(
        f"The datetime pattern is not supported for string '{s}'. "
        f"Available patterns: {DATETIME_FORMATS}"
    )


class _nothing(type):
    def __repr__(cls):
        return cls.__name__


class nothing(metaclass=_nothing):
    """
    This is nothing.
    """


class OtpProperty:
    """
    .. attribute:: {name}
       :type: {base_type}
       :value: {base_value}

       {description}

       {env_var_name}

       {env_var_desc}
    """
    def __init__(self, description, base_default, env_var_name=None, env_var_func=None,
                 env_var_desc=None, set_value=nothing, allowed_types: Union[Type, Iterable] = nothing):
        self._base_default = base_default
        self._env_var_name = env_var_name
        self._env_var_func = env_var_func
        self._env_var_desc = env_var_desc
        self._set_value = set_value
        self._description = description
        if self._base_default is nothing:
            self._allowed_types = []
        else:
            self._allowed_types = [type(self._base_default)]
        if allowed_types is not nothing:
            if isinstance(allowed_types, Iterable):
                self._allowed_types.extend(allowed_types)
            else:
                self._allowed_types.append(allowed_types)
        self._allowed_types = tuple(set(self._allowed_types))
        # will be monkeypatched later
        self._name = None

    def _get_doc(self, name):
        env_var_name = ''
        env_var_desc = ''
        if self._env_var_name is not None:
            env_var_name = f'Can be set using environment variable :envvar:`{self._env_var_name}`.'
            if self._env_var_desc is not None:
                env_var_desc = self._env_var_desc
        return self.__class__.__doc__.format(
            name=name,
            description=self._description,
            base_type=','.join(t.__name__ for t in self._allowed_types),
            base_value=repr(self._base_default),
            env_var_name=env_var_name,
            env_var_desc=env_var_desc
        )

    def __get__(self, obj, objtype=None):
        if self._set_value is not nothing:
            return self._set_value
        if self._env_var_name:
            env_var_value = os.environ.get(self._env_var_name, None)
            if env_var_value is not None:
                if self._env_var_func:
                    return self._env_var_func(env_var_value)
                return env_var_value
        if obj is not None:
            # get value from default config
            if self._env_var_name and self._env_var_name in obj.default_config:
                var_value = obj.default_config[self._env_var_name]
                if self._env_var_func:
                    return self._env_var_func(var_value)
                return var_value
        if self._base_default is nothing:
            raise ValueError(f'onetick.py.config.{self._name} is not set!')
        return self._base_default

    def __set__(self, obj, value):
        # assigning to nothing is permitted
        # assigning to nothing will reset value to default
        if not isinstance(value, self._allowed_types) and value is not nothing:
            raise ValueError(f'Type of passed configuration value "{type(value)}" should be one of '
                             f'the allowed types for this configuration {self._allowed_types}')
        self._set_value = value


class OtpDerivedProperty:
    """
    .. attribute:: {name}
       :type: {base_type}
       :value: {base_value}

       {description}
    """

    def __init__(self, description, definition_function):
        self._description = description
        self.__definition_function = definition_function

    def __get__(self, obj, objtype=None):
        return self.__definition_function(obj)

    def _get_doc(self, name, base_object):
        value = self.__definition_function(base_object, docs=True)
        return self.__doc__.format(
            name=name,
            description=self._description,
            base_type=type(value).__name__,
            base_value=value,
        )

    def __set__(self, obj, value):
        raise AttributeError('It\'s not allowed to change a derived property. Change source properties, and its value '
                             'will be updated automatically.')


class OtpShowStackInfoProperty(OtpProperty):
    """
    .. attribute:: {name}
       :type: {base_type}
       :value: {base_value}

       {description}
    """
    @staticmethod
    def parser(value):
        return str(value).lower() in ('1', 'true', 'yes')

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # set default value on module loading
        self.__set_in_onetick_query__()

    def __get__(self, obj, objtype=None):
        value = super().__get__(obj, objtype)
        return self.parser(value)

    def __set__(self, obj, value):
        super().__set__(obj, value)
        self.__set_in_onetick_query__()

    def __set_in_onetick_query__(self):
        value = 1 if self.__get__(None) else 0
        otq.config.API_CONFIG['SHOW_STACK_INFO'] = value


def document_config(cls):
    manually_set_properties_doc = ''
    changeable_properties_doc = ''
    derived_properties_doc = ''

    manually_set_options = cls.manually_set_options()
    for b in manually_set_options:
        manually_set_properties_doc += cls.__dict__[b]._get_doc(b)
    for c in cls.get_changeable_config_options():
        cls.__dict__[c]._name = c
        if c not in manually_set_options:
            changeable_properties_doc += cls.__dict__[c]._get_doc(c)
    for d in cls.get_derived_config_options():
        cls.__dict__[d]._name = d
        derived_properties_doc += cls.__dict__[d]._get_doc(d, base_object=cls)

    cls.__doc__ = cls.__doc__.format(
        manually_set_properties=manually_set_properties_doc,
        changeable_properties=changeable_properties_doc,
        derived_properties=derived_properties_doc,
    )
    cls.__doc__ = dedent(cls.__doc__)

    return cls


@document_config
class Config:
    """
    This object is used to access ``onetick.py`` configuration variables.

    Configuration variables may be accessed via :code:`otp.config['...']` syntax, e.g. :code:`otp.config['tz']`.

    Configuration variables may be changed by:

    * during python runtime by modifying properties of object ``otp.config``,
    * by setting environment variables *before* importing ``onetick.py`` module.

    Also special environment variable ``OTP_DEFAULT_CONFIG_PATH`` can be used to specify a file,
    from which configuration variables will be taken.
    This file will be read only once on module loading or when getting one of the configuration variables
    when the environment variable is discovered.
    The names of the variables in this file are the same as the names of environment variables.

    In case several methods of setting configuration variables are used,
    the following order of priority is in place:

    1. Value that is set by modifying object ``otp.config``
    2. Value that is set via environment variable
    3. Value that is set in file ``OTP_DEFAULT_CONFIG_PATH``
    4. Default value specified in the source code

    To reset configuration value that has been set by modifying object ``otp.config``,
    special value ``otp.config.default`` should be assigned to it.

    Most of the config vars are optional and have default values,
    but some of them need to be set manually.

    There are also some environment variables that do not have
    corresponding property in ``otp.config`` object:

    * ``OTP_BASE_FOLDER_FOR_GENERATED_RESOURCE``:
      a folder where all intermediate queries, files and databases
      generated by ``onetick-py`` are located.
      The default value is system-dependent, e.g. some generated
      directory with a unique name under a standard directory **/tmp** for Linux.

    **The following properties must be set manually in most cases:**
    {manually_set_properties}

    **The following properties can be changed:**
    {changeable_properties}

    **The following properties are derived and thus read-only:**
    {derived_properties}
    """
    __default_config = None

    @property
    def default_config(self):
        default_config_path = os.environ.get('OTP_DEFAULT_CONFIG_PATH')
        if not default_config_path:
            return {}
        if self.__default_config is None:
            default_config = dotenv.dotenv_values(default_config_path)
            available_option_names = []
            for name, option in self.get_changeable_config_options().items():
                if option._env_var_name:
                    available_option_names.append(option._env_var_name)
            diff = set(default_config).difference(available_option_names)
            if diff:
                raise ValueError(f'Configuration options {diff} from file'
                                 f' OTP_DEFAULT_CONFIG_PATH="{default_config_path}" are not supported.'
                                 f' Available options: {available_option_names}.')
            Config.__default_config = default_config
        return self.__default_config or {}

    def __getitem__(self, item):
        if item not in self.__class__.__dict__.keys():
            raise AttributeError(f'"{item}" is not in the list of onetick.py config options!')
        return self.__class__.__dict__[item].__get__(self)

    def __setitem__(self, item, value):
        if item not in self.__class__.__dict__.keys():
            raise AttributeError(f'"{item}" is not in the list of onetick.py config options!')
        self.__class__.__dict__[item].__set__(self, value)

    def get(self, key, default=None):
        try:
            return self.__getitem__(key)
        except ValueError:
            return default

    def __setattr__(self, item, value):
        """
        To avoid accidental declaration of non-existing properties, e.g. `otp.config.timezone = "GMT"`
        """
        self.__setitem__(item, value)

    @classmethod
    def get_changeable_config_options(cls):
        """
        useful for tests where you may want to memorize all existing configuration options before changing them
        """
        return {
            option: value
            for option, value in cls.__dict__.items()
            if isinstance(value, OtpProperty)
        }

    @classmethod
    def get_derived_config_options(cls):
        return {
            option: value
            for option, value in cls.__dict__.items()
            if isinstance(value, OtpDerivedProperty)
        }

    @classmethod
    def manually_set_options(cls):
        return {
            option: value
            for option, value in cls.__dict__.items()
            if isinstance(value, OtpProperty) and value._base_default is nothing
        }

    default = nothing

    tz = OtpProperty(
        description='Default timezone used for running queries and creating databases, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`. '
                    'Default value is the local timezone of your machine.',
        base_default=None,
        allowed_types=str,
        env_var_name='OTP_DEFAULT_TZ',
    )

    context = OtpProperty(
        description='Default context used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`.',
        base_default='DEFAULT',
        env_var_name='OTP_CONTEXT',
    )

    default_start_time = OtpProperty(
        description='Default start time used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`.',
        base_default=nothing,
        allowed_types=datetime,
        env_var_name='OTP_DEFAULT_START_TIME',
        env_var_func=parse_datetime,
        env_var_desc=DATETIME_DESCRIPTION,
    )

    default_end_time = OtpProperty(
        description='Default end time used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`.',
        base_default=nothing,
        allowed_types=datetime,
        env_var_name='OTP_DEFAULT_END_TIME',
        env_var_func=parse_datetime,
        env_var_desc=DATETIME_DESCRIPTION,
    )

    default_db = OtpProperty(
        description='Default database name used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`.',
        base_default=nothing,
        allowed_types=str,
        env_var_name='OTP_DEFAULT_DB',
    )

    default_symbol = OtpProperty(
        description='Default symbol name used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`.',
        base_default=nothing,
        allowed_types=str,
        env_var_name='OTP_DEFAULT_SYMBOL',
    )

    default_symbology = OtpProperty(
        description='Default database symbology.',
        base_default='BZX',
        env_var_name='OTP_DEFAULT_SYMBOLOGY',
    )

    def default_db_symbol(obj, docs=False):  # noqa
        try:
            return obj.default_db + '::' + obj.default_symbol
        except (ValueError, TypeError):
            if not docs:
                raise

    default_db_symbol = OtpDerivedProperty(
        description='Default symbol with database. '
                    'Defined with :py:attr:`default_db` and :py:attr:`default_symbol` '
                    'as string **default_db::default_symbol**.',
        definition_function=default_db_symbol,
    )

    def default_date(obj, docs=False):  # noqa
        try:
            return datetime.combine(obj.default_start_time.date(), datetime.min.time())
        except (ValueError, TypeError, AttributeError):
            if not docs:
                raise

    default_date = OtpDerivedProperty(
        description='Default date. '
                    'Defined as a date part of :py:attr:`default_start_time`.',
        definition_function=default_date,
    )

    # default concurrency is set to the number of cores on the machine
    default_concurrency = OtpProperty(
        description='Default concurrency level used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`. '
                    'Default value is the number of cores/threads on your machine.',
        base_default=DEFAULT_CONCURRENCY,
        env_var_name='OTP_DEFAULT_CONCURRENCY',
        env_var_func=int,
    )

    # default batch size is set to 0, so the number of symbols in batch is not limited
    # it should work better in simple cases, but may use too much memory for complex queries
    default_batch_size = OtpProperty(
        description='Default batch size used for running queries, '
                    'e.g. with :py:func:`otp.run<onetick.py.run>`. '
                    'Batch size is the maximum number of symbols that are processed at once. '
                    'The value of 0 means unlimited -- works faster for simple queries, '
                    'but may consume too much memory for complex queries.',
        base_default=0,
        env_var_name='OTP_DEFAULT_BATCH_SIZE',
        env_var_func=int,
    )

    default_license_dir = OtpProperty(
        description='Default path for license directory. '
                    'Needed for user to be allowed to use OneTick API. '
                    'Default value is system-dependent: '
                    '**/license** for Linux systems and '
                    '**C:/OMD/client_data/config/license_repository** for Windows systems.',
        base_default=DEFAULT_LICENSE_DIR,
        env_var_name='OTP_DEFAULT_LICENSE_DIR',
        allowed_types=str,
    )

    default_license_file = OtpProperty(
        description='Default path for license file. '
                    'Needed for user to be allowed to use OneTick API. '
                    'Default value is system-dependent: '
                    '**/license/license.dat** for Linux systems and '
                    '**C:/OMD/client_data/config/license.dat** for Windows systems.',
        base_default=DEFAULT_LICENSE_FILE,
        env_var_name='OTP_DEFAULT_LICENSE_FILE',
        allowed_types=str,
    )

    default_fault_tolerance = OtpProperty(
        description='Default value for USE_FT query property.',
        base_default='FALSE',
        env_var_name='OTP_DEFAULT_FAULT_TOLERANCE',
        allowed_types=str,
    )

    default_auth_username = OtpProperty(
        description='Default username used for authentication.',
        base_default=None,
        allowed_types=str,
        env_var_name='OTP_DEFAULT_AUTH_USERNAME',
    )

    default_password = OtpProperty(
        description='Default password used for authentication.',
        base_default=None,
        allowed_types=str,
        env_var_name='OTP_DEFAULT_PASSWORD',
    )

    max_expected_ticks_per_symbol = OtpProperty(
        description='Expected maximum number of ticks per symbol (used for performance optimizations).',
        base_default=2000,
        allowed_types=int,
        env_var_name='OTP_MAX_EXPECTED_TICKS_PER_SYMBOL',
    )

    show_stack_info = OtpShowStackInfoProperty(
        description='Show stack info (filename and line or stack trace) in OneTick exceptions.',
        base_default=False,
        allowed_types=(str, bool, int),
        env_var_name='OTP_SHOW_STACK_INFO',
        env_var_func=OtpShowStackInfoProperty.parser,
    )

    log_symbol = OtpProperty(
        description='Log currently executed symbol. Note, this only works with unbound symbols. '
                    'Note, in this case :py:func:`otp.run<onetick.py.run>` does not produce the output '
                    'so it should be used only for debugging purposes.',
        base_default=False,
        allowed_types=(str, bool, int),
        env_var_name='OTP_LOG_SYMBOL',
        env_var_func=OtpShowStackInfoProperty.parser,
    )


def get_options_table(cls):
    options_table = ('\n'
                     '.. csv-table::\n'
                     '   :header: "Name", "Environment Variable", "Description"\n'
                     '   :widths: auto\n\n')
    for name, prop in cls.get_changeable_config_options().items():
        name = f':py:attr:`otp.config.{name}<onetick.py.configuration.Config.{name}>`'
        options_table += f'   {name},"``{prop._env_var_name}``","{prop._description}"\n'
    for name, prop in cls.get_derived_config_options().items():
        name = f':py:attr:`otp.config.{name}<onetick.py.configuration.Config.{name}>`'
        options_table += f'   {name},"","{prop._description}"\n'
    return options_table


class OptionsTable:
    __doc__ = get_options_table(Config)


config = Config()