Stefan Nenzén
30361abdd9
Create a VisibilityHandler Co-authored-by: nenzen <stefan@nenzen.com> Reviewed-on: #3 Reviewed-by: erth9960 <thuning@dsv.su.se> Co-authored-by: Stefan Nenzén <nenzen@dsv.su.se> Co-committed-by: Stefan Nenzén <nenzen@dsv.su.se>
58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
import copy
|
|
from pathlib import Path
|
|
|
|
from .handler import Handler
|
|
from ..exceptions import ValidationException
|
|
from ..package import Package
|
|
|
|
@Handler.register
|
|
class VisibilityHandler(Handler):
|
|
"""
|
|
This class handles visibility settings for streams.
|
|
"""
|
|
@classmethod
|
|
def wants(cls, jobspec: dict, existing_package: Package) -> bool | ValidationException:
|
|
"""
|
|
Return True if this handler wants to process this jobspec.
|
|
Raises an exception if the job is wanted but doesn't pass validation.
|
|
|
|
In order for a job to be wanted, the field 'sources' must exist and
|
|
at least one of the source items must contain a 'enabled' field.
|
|
|
|
"""
|
|
if 'sources' in jobspec and any('enabled' in source for source in jobspec['sources'].values()):
|
|
return cls._validate(jobspec, existing_package)
|
|
return False
|
|
|
|
@classmethod
|
|
def _validate(cls, jobspec: dict, existing_package: Package) -> bool | ValidationException:
|
|
"""
|
|
Return True if the job is valid for this handler.
|
|
|
|
A job is valid as long as at least one of the package's source items
|
|
has its 'enabled' field set to True.
|
|
"""
|
|
sources = copy.deepcopy(existing_package.get('sources', {}))
|
|
for name, source in jobspec['sources'].items():
|
|
if name not in sources:
|
|
sources[name] = {}
|
|
if 'enabled' in source:
|
|
sources[name]['enabled'] = source['enabled']
|
|
for source in sources.values():
|
|
if 'enabled' in source:
|
|
return True
|
|
raise ValidationException("No enabled sources")
|
|
|
|
def _handle(self, jobspec: dict, existing_package: Package, tempdir: Path) -> callable:
|
|
"""
|
|
"""
|
|
def apply_func(package: Package) -> None:
|
|
sources = package.get('sources', {})
|
|
for name, source in jobspec['sources'].items():
|
|
if name not in sources:
|
|
sources[name] = {}
|
|
if 'enabled' in source:
|
|
sources[name]['enabled'] = source['enabled']
|
|
|
|
return apply_func
|