play-daemon-threaded/pipeline/utils.py
Erik Thuning 11917608bb Made raise_for_structure provide better feedback on errors
The function now takes an argument corresponding to the key that is
being worked on. Also made the handling of dict validation more explicit
about keys and the reasoning around why it does what it does.
2024-06-18 11:26:20 +02:00

118 lines
4.2 KiB
Python

from types import FunctionType
"""
The canonical structure of a package going into the pipeline.
"""
canonical_jobspec = {
'pkg_id': str,
'upload_dir': str,
'title': {'en': str,
'sv': str},
'description': str,
'created': int,
'presenters': [str],
'courses': [{'designation': str,
'semester': str}],
'tags': [str],
'thumb': str,
'subtitles': {str: str},
'generate_subtitles': {str: {'type': str,
'language': str,
'source': str}},
'sources': {str: {'poster': str,
'playAudio': bool,
'enabled': bool,
'video': str}},
'slides': str,
'notification_url': str,
}
"""
The canonical structure of a fully populated package as stored on disk.
"""
canonical_manifest = {
'title': {'en': str,
'sv': str},
'description': str,
'created': lambda v: (isinstance(v, int)
or v is None
or f"{type(v)}, expected int"),
'duration': lambda v: (isinstance(v, float)
or v is None
or f"{type(v)}, expected float"),
'presenters': [str],
'courses': [{'designation': str,
'semester': str}],
'tags': [str],
'thumb': str,
'subtitles': {str: str},
'sources': {str: {'poster': str,
'playAudio': bool,
'enabled': bool,
'video': {'720': str,
'1080': str}}}
}
def raise_for_structure(data, structure, key=''):
"""
Validate the structure of a json-like object (data) against an expected
format (structure), throwing an exception on invalid structure.
The data object need not contain all keys in the format, but provided keys
must match the format and have valid values.
"""
if isinstance(structure, type):
# Base case, validate that data type matches structure type
if not isinstance(data, structure):
raise ValueError(
f"Invalid type for {key}: {type(data)}, expected {structure}. "
f"Value was {data}.")
elif isinstance(structure, FunctionType):
# The expected type is complicated and defined by function. Call it.
result = structure(data)
if result != True:
raise ValueError(f"Invalid type for {key}: {result}. "
f"Value was {data}.")
elif isinstance(structure, list):
# List case, validate that data is list and recurse over items
if not isinstance(data, list):
raise ValueError(
f"Invalid type for {key}: {type(data)}, expected list. "
f"Value was {data}.")
for c in data:
raise_for_structure(c, structure[0], f"{key} listitem")
elif isinstance(structure, dict):
# Dict case, validate that data is dict
# and recurse over keys and values
if not isinstance(data, dict):
raise ValueError(
f"Invalid type for {key}: {type(data)}, expected dict. "
f"Value was {data}.")
for k in data.keys():
if k not in structure.keys():
# Maybe the key in structure is a type?
# In that case there must only be one key in the structure.
str_key = list(structure.keys())[0]
if (len(structure.keys()) == 1
and isinstance(str_key, type)):
# Check that all keys in that structure are the right type
raise_for_structure(k, str_key, f"{key} dictkey")
# Keep checking the values as normal
raise_for_structure(data[k], structure[str_key], k)
else:
raise KeyError(f"Unexpected key {k}")
else:
raise_for_structure(data[k], structure[k], k)
return True
def get_section(config, name):
"""
ConfigParser objects implement the dict.get() interface in a highly
surprising manner for sections, so we need a wrapper.
"""
if name in config:
return config[name]
return {}