@solid(
    description="A solid to invoke ranch run.",
    input_defs=[InputDefinition(name="start", dagster_type=Nothing, description="Nothing.")],
    output_defs=[OutputDefinition(name="end", dagster_type=Nothing, description="Nothing.")],
    config={
        "command": Field(
            dagster_type=String, is_optional=False, description="The command to run against the ranch application.",
        ),
        "app": Field(
            dagster_type=String, is_optional=False, description="The ranch application to run the command against.",
        ),
        "env": Field(
            dagster_type=PermissiveDict(),
            default_value=None,
            is_optional=True,
            description="An optional dict of environment variables to pass to the subprocess.",
        ),
        "output_logging": Field(
            dagster_type=Enum(
                name="OutputType",
                enum_values=[
                    EnumValue("STREAM", description="Stream script stdout/stderr."),
                    EnumValue("BUFFER", description="Buffer bash script stdout/stderr, then log upon completion.",),
                    EnumValue("NONE", description="No logging"),
                ],
            ),
            is_optional=True,
            default_value="STREAM",
        ),
    },
    required_resource_keys={"ranch"},
)
def ranch_run(context):

    bash_command = f"ranch run {context.solid_config['command']}" + f" --app {context.solid_config['app']}"

    <http://context.log.info|context.log.info>("Setting up environment for subprocess..")
    if context.solid_config["env"] is not None:
        env = context.solid_config["env"]
    else:
        env = os.environ.copy()
    env.update({"RANCH_TOKEN": context.resources.ranch.token})

    <http://context.log.info|context.log.info>(f"Temporary directory root location: \n {gettempdir()}")

    with TemporaryDirectory(prefix="dagstertmp_") as tmpdir:
        with NamedTemporaryFile(dir=tmpdir, prefix=context.run_id) as f:

            f.write(bytes(bash_command, "utf-8"))
            f.flush()
            fname = f.name
            script_location = os.path.abspath(fname)
            <http://context.log.info|context.log.info>(f"Temporary script location: {script_location}")

            def pre_exec():
                for sig in ("SIGPIPE", "SIGXFZ", "SIGXFSZ"):
                    if hasattr(signal, sig):
                        signal.signal(getattr(signal, sig), signal.SIG_DFL)
                os.setsid()

            <http://context.log.info|context.log.info>(f"Running command: {bash_command}")

            sub_process = Popen(["bash", f.name], stdout=PIPE, stderr=STDOUT, cwd=tmpdir, env=env, preexec_fn=pre_exec,)

            # Stream back logs as they are emitted
            if context.solid_config["output_logging"] == "STREAM":
                line = ""
                for raw_line in iter(sub_process.stdout.readline, b""):
                    line = raw_line.decode("utf-8").rstrip()
                    <http://context.log.info|context.log.info>(line)

            sub_process.wait()

            # Collect and buffer all logs, then emit
            if context.solid_config["output_logging"] == "BUFFER":
                line = ""
                for raw_line in iter(sub_process.stdout.readline, b""):
                    line += raw_line.decode("utf-8")
                <http://context.log.info|context.log.info>(line)

            # no logging in this case
            elif context.solid_config["output_logging"] == "NONE":
                pass

            <http://context.log.info|context.log.info>(f"Command exited with return code {sub_process.returncode}")

            if sub_process.returncode:
                raise Failure(description=f"Bash command failed: {bash_command}")


def gsheet_wrap_ranch_run_fn(cfg) -> Dict:

    bash_command = (
        "ranch run -- /entrypoint /project/platform/utils/launch.sh"
        + f" ./bin/mk-sheets-importer --id {cfg['gsheet_id']}"
    )

    return {
        "ranch_run": {
            "config": {
                "command": bash_command,
                "app": "etl-ng-mk-sheets-importer",
                "env": cfg["env"],
                "output_logging": cfg["output_logging"],
            }
        }
    }


@composite_solid(
    description="A solid to invoke a ranch process to ingest a gsheet to the data warehouse.",
    input_defs=[InputDefinition(name="start", dagster_type=Nothing, description="Nothing.")],
    output_defs=[OutputDefinition(name="end", dagster_type=Nothing, description="Nothing.")],
    config={
        "ghseet_id": Field(dagster_type=String, is_optional=False, description="The ID of the gsheet to ingest.",),
        "env": Field(
            dagster_type=PermissiveDict(),
            default_value=None,
            is_optional=True,
            description="An optional dict of environment variables to pass to the subprocess.",
        ),
        "output_logging": Field(
            dagster_type=Enum(
                name="OutputType",
                enum_values=[
                    EnumValue("STREAM", description="Stream script stdout/stderr."),
                    EnumValue("BUFFER", description="Buffer bash script stdout/stderr, then log upon completion.",),
                    EnumValue("NONE", description="No logging"),
                ],
            ),
            is_optional=True,
            default_value="STREAM",
        ),
    },
    config_fn=gsheet_wrap_ranch_run_fn,
)
def ranch_ingest_gsheet():
    return ranch_run()

ranch_ingest_gsheet

dagster.core.errors.DagsterInvalidDefinitionError: @composite_solid 'ranch_ingest_gsheet' decorated function does not have parameter(s) 'start', which are in solid's input_defs. Solid functions should only have keyword arguments that match input names and a first positional parameter named 'context'.

return ranch_run(start)

config_fn

cfg

def gsheet_wrap_ranch_run_fn(_, cfg) -> Dict:

Nothing

dagster-bash