Skip to content

Commit

Permalink
Testing: Add type annotations for schema functions; #6588
Browse files Browse the repository at this point in the history
  • Loading branch information
rdimaio authored and bari12 committed Apr 17, 2024
1 parent 2772147 commit 71d75d5
Showing 1 changed file with 11 additions and 7 deletions.
18 changes: 11 additions & 7 deletions lib/rucio/common/schema/__init__.py
Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand Down

0 comments on commit 71d75d5

Please sign in to comment.