ayon-core/client/ayon_core/settings/lib.py

import os
import json
import logging
import collections
import copy
import time

import ayon_api

log = logging.getLogger(__name__)


class CacheItem:
    lifetime = 10

    def __init__(self, value, outdate_time=None):
        self._value = value
        if outdate_time is None:
            outdate_time = time.time() + self.lifetime
        self._outdate_time = outdate_time

    @classmethod
    def create_outdated(cls):
        return cls({}, 0)

    def get_value(self):
        return copy.deepcopy(self._value)

    def update_value(self, value):
        self._value = value
        self._outdate_time = time.time() + self.lifetime

    @property
    def is_outdated(self):
        return time.time() > self._outdate_time


class _AyonSettingsCache:
    use_bundles = None
    variant = None
    addon_versions = CacheItem.create_outdated()
    studio_settings = CacheItem.create_outdated()
    cache_by_project_name = collections.defaultdict(
        CacheItem.create_outdated)

    @classmethod
    def _use_bundles(cls):
        if _AyonSettingsCache.use_bundles is None:
            major, minor, _, _, _ = ayon_api.get_server_version_tuple()
            use_bundles = True
            if (major, minor) < (0, 3):
                use_bundles = False
            _AyonSettingsCache.use_bundles = use_bundles
        return _AyonSettingsCache.use_bundles

    @classmethod
    def _get_variant(cls):
        if _AyonSettingsCache.variant is None:
            from ayon_core.lib import is_staging_enabled, is_dev_mode_enabled

            variant = "production"
            if is_dev_mode_enabled():
                variant = cls._get_bundle_name()
            elif is_staging_enabled():
                variant = "staging"

            # Cache variant
            _AyonSettingsCache.variant = variant

            # Set the variant to global ayon api connection
            ayon_api.set_default_settings_variant(variant)
        return _AyonSettingsCache.variant

    @classmethod
    def _get_bundle_name(cls):
        return os.environ["AYON_BUNDLE_NAME"]

    @classmethod
    def get_value_by_project(cls, project_name):
        cache_item = _AyonSettingsCache.cache_by_project_name[project_name]
        if cache_item.is_outdated:
            if cls._use_bundles():
                value = ayon_api.get_addons_settings(
                    bundle_name=cls._get_bundle_name(),
                    project_name=project_name,
                    variant=cls._get_variant()
                )
            else:
                value = ayon_api.get_addons_settings(project_name)
            cache_item.update_value(value)
        return cache_item.get_value()

    @classmethod
    def _get_addon_versions_from_bundle(cls):
        expected_bundle = cls._get_bundle_name()
        bundles = ayon_api.get_bundles()["bundles"]
        bundle = next(
            (
                bundle
                for bundle in bundles
                if bundle["name"] == expected_bundle
            ),
            None
        )
        if bundle is not None:
            return bundle["addons"]
        return {}

    @classmethod
    def get_addon_versions(cls):
        cache_item = _AyonSettingsCache.addon_versions
        if cache_item.is_outdated:
            if cls._use_bundles():
                addons = cls._get_addon_versions_from_bundle()
            else:
                settings_data = ayon_api.get_addons_settings(
                    only_values=False,
                    variant=cls._get_variant()
                )
                addons = settings_data["versions"]
            cache_item.update_value(addons)

        return cache_item.get_value()


def get_ayon_settings(project_name=None):
    """AYON studio settings.

    Raw AYON settings values.

    Args:
        project_name (Optional[str]): Project name.

    Returns:
        dict[str, Any]: AYON settings.
    """

    return _AyonSettingsCache.get_value_by_project(project_name)


def get_studio_settings(*args, **kwargs):
    return _AyonSettingsCache.get_value_by_project(None)


def get_project_settings(project_name, *args, **kwargs):
    return _AyonSettingsCache.get_value_by_project(project_name)


def get_general_environments(studio_settings=None):
    """General studio environment variables.

    Args:
        studio_settings (Optional[dict]): Pre-queried studio settings.

    Returns:
        dict[str, Any]: General studio environment variables.

    """
    if studio_settings is None:
        studio_settings = get_ayon_settings()
    return json.loads(studio_settings["core"]["environments"])


def get_project_environments(project_name, project_settings=None):
    """Project environment variables.

    Args:
        project_name (str): Project name.
        project_settings (Optional[dict]): Pre-queried project settings.

    Returns:
        dict[str, Any]: Project environment variables.

    """
    if project_settings is None:
        project_settings = get_project_settings(project_name)
    return json.loads(
        project_settings["core"]["project_environments"]
    )


def get_current_project_settings():
    """Project settings for current context project.

    Project name should be stored in environment variable `AYON_PROJECT_NAME`.
    This function should be used only in host context where environment
    variable must be set and should not happen that any part of process will
    change the value of the environment variable.
    """
    project_name = os.environ.get("AYON_PROJECT_NAME")
    if not project_name:
        raise ValueError(
            "Missing context project in environment"
            " variable `AYON_PROJECT_NAME`."
        )
    return get_project_settings(project_name)