import logging
from .transform import loader as transform_loader
logger = logging.getLogger(__name__)
DEFAULT_TRANSFORMS = [
"taskgraph.transforms.run:transforms",
"taskgraph.transforms.task:transforms",
]
def loader(kind, path, config, params, loaded_tasks):
transform_refs = config.setdefault("transforms", [])
for t in DEFAULT_TRANSFORMS:
if t in config.get("transforms", ()):
raise KeyError(
f"Transform {t} is already present in the loader's default transforms; it must not be defined in the kind"
)
transform_refs.extend(DEFAULT_TRANSFORMS)
return transform_loader(kind, path, config, params, loaded_tasks)