@graph

to_job

repo.py

ins

out

@graph

ins

out

@graph

Missing required config entry "inputs" at the root

inputs

Error 1: Received unexpected config entry "inputs" at the root.

UserWarning: Module builder was resolved using the working directory. The ability to load uninstalled modules from the working directory is deprecated and will be removed in a future release.  Please use the python-file based load arguments or install builder to your python environment.

warnings.warn(

reconstructable_pipeline = build_reconstructable_pipeline(
            'builder',
            'reconstruct_pipeline',
            reconstructable_args=(pipe,)
        )

output = execute_pipeline(
            pipeline=reconstructable_pipeline,
            run_config=conf,
            instance=DagsterInstance.local_temp(),
        ).output_capture

dagit

dagit

Operation name: RootworkspaceQuery. psycopg2.OperationError server closed the connection unexpectedly.

@repository
def ingestion_pipeline_repository():
    return get_pipelines() + [my_schedule]


@schedule(cron_schedule="0 22 * * *", pipeline_name="my_pipeline", execution_timezone="US/Eastern")
def my_schedule(context):
    return {}

build_op_context()

../../dagster_0.13.17/lib/python3.9/site-packages/dagster/core/execution/context/invocation.py:459: in build_op_context
    return build_solid_context(
../../dagster_0.13.17/lib/python3.9/site-packages/dagster/core/execution/context/invocation.py:512: in build_solid_context
    return UnboundSolidExecutionContext(
../../dagster_0.13.17/lib/python3.9/site-packages/dagster/core/execution/context/invocation.py:68: in __init__
    self._instance = self._instance_cm.__enter__()  # pylint: disable=no-member
/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/contextlib.py:119: in __enter__
    return next(self.gen)
../../dagster_0.13.17/lib/python3.9/site-packages/dagster/core/execution/api.py:333: in ephemeral_instance_if_missing
    with DagsterInstance.ephemeral() as ephemeral_instance:
../../dagster_0.13.17/lib/python3.9/site-packages/dagster/core/instance/__init__.py:363: in ephemeral
    tempdir = DagsterInstance.temp_storage()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    @staticmethod
    def temp_storage() -> str:
>       from dagster.core.test_utils import environ
E       ImportError: cannot import name 'environ' from 'dagster.core.test_utils' (/Users/zachary.romer/Documents/empirico/etxpipelines/dagster_0.13.17/lib/python3.9/site-packages/dagster/core/test_utils.py)

../../dagster_0.13.17/lib/python3.9/site-packages/dagster/core/instance/__init__.py:488: ImportError

def test_nominal_case(self, mock_file):
        context = build_op_context(op_config={'bucket': 'test-bucket',
                                              'key': 'test-key'})

run_config

@repository
def my_repo():
    return {"jobs": {"my_job": lambda: my_job}}

my_job

JobDefinition

def storage_bucket(dependent_resource_name):
    @resource(
        required_resource_keys={dependent_resource_name},
    )
    def _storage_bucket(init_context):
        credentials = init_context.resources[dependent_resource_name]
        return StorageBucket(credentials)
    return _storage_bucket

credentials = init_context.resources[dependent_resource_name]

TypeError: tuple indices must be integers or slices, not str

init_context.resources

_ScopeResources

_ScopedResource

__getitem__

_ScopedResources._asdict()

GCPFileManager.read_data

blah.tar.gz

Operation name: RootworkspaceQuery. psycopg2.OperationError server closed the connection unexpectedly.

@op(ins={'front': In(str)}, out=Out(None))
def start(context, front):
	<http://context.log.info|context.log.info>(f'{front}')

	yield Output(None)


@job
def data_parser():
    start(1)
    start(1)

job_data_parser_schedule = ScheduleDefinition(cron_schedule='00 00 * * *', job=data_parser,
                                              execution_timezone='Europe/London')


@repository
def repository_parser():
    return [job_data_parser_schedule]

dagster.core.errors.DagsterInvalidDefinitionError: In @job data_parser, received invalid type <class 'str'> for input "front" (at position 0) in op invocation "start". Must pass the output from previous node invocations or inputs to the composition function as inputs when invoking nodes during composition.

volumes:
      - ./files:/work

get_runs

run_key

PipelineRunsFilter

job

op

class FolderExtractor:
    def __init__(
        self, path_to_input: str
    ) -> None:
       # some code

    def run(self) -> list[str]:
       # other code

class DataFrameBuilder:
    def __init__(self, files: list[str]) -> None:
       # code 

    def run(self) -> pd.DataFrame:
       # code

DataFrameBuilder(FolderExtractor(inp).run()).run()

@op
def folder_extractor: 

@op
def data_frame_builder

op