Page MenuHomeElementl
Differential D4837 Diff 24279 python_modules/dagster/dagster/core/definitions/decorators/repository.py

Changeset View

Standalone View

View Options

python_modules/dagster/dagster/core/definitions/decorators/repository.py

from functools import update_wrapper from functools import update_wrapper
from dagster import check from dagster import check
from dagster.core.errors import DagsterInvalidDefinitionError from dagster.core.errors import DagsterInvalidDefinitionError
from ..executable import ExecutableDefinition from ..job import JobDefinition
from ..partition import PartitionSetDefinition from ..partition import PartitionSetDefinition
from ..pipeline import PipelineDefinition from ..pipeline import PipelineDefinition
from ..repository import VALID_REPOSITORY_DATA_DICT_KEYS, RepositoryData, RepositoryDefinition from ..repository import VALID_REPOSITORY_DATA_DICT_KEYS, RepositoryData, RepositoryDefinition
from ..schedule import ScheduleDefinition from ..schedule import ScheduleDefinition
class _Repository(object): class _Repository(object):
def __init__(self, name=None, description=None): def __init__(self, name=None, description=None):
Show All 21 Lines def __call__(self, fn):
if isinstance(repository_definitions, list): if isinstance(repository_definitions, list):
bad_definitions = [] bad_definitions = []
for i, definition in enumerate(repository_definitions): for i, definition in enumerate(repository_definitions):
if not ( if not (
isinstance(definition, PipelineDefinition) isinstance(definition, PipelineDefinition)
or isinstance(definition, PartitionSetDefinition) or isinstance(definition, PartitionSetDefinition)
or isinstance(definition, ScheduleDefinition) or isinstance(definition, ScheduleDefinition)
or isinstance(definition, ExecutableDefinition) or isinstance(definition, JobDefinition)
): ):
bad_definitions.append((i, type(definition))) bad_definitions.append((i, type(definition)))
if bad_definitions: if bad_definitions:
raise DagsterInvalidDefinitionError( raise DagsterInvalidDefinitionError(
"Bad return value from repository construction function: all elements of list " "Bad return value from repository construction function: all elements of list "
"must be of type PipelineDefinition, PartitionSetDefinition, " "must be of type PipelineDefinition, PartitionSetDefinition, "
"ScheduleDefinition, or ExecutableDefinition. Got {bad_definitions_formatted}.".format( "ScheduleDefinition, or JobDefinition. Got {bad_definitions_formatted}.".format(
bad_definitions_formatted=", ".join( bad_definitions_formatted=", ".join(
[ [
"value of type {type_} at index {i}".format(type_=type_, i=i) "value of type {type_} at index {i}".format(type_=type_, i=i)
for i, type_ in bad_definitions for i, type_ in bad_definitions
] ]
) )
) )
) )
repository_data = RepositoryData.from_list(repository_definitions) repository_data = RepositoryData.from_list(repository_definitions)
elif isinstance(repository_definitions, dict): elif isinstance(repository_definitions, dict):
if not set(repository_definitions.keys()).issubset(VALID_REPOSITORY_DATA_DICT_KEYS): if not set(repository_definitions.keys()).issubset(VALID_REPOSITORY_DATA_DICT_KEYS):
raise DagsterInvalidDefinitionError( raise DagsterInvalidDefinitionError(
"Bad return value from repository construction function: dict must not contain " "Bad return value from repository construction function: dict must not contain "
"keys other than {{'pipelines', 'partition_sets', 'schedules', 'executables'}}: found " "keys other than {{'pipelines', 'partition_sets', 'schedules', 'jobs'}}: found "
"{bad_keys}".format( "{bad_keys}".format(
bad_keys=", ".join( bad_keys=", ".join(
[ [
"'{key}'" "'{key}'"
for key in repository_definitions.keys() for key in repository_definitions.keys()
if key not in VALID_REPOSITORY_DATA_DICT_KEYS if key not in VALID_REPOSITORY_DATA_DICT_KEYS
] ]
) )
▲ Show 20 Lines • Show All 153 Lines • Show Last 20 Lines