Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, solid, depends_on=None, depended_by=None):
super(DauphinSolid, self).__init__(name=solid.name)
check.opt_dict_param(depends_on, 'depends_on', key_type=SolidInputHandle, value_type=list)
check.opt_dict_param(
depended_by, 'depended_by', key_type=SolidOutputHandle, value_type=list
)
self._solid = check.inst_param(solid, 'solid', Solid)
if depends_on:
self.depends_on = {
input_handle: output_handles for input_handle, output_handles in depends_on.items()
}
else:
self.depends_on = {}
if depended_by:
self.depended_by = {
output_handle: input_handles for output_handle, input_handles in depended_by.items()
}
def __new__(cls, resource_instance_dict=None):
return super(ScopedResourcesBuilder, cls).__new__(
cls,
resource_instance_dict=check.opt_dict_param(
resource_instance_dict, 'resource_instance_dict', key_type=str
),
def create_execution_plan(pipeline, environment_dict=None, mode=None):
check.inst_param(pipeline, 'pipeline', PipelineDefinition)
environment_dict = check.opt_dict_param(environment_dict, 'environment_dict', key_type=str)
check.opt_str_param(mode, 'mode')
environment_config = create_environment_config(pipeline, environment_dict)
return ExecutionPlan.build(pipeline, environment_config)
):
from dagster.core.definitions.pipeline import ExecutionSelector
tags = check.opt_dict_param(tags, 'tags', key_type=str)
selector = check.opt_inst_param(selector, 'selector', ExecutionSelector)
if not selector:
selector = ExecutionSelector(pipeline_name)
if not status:
status = PipelineRunStatus.NOT_STARTED
return super(PipelineRun, cls).__new__(
cls,
pipeline_name=check.str_param(pipeline_name, 'pipeline_name'),
run_id=check.str_param(run_id, 'run_id'),
environment_dict=check.opt_dict_param(
environment_dict, 'environment_dict', key_type=str
),
mode=check.str_param(mode, 'mode'),
selector=selector,
reexecution_config=reexecution_config, # deprecated
step_keys_to_execute=None
if step_keys_to_execute is None
else check.list_param(step_keys_to_execute, 'step_keys_to_execute', of_type=str),
status=status,
tags=check.opt_dict_param(tags, 'tags', key_type=str),
previous_run_id=check.opt_str_param(previous_run_id, 'previous_run_id'),
)
def execute_query(handle, query, variables=None, use_sync_executor=False, instance=None):
check.inst_param(handle, 'handle', ExecutionTargetHandle)
check.str_param(query, 'query')
check.opt_dict_param(variables, 'variables')
# We allow external creation of the pipeline_run_storage to support testing contexts where we
# need access to the underlying run storage
instance = check.opt_inst_param(instance, 'instance', DagsterInstance, DagsterInstance.get())
check.bool_param(use_sync_executor, 'use_sync_executor')
query = query.strip('\'" \n\t')
execution_manager = SynchronousExecutionManager()
context = DagsterGraphQLContext(
handle=handle, instance=instance, execution_manager=execution_manager, version=__version__
)
executor = SyncExecutor() if use_sync_executor else GeventExecutor()
result = graphql(
)
# Only used for Airflow; internally we continue to use pipeline.name
dag_id = check.opt_str_param(dag_id, 'dag_id', _rename_for_airflow(pipeline_name))
dag_description = check.opt_str_param(
dag_description, 'dag_description', _make_dag_description(pipeline_name)
)
check.subclass_param(operator, 'operator', BaseOperator)
dag_kwargs = dict(
{'default_args': DEFAULT_ARGS},
**check.opt_dict_param(dag_kwargs, 'dag_kwargs', key_type=str)
)
op_kwargs = check.opt_dict_param(op_kwargs, 'op_kwargs', key_type=str)
dag = DAG(dag_id=dag_id, description=dag_description, **dag_kwargs)
pipeline = handle.build_pipeline_definition()
if mode is None:
mode = pipeline.get_default_mode_name()
execution_plan = create_execution_plan(
pipeline, environment_dict, run_config=RunConfig(mode=mode)
)
tasks = {}
coalesced_plan = coalesce_execution_steps(execution_plan)
def __init__(
self,
task_id,
environment_dict=None,
pipeline_name=None,
mode=None,
step_keys=None,
dag=None,
instance_ref=None,
*args,
**kwargs
):
check.str_param(pipeline_name, 'pipeline_name')
step_keys = check.opt_list_param(step_keys, 'step_keys', of_type=str)
environment_dict = check.opt_dict_param(environment_dict, 'environment_dict', key_type=str)
check.opt_inst_param(instance_ref, 'instance_ref', InstanceRef)
kwargs['name'] = 'dagster.{pipeline_name}.{task_id}'.format(
pipeline_name=pipeline_name, task_id=task_id
).replace(
'_', '-' # underscores are not permissible DNS names
)
if 'storage' not in environment_dict:
raise AirflowException(
'No storage config found -- must configure either filesystem or s3 storage for '
'the DagsterKubernetesPodOperator. Ex.: \n'
'storage:\n'
' filesystem:\n'
' base_dir: \'/some/shared/volume/mount/special_place\''
'\n\n --or--\n\n'
def check_opt_two_dim_dict(ddict, param_name, key_type=None, value_type=None):
ddict = check.opt_dict_param(ddict, param_name, key_type=key_type, value_type=dict)
for sub_dict in ddict.values():
check.dict_param(sub_dict, 'sub_dict', key_type=key_type, value_type=value_type)
return ddict
def create_environment_config(pipeline, environment_dict=None, mode=None):
check.inst_param(pipeline, 'pipeline', PipelineDefinition)
check.opt_dict_param(environment_dict, 'environment')
check.opt_str_param(mode, 'mode')
environment_type = create_environment_type(pipeline, mode)
result = evaluate_config_value(environment_type, environment_dict)
if not result.success:
raise PipelineConfigEvaluationError(pipeline, result.errors, environment_dict)
return construct_environment_config(result.value)
def __init__(
self,
name,
cron_schedule,
pipeline_name,
environment_dict=None,
environment_dict_fn=None,
tags=None,
tags_fn=None,
mode="default",
should_execute=lambda: True,
environment_vars=None,
):
check.str_param(pipeline_name, 'pipeline_name')
check.str_param(mode, 'mode')
check.opt_dict_param(environment_dict, 'environment_dict')
check.opt_callable_param(environment_dict_fn, 'environment_dict_fn')
check.opt_dict_param(tags, 'tags', key_type=str, value_type=str)
check.opt_callable_param(tags_fn, 'tags_fn')
check.callable_param(should_execute, 'should_execute')
if environment_dict:
warnings.warn(
"The `environment_dict` argument to `ScheduleDefinition` is deprecated. "
"Use the `environment_dict_fn` argument instead. "
"The argument `environment_dict_fn` should be a function that takes `DagsterInstance` "
"as an argument and returns an environment dict",
DeprecationWarning,
)
if tags:
warnings.warn(