# Copyright (C) 2010-2024 RhodeCode GmbH # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License, version 3 # (only), as published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . # # This program is dual-licensed. If you wish to learn more about the # RhodeCode Enterprise Edition, including its added features, Support services, # and proprietary license terms, please see https://rhodecode.com/licenses/ import os import re import logging import time import functools from collections import namedtuple from rhodecode.lib.pyramid_utils import get_current_request from rhodecode.lib import rc_cache from rhodecode.lib.hash_utils import sha1_safe from rhodecode.lib.html_filters import sanitize_html from rhodecode.lib.utils2 import Optional, AttributeDict, safe_str, remove_prefix, str2bool from rhodecode.lib.vcs.backends import base from rhodecode.lib.statsd_client import StatsdClient from rhodecode.model import BaseModel from rhodecode.model.db import RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi, RhodeCodeSetting from rhodecode.model.meta import Session log = logging.getLogger(__name__) UiSetting = namedtuple("UiSetting", ["section", "key", "value", "active"]) SOCIAL_PLUGINS_LIST = ["github", "bitbucket", "twitter", "google"] class SettingNotFound(Exception): def __init__(self, setting_id): msg = f"Setting `{setting_id}` is not found" super().__init__(msg) class SettingsModel(BaseModel): BUILTIN_HOOKS = ( RhodeCodeUi.HOOK_REPO_SIZE, RhodeCodeUi.HOOK_PUSH, RhodeCodeUi.HOOK_PRE_PUSH, RhodeCodeUi.HOOK_PRETX_PUSH, RhodeCodeUi.HOOK_PULL, RhodeCodeUi.HOOK_PRE_PULL, RhodeCodeUi.HOOK_PUSH_KEY, ) HOOKS_SECTION = "hooks" def __init__(self, sa=None, repo=None): self.repo = repo self.UiDbModel = RepoRhodeCodeUi if repo else RhodeCodeUi self.SettingsDbModel = RepoRhodeCodeSetting if repo else RhodeCodeSetting super().__init__(sa) def get_keyname(self, key_name, prefix="rhodecode_"): return f"{prefix}{key_name}" def get_ui_by_key(self, key): q = self.UiDbModel.query() q = q.filter(self.UiDbModel.ui_key == key) q = self._filter_by_repo(RepoRhodeCodeUi, q) return q.scalar() def get_ui_by_section(self, section): q = self.UiDbModel.query() q = q.filter(self.UiDbModel.ui_section == section) q = self._filter_by_repo(RepoRhodeCodeUi, q) return q.all() def get_ui_by_section_and_key(self, section, key): q = self.UiDbModel.query() q = q.filter(self.UiDbModel.ui_section == section) q = q.filter(self.UiDbModel.ui_key == key) q = self._filter_by_repo(RepoRhodeCodeUi, q) return q.scalar() def get_ui(self, section=None, key=None): q = self.UiDbModel.query() q = self._filter_by_repo(RepoRhodeCodeUi, q) if section: q = q.filter(self.UiDbModel.ui_section == section) if key: q = q.filter(self.UiDbModel.ui_key == key) # TODO: mikhail: add caching result = [ UiSetting( section=safe_str(r.ui_section), key=safe_str(r.ui_key), value=safe_str(r.ui_value), active=r.ui_active ) for r in q.all() ] return result def get_builtin_hooks(self): q = self.UiDbModel.query() q = q.filter(self.UiDbModel.ui_key.in_(self.BUILTIN_HOOKS)) return self._get_hooks(q) def get_custom_hooks(self): q = self.UiDbModel.query() q = q.filter(~self.UiDbModel.ui_key.in_(self.BUILTIN_HOOKS)) return self._get_hooks(q) def create_ui_section_value(self, section, val, key=None, active=True): new_ui = self.UiDbModel() new_ui.ui_section = section new_ui.ui_value = val new_ui.ui_active = active repository_id = "" if self.repo: repo = self._get_repo(self.repo) repository_id = repo.repo_id new_ui.repository_id = repository_id if not key: # keys are unique so they need appended info if self.repo: key = sha1_safe(f"{section}{val}{repository_id}") else: key = sha1_safe(f"{section}{val}") new_ui.ui_key = key Session().add(new_ui) return new_ui def create_or_update_hook(self, key, value): ui = self.get_ui_by_section_and_key(self.HOOKS_SECTION, key) or self.UiDbModel() ui.ui_section = self.HOOKS_SECTION ui.ui_active = True ui.ui_key = key ui.ui_value = value if self.repo: repo = self._get_repo(self.repo) repository_id = repo.repo_id ui.repository_id = repository_id Session().add(ui) return ui def delete_ui(self, id_): ui = self.UiDbModel.get(id_) if not ui: raise SettingNotFound(id_) Session().delete(ui) def get_setting_by_name(self, name): q = self._get_settings_query() q = q.filter(self.SettingsDbModel.app_settings_name == name) return q.scalar() def create_or_update_setting( self, name, val: Optional | str = Optional(""), type_: Optional | str = Optional("unicode") ): """ Creates or updates RhodeCode setting. If updates are triggered, it will only update parameters that are explicitly set Optional instance will be skipped :param name: :param val: :param type_: :return: """ res = self.get_setting_by_name(name) repo = self._get_repo(self.repo) if self.repo else None if not res: val = Optional.extract(val) type_ = Optional.extract(type_) args = (repo.repo_id, name, val, type_) if repo else (name, val, type_) res = self.SettingsDbModel(*args) else: if self.repo: res.repository_id = repo.repo_id res.app_settings_name = name if not isinstance(type_, Optional): # update if set res.app_settings_type = type_ if not isinstance(val, Optional): # update if set res.app_settings_value = val Session().add(res) return res def get_cache_region(self): repo = self._get_repo(self.repo) if self.repo else None cache_key = f"repo.v1.{repo.repo_id}" if repo else "repo.v1.ALL" cache_namespace_uid = f"cache_settings.{cache_key}" region = rc_cache.get_or_create_region("cache_general", cache_namespace_uid) return region, cache_namespace_uid def invalidate_settings_cache(self, hard=False): region, namespace_key = self.get_cache_region() log.debug( "Invalidation cache [%s] region %s for cache_key: %s", "invalidate_settings_cache", region, namespace_key ) # we use hard cleanup if invalidation is sent rc_cache.clear_cache_namespace(region, namespace_key, method=rc_cache.CLEAR_DELETE) def get_cache_call_method(self, cache=True): region, cache_key = self.get_cache_region() @region.conditional_cache_on_arguments(condition=cache) def _get_all_settings(name, key): q = self._get_settings_query() if not q: raise Exception("Could not get application settings !") settings = {self.get_keyname(res.app_settings_name): res.app_settings_value for res in q} return settings return _get_all_settings def get_all_settings(self, cache=False, from_request=True): # defines if we use GLOBAL, or PER_REPO repo = self._get_repo(self.repo) if self.repo else None # initially try the request context; this is the fastest # we only fetch global config, NOT for repo-specific if from_request and not repo: request = get_current_request() if request and hasattr(request, "call_context") and hasattr(request.call_context, "rc_config"): log.debug("Using cached request settings [%s]", request.call_context.rc_config) rc_config = request.call_context.rc_config if rc_config: return rc_config _region, cache_key = self.get_cache_region() _get_all_settings = self.get_cache_call_method(cache=cache) start = time.time() result = _get_all_settings("rhodecode_settings", cache_key) compute_time = time.time() - start log.debug("cached method:%s took %.4fs", _get_all_settings.__name__, compute_time) statsd = StatsdClient.statsd if statsd: elapsed_time_ms = round(1000.0 * compute_time) # use ms only statsd.timing("rhodecode_settings_timing.histogram", elapsed_time_ms, use_decimals=False) log.debug("Fetching app settings for key: %s took: %.4fs: cache: %s", cache_key, compute_time, cache) return result def get_auth_settings(self): q = self._get_settings_query() q = q.filter(self.SettingsDbModel.app_settings_name.startswith("auth_")) rows = q.all() auth_settings = {row.app_settings_name: row.app_settings_value for row in rows} return auth_settings def get_auth_plugins(self): auth_plugins = self.get_setting_by_name("auth_plugins") return auth_plugins.app_settings_value def get_default_repo_settings(self, strip_prefix=False): q = self._get_settings_query() q = q.filter(self.SettingsDbModel.app_settings_name.startswith("default_")) rows = q.all() result = {} for row in rows: key = row.app_settings_name if strip_prefix: key = remove_prefix(key, prefix="default_") result.update({key: row.app_settings_value}) return result def get_repo(self): repo = self._get_repo(self.repo) if not repo: raise Exception(f"Repository `{self.repo}` cannot be found inside the database") return repo def _filter_by_repo(self, model, query): if self.repo: repo = self.get_repo() query = query.filter(model.repository_id == repo.repo_id) return query def _get_hooks(self, query): query = query.filter(self.UiDbModel.ui_section == self.HOOKS_SECTION) query = self._filter_by_repo(RepoRhodeCodeUi, query) return query.all() def _get_settings_query(self): q = self.SettingsDbModel.query() return self._filter_by_repo(RepoRhodeCodeSetting, q) def list_enabled_social_plugins(self, settings): enabled = [] for plug in SOCIAL_PLUGINS_LIST: if str2bool(settings.get(f"rhodecode_auth_{plug}_enabled")): enabled.append(plug) return enabled def assert_repo_settings(func): @functools.wraps(func) def _wrapper(self, *args, **kwargs): if not self.repo_settings: raise Exception("Repository is not specified") return func(self, *args, **kwargs) return _wrapper class IssueTrackerSettingsModel(object): INHERIT_SETTINGS = "inherit_issue_tracker_settings" SETTINGS_PREFIX = "issuetracker_" def __init__(self, sa=None, repo=None): self.global_settings = SettingsModel(sa=sa) self.repo_settings = SettingsModel(sa=sa, repo=repo) if repo else None @property def inherit_global_settings(self): if not self.repo_settings: return True setting = self.repo_settings.get_setting_by_name(self.INHERIT_SETTINGS) return setting.app_settings_value if setting else True @inherit_global_settings.setter def inherit_global_settings(self, value): if self.repo_settings: settings = self.repo_settings.create_or_update_setting(self.INHERIT_SETTINGS, value, type_="bool") Session().add(settings) def _get_keyname(self, key, uid, prefix="rhodecode_"): return f"{prefix}{self.SETTINGS_PREFIX}{key}_{uid}" def _make_dict_for_settings(self, qs): prefix_match = self._get_keyname( "pat", "", ) issuetracker_entries = {} # create keys for k, v in qs.items(): if k.startswith(prefix_match): uid = k[len(prefix_match) :] issuetracker_entries[uid] = None def url_cleaner(input_str): input_str = input_str.replace('"', "").replace("'", "") input_str = sanitize_html(input_str, strip=True) return input_str # populate for uid in issuetracker_entries: url_data = qs.get(self._get_keyname("url", uid)) pat = qs.get(self._get_keyname("pat", uid)) try: pat_compiled = re.compile(r"%s" % pat) except re.error: pat_compiled = None issuetracker_entries[uid] = AttributeDict( { "pat": pat, "pat_compiled": pat_compiled, "url": url_cleaner(qs.get(self._get_keyname("url", uid)) or ""), "pref": sanitize_html(qs.get(self._get_keyname("pref", uid)) or ""), "desc": qs.get(self._get_keyname("desc", uid)), } ) return issuetracker_entries def get_global_settings(self, cache=False): """ Returns list of global issue tracker settings """ defaults = self.global_settings.get_all_settings(cache=cache) settings = self._make_dict_for_settings(defaults) return settings def get_repo_settings(self, cache=False): """ Returns list of issue tracker settings per repository """ if not self.repo_settings: raise Exception("Repository is not specified") all_settings = self.repo_settings.get_all_settings(cache=cache) settings = self._make_dict_for_settings(all_settings) return settings def get_settings(self, cache=False): if self.inherit_global_settings: return self.get_global_settings(cache=cache) else: return self.get_repo_settings(cache=cache) def delete_entries(self, uid): if self.repo_settings: all_patterns = self.get_repo_settings() settings_model = self.repo_settings else: all_patterns = self.get_global_settings() settings_model = self.global_settings entries = all_patterns.get(uid, []) for del_key in entries: setting_name = self._get_keyname(del_key, uid, prefix="") entry = settings_model.get_setting_by_name(setting_name) if entry: Session().delete(entry) Session().commit() def create_or_update_setting(self, name, val=Optional(""), type_=Optional("unicode")): if self.repo_settings: setting = self.repo_settings.create_or_update_setting(name, val, type_) else: setting = self.global_settings.create_or_update_setting(name, val, type_) return setting class VcsSettingsModel(object): INHERIT_SETTINGS = "inherit_vcs_settings" GENERAL_SETTINGS = ( "use_outdated_comments", "pr_merge_enabled", "auto_merge_enabled", "hg_use_rebase_for_merging", "hg_close_branch_before_merging", "hg_merge_strategy_selector", "git_use_rebase_for_merging", "git_close_branch_before_merging", "git_merge_strategy_selector", "diff_cache", ) HOOKS_SETTINGS = ( ("hooks", "changegroup.repo_size"), ("hooks", "changegroup.push_logger"), ("hooks", "outgoing.pull_logger"), ) HG_SETTINGS = ( ("extensions", "largefiles"), ("phases", "publish"), ("extensions", "evolve"), ("extensions", "topic"), ("experimental", "evolution"), ("experimental", "evolution.exchange"), ) GIT_SETTINGS = (("vcs_git_lfs", "enabled"),) GLOBAL_HG_SETTINGS = ( ("extensions", "largefiles"), ("phases", "publish"), ("extensions", "evolve"), ("extensions", "topic"), ("experimental", "evolution"), ("experimental", "evolution.exchange"), ) GLOBAL_GIT_SETTINGS = (("vcs_git_lfs", "enabled"),) SVN_BRANCH_SECTION = "vcs_svn_branch" SVN_TAG_SECTION = "vcs_svn_tag" PATH_SETTING = ("paths", "/") def __init__(self, sa=None, repo=None): self.global_settings = SettingsModel(sa=sa) self.repo_settings = SettingsModel(sa=sa, repo=repo) if repo else None self._ui_settings = self.HG_SETTINGS + self.GIT_SETTINGS + self.HOOKS_SETTINGS self._svn_sections = (self.SVN_BRANCH_SECTION, self.SVN_TAG_SECTION) @property @assert_repo_settings def inherit_global_settings(self): setting = self.repo_settings.get_setting_by_name(self.INHERIT_SETTINGS) return setting.app_settings_value if setting else True @inherit_global_settings.setter @assert_repo_settings def inherit_global_settings(self, value): self.repo_settings.create_or_update_setting(self.INHERIT_SETTINGS, value, type_="bool") def get_keyname(self, key_name, prefix="rhodecode_"): return f"{prefix}{key_name}" def get_global_svn_branch_patterns(self): return self.global_settings.get_ui_by_section(self.SVN_BRANCH_SECTION) @assert_repo_settings def get_repo_svn_branch_patterns(self): return self.repo_settings.get_ui_by_section(self.SVN_BRANCH_SECTION) def get_global_svn_tag_patterns(self): return self.global_settings.get_ui_by_section(self.SVN_TAG_SECTION) @assert_repo_settings def get_repo_svn_tag_patterns(self): return self.repo_settings.get_ui_by_section(self.SVN_TAG_SECTION) def get_global_settings(self): return self._collect_all_settings(global_=True) @assert_repo_settings def get_repo_settings(self): return self._collect_all_settings(global_=False) @assert_repo_settings def get_repo_settings_inherited(self): global_settings = self.get_global_settings() global_settings.update(self.get_repo_settings()) return global_settings @assert_repo_settings def create_or_update_repo_settings(self, data, inherit_global_settings=False): from rhodecode.model.scm import ScmModel self.inherit_global_settings = inherit_global_settings repo = self.repo_settings.get_repo() if not inherit_global_settings: if repo.repo_type == "svn": self.create_repo_svn_settings(data) else: self.create_or_update_repo_hook_settings(data) self.create_or_update_repo_pr_settings(data) if repo.repo_type == "hg": self.create_or_update_repo_hg_settings(data) if repo.repo_type == "git": self.create_or_update_repo_git_settings(data) ScmModel().mark_for_invalidation(repo.repo_name, delete=True) @assert_repo_settings def create_or_update_repo_hook_settings(self, data): for section, key in self.HOOKS_SETTINGS: data_key = self._get_form_ui_key(section, key) if data_key not in data: raise ValueError(f"The given data does not contain {data_key} key") active = data.get(data_key) repo_setting = self.repo_settings.get_ui_by_section_and_key(section, key) if not repo_setting: global_setting = self.global_settings.get_ui_by_section_and_key(section, key) self.repo_settings.create_ui_section_value(section, global_setting.ui_value, key=key, active=active) else: repo_setting.ui_active = active Session().add(repo_setting) def update_global_hook_settings(self, data): for section, key in self.HOOKS_SETTINGS: data_key = self._get_form_ui_key(section, key) if data_key not in data: raise ValueError(f"The given data does not contain {data_key} key") active = data.get(data_key) repo_setting = self.global_settings.get_ui_by_section_and_key(section, key) repo_setting.ui_active = active Session().add(repo_setting) @assert_repo_settings def create_or_update_repo_pr_settings(self, data): return self._create_or_update_general_settings(self.repo_settings, data) def create_or_update_global_pr_settings(self, data): return self._create_or_update_general_settings(self.global_settings, data) @assert_repo_settings def create_repo_svn_settings(self, data): return self._create_svn_settings(self.repo_settings, data) def _set_evolution(self, settings, is_enabled): if is_enabled: # if evolve is active set evolution=all self._create_or_update_ui(settings, *("experimental", "evolution"), value="all", active=True) self._create_or_update_ui(settings, *("experimental", "evolution.exchange"), value="yes", active=True) # if evolve is active set topics server support self._create_or_update_ui(settings, *("extensions", "topic"), value="", active=True) else: self._create_or_update_ui(settings, *("experimental", "evolution"), value="", active=False) self._create_or_update_ui(settings, *("experimental", "evolution.exchange"), value="no", active=False) self._create_or_update_ui(settings, *("extensions", "topic"), value="", active=False) @assert_repo_settings def create_or_update_repo_hg_settings(self, data): largefiles, phases, evolve = self.HG_SETTINGS[:3] largefiles_key, phases_key, evolve_key = self._get_settings_keys(self.HG_SETTINGS[:3], data) self._create_or_update_ui(self.repo_settings, *largefiles, value="", active=data[largefiles_key]) self._create_or_update_ui(self.repo_settings, *evolve, value="", active=data[evolve_key]) self._set_evolution(self.repo_settings, is_enabled=data[evolve_key]) self._create_or_update_ui(self.repo_settings, *phases, value=safe_str(data[phases_key])) def create_or_update_global_hg_settings(self, data): opts_len = 3 largefiles, phases, evolve = self.GLOBAL_HG_SETTINGS[:opts_len] largefiles_key, phases_key, evolve_key = self._get_settings_keys(self.GLOBAL_HG_SETTINGS[:opts_len], data) self._create_or_update_ui(self.global_settings, *largefiles, value="", active=data[largefiles_key]) self._create_or_update_ui(self.global_settings, *phases, value=safe_str(data[phases_key])) self._create_or_update_ui(self.global_settings, *evolve, value="", active=data[evolve_key]) self._set_evolution(self.global_settings, is_enabled=data[evolve_key]) def create_or_update_repo_git_settings(self, data): # NOTE(marcink): # comma makes unpack work properly (lfs_enabled,) = self.GIT_SETTINGS (lfs_enabled_key,) = self._get_settings_keys(self.GIT_SETTINGS, data) self._create_or_update_ui( self.repo_settings, *lfs_enabled, value=data[lfs_enabled_key], active=data[lfs_enabled_key] ) def create_or_update_global_git_settings(self, data): lfs_enabled = self.GLOBAL_GIT_SETTINGS[0] lfs_enabled_key = self._get_settings_keys(self.GLOBAL_GIT_SETTINGS, data)[0] self._create_or_update_ui( self.global_settings, *lfs_enabled, value=data[lfs_enabled_key], active=data[lfs_enabled_key] ) def create_or_update_global_svn_settings(self, data): # branch/tags patterns self._create_svn_settings(self.global_settings, data) @assert_repo_settings def delete_repo_svn_pattern(self, id_): ui = self.repo_settings.UiDbModel.get(id_) if ui and ui.repository.repo_name == self.repo_settings.repo: # only delete if it's the same repo as initialized settings self.repo_settings.delete_ui(id_) else: # raise error as if we wouldn't find this option self.repo_settings.delete_ui(-1) def delete_global_svn_pattern(self, id_): self.global_settings.delete_ui(id_) @assert_repo_settings def get_repo_ui_settings(self, section=None, key=None): global_uis = self.global_settings.get_ui(section, key) repo_uis = self.repo_settings.get_ui(section, key) filtered_repo_uis = self._filter_ui_settings(repo_uis) filtered_repo_uis_keys = [(s.section, s.key) for s in filtered_repo_uis] def _is_global_ui_filtered(ui): return (ui.section, ui.key) in filtered_repo_uis_keys or ui.section in self._svn_sections filtered_global_uis = [ui for ui in global_uis if not _is_global_ui_filtered(ui)] return filtered_global_uis + filtered_repo_uis def get_global_ui_settings(self, section=None, key=None): return self.global_settings.get_ui(section, key) def get_ui_settings_as_config_obj(self, section=None, key=None): config = base.Config() ui_settings = self.get_ui_settings(section=section, key=key) for entry in ui_settings: config.set(entry.section, entry.key, entry.value) return config def get_ui_settings(self, section=None, key=None): if not self.repo_settings or self.inherit_global_settings: return self.get_global_ui_settings(section, key) else: return self.get_repo_ui_settings(section, key) def get_svn_patterns(self, section=None): if not self.repo_settings: return self.get_global_ui_settings(section) else: return self.get_repo_ui_settings(section) @assert_repo_settings def get_repo_general_settings(self): global_settings = self.global_settings.get_all_settings() repo_settings = self.repo_settings.get_all_settings() filtered_repo_settings = self._filter_general_settings(repo_settings) global_settings.update(filtered_repo_settings) return global_settings def get_global_general_settings(self): return self.global_settings.get_all_settings() def get_general_settings(self): if not self.repo_settings or self.inherit_global_settings: return self.get_global_general_settings() else: return self.get_repo_general_settings() def _filter_ui_settings(self, settings): filtered_settings = [s for s in settings if self._should_keep_setting(s)] return filtered_settings def _should_keep_setting(self, setting): keep = (setting.section, setting.key) in self._ui_settings or setting.section in self._svn_sections return keep def _filter_general_settings(self, settings): keys = [self.get_keyname(key) for key in self.GENERAL_SETTINGS] return {k: settings[k] for k in settings if k in keys} def _collect_all_settings(self, global_=False): settings = self.global_settings if global_ else self.repo_settings result = {} for section, key in self._ui_settings: ui = settings.get_ui_by_section_and_key(section, key) result_key = self._get_form_ui_key(section, key) if ui: if section in ("hooks", "extensions"): result[result_key] = ui.ui_active elif result_key in ["vcs_git_lfs_enabled"]: result[result_key] = ui.ui_active else: result[result_key] = ui.ui_value for name in self.GENERAL_SETTINGS: setting = settings.get_setting_by_name(name) if setting: result_key = self.get_keyname(name) result[result_key] = setting.app_settings_value return result def _get_form_ui_key(self, section, key): return "{section}_{key}".format(section=section, key=key.replace(".", "_")) def _create_or_update_ui(self, settings, section, key, value=None, active=None): ui = settings.get_ui_by_section_and_key(section, key) if not ui: active = True if active is None else active settings.create_ui_section_value(section, value, key=key, active=active) else: if active is not None: ui.ui_active = active if value is not None: ui.ui_value = value Session().add(ui) def _create_svn_settings(self, settings, data): svn_settings = {"new_svn_branch": self.SVN_BRANCH_SECTION, "new_svn_tag": self.SVN_TAG_SECTION} for key in svn_settings: if data.get(key): settings.create_ui_section_value(svn_settings[key], data[key]) def _create_or_update_general_settings(self, settings, data): for name in self.GENERAL_SETTINGS: data_key = self.get_keyname(name) if data_key not in data: raise ValueError(f"The given data does not contain {data_key} key") setting = settings.create_or_update_setting(name, data[data_key], "bool") Session().add(setting) def _get_settings_keys(self, settings, data): data_keys = [self._get_form_ui_key(*s) for s in settings] for data_key in data_keys: if data_key not in data: raise ValueError(f"The given data does not contain {data_key} key") return data_keys