From 123139b35304a94becc99986e2a3d05345c52544 Mon Sep 17 00:00:00 2001 From: rdimaio Date: Wed, 27 Mar 2024 13:41:28 +0100 Subject: [PATCH] Testing: Add type annotations for schema functions; #6588 --- lib/rucio/common/schema/__init__.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/lib/rucio/common/schema/__init__.py b/lib/rucio/common/schema/__init__.py index 31d383f05f..75e0b1f05f 100644 --- a/lib/rucio/common/schema/__init__.py +++ b/lib/rucio/common/schema/__init__.py @@ -15,15 +15,19 @@ import importlib from configparser import NoOptionError, NoSectionError from os import environ +from typing import TYPE_CHECKING, Any from rucio.common import config, exception from rucio.common.utils import check_policy_package_version +if TYPE_CHECKING: + from types import ModuleType + # dictionary of schema modules for each VO -schema_modules = {} +schema_modules: dict[str, "ModuleType"] = {} # list of unique SCOPE_NAME_REGEXP values from all schemas -scope_name_regexps = [] +scope_name_regexps: list[str] = [] try: multivo = config.config_get_bool('common', 'multi_vo', check_config_table=False) @@ -63,7 +67,7 @@ scope_name_regexps.append(module.SCOPE_NAME_REGEXP) -def load_schema_for_vo(vo): +def load_schema_for_vo(vo: str) -> None: GENERIC_FALLBACK = 'generic_multi_vo' if config.config_has_section('policy'): try: @@ -92,19 +96,19 @@ def load_schema_for_vo(vo): schema_modules[vo] = module -def validate_schema(name, obj, vo='def'): +def validate_schema(name: str, obj: Any, vo: str = 'def') -> None: if vo not in schema_modules: load_schema_for_vo(vo) schema_modules[vo].validate_schema(name, obj) -def get_schema_value(key, vo='def'): +def get_schema_value(key: str, vo: str = 'def') -> Any: if vo not in schema_modules: load_schema_for_vo(vo) return getattr(schema_modules[vo], key) -def get_scope_name_regexps(): +def get_scope_name_regexps() -> list[str]: """ returns a list of all unique SCOPE_NAME_REGEXPs from all schemas """ if len(scope_name_regexps) == 0: @@ -120,7 +124,7 @@ def get_scope_name_regexps(): return scope_name_regexps -def insert_scope_name(urls): +def insert_scope_name(urls: tuple[str, ...]) -> tuple[str, str]: """ given a tuple of URLs for webpy with '%s' as a placeholder for SCOPE_NAME_REGEXP, return a finalised tuple of URLs that will work for all