context.log

define_asset_job

dagster._core.errors.DagsterInvariantViolationError: Attempting to access log, but it was not provided when constructing the OutputContext

def load_input(self, context):
        <http://context.log.info|context.log.info>(f"CSVxxxxxxxxxxxxx\n {context} AND {context.upstream_output}")
        <http://context.log.info|context.log.info>(dir(InitResourceContext))
        # remove because dagstermill processes dont seem to have a context
        # <http://context.log.info|context.log.info>(f"{context.metadata}<>{context.name}<>{context.resource_config}")
        if context.has_asset_key or (context.upstream_output is None and 'file_name' in context.resource_config): # input manager
            <http://context.log.info|context.log.info>("xxxxxxxxxxx Input Manager Path")
            path = self._get_path(context)
        else:
            <http://context.log.info|context.log.info>("xxxxxxxxxxx Upstrem Output Path")
            bla = context.upstream_output
            <http://context.log.info|context.log.info>(dir(bla))
            <http://context.log.info|context.log.info>(bla.asset_info)
            path = self._get_path(context.upstream_output)
        #path = self._get_path(context)
        with path.open("rb") as file:
            return pd.read_csv(file)

if context.has_asset_key or (context.upstream_output is None and 'file_name' in context.resource_config):

SourceAsset

SourceAsset

context.has_asset_key

load_input

input_manager

iomanager

SourceAssets:

def load_input(self, context):
        <http://context.log.info|context.log.info>(f"PPxxxxxxxxxxxxx\n {context} AND {context.upstream_output}")
        source_asset = False
        asset_mode = False
        if context.has_asset_key: # this io manager is being used in an asset op
            asset_mode = True
            <http://context.log.info|context.log.info>("asset_mode=True")
            <http://context.log.info|context.log.info>(context.upstream_output.asset_key.path[0])
            if context.upstream_output.asset_key.path[0][0:7]=='source_':
                source_asset = True
        if (source_asset and asset_mode) or (context.upstream_output is None and 'file_name' in context.resource_config): # input manager
            #if context.upstream_output is None and 'file_name' in context.resource_config: # input manager
            <http://context.log.info|context.log.info>("xxxxxxxxxxx Input Manager Path")
            path = self._get_path(context)
        else:
            <http://context.log.info|context.log.info>("xxxxxxxxxxx Upstrem Output Path")
            bla = context.upstream_output
            <http://context.log.info|context.log.info>(dir(bla))
            <http://context.log.info|context.log.info>(bla.asset_info)
            path = self._get_path(context.upstream_output)
        with path.open("rb") as file:
            return pd.read_parquet(file)

context._log.

In this case I would want to use the regular io manager anyway so am not sure how to branch. I need a predicate which says that this is the "root" of the graph...

upstream_output

And does it makes sense that _log is not set in the upstream context?

target

target_downstream

So everything works out!