Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Testing: Add type annotations for schema functions #6612

Merged
merged 1 commit into from Apr 17, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
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