from dagster import fs_io_manager, op, job, configured
from dagster_dask import dask_executor
from semaphore_scripts import template_runner
import os
from pathlib import Path

@op(
    config_schema={
        "command": str,
        "command_parameter": str,
        "semaphore_audience_version": str,
        "workspace": str
    },
    required_resource_keys={"io_manager"}
)
def template_runner_op(context):
    workspace: Path = Path(
        os.path.join(context.op_config["workspace"], f'{context.op_config["command"]}-{os.environ["PBS_JOBID"]}'))

    template_runner.TemplateRunner(
        command=context.op_config["command"],
        version=context.op_config["semaphore_audience_version"],
        command_parameter=context.op_config["command_parameter"],
        workspace=workspace,
        skip_signal=True
    ).run(exit=False)

purify = configured(template_runner_op, name="purify")(IDONTKNOW)
compact_purify = configured(template_runner_op, name="compact_purify")(IDONTKNOW)

@job(
    name='template_runner_job',
    resource_defs={"io_manager": fs_io_manager},
    executor_def=dask_executor
)
def template_runner_job():
    purify()
    compact_purify()