defmodule Gust.DSL do @moduledoc """ The Gust DSL is how you turn a module into a DAG. When you add `use Gust.DSL` to a module in the `dags/` folder, Gust automatically detects it and creates a DAG based on the file name. You can configure a schedule, define callbacks, and in the `dev` environment the code is automatically reloaded when files change. After enabling the DSL, use `task` definitions to declare the steps that should be executed. ## Example defmodule HelloWorld do # `schedule` and `on_finished_callback` are optional. # Note: if you change `schedule`, restart the server to update the cron job. use Gust.DSL, schedule: "* * * * *", on_finished_callback: :notify_something # Gust logs are stored and displayed through GustWeb via Logger. require Logger # Gust.Flows is used to query Dag, Run, and Task. alias Gust.Flows def notify_something(status, run) do dag = Flows.get_dag!(run.dag_id) message = "DAG: \#{dag.name}; completed with status: \#{status}" Logger.info(message) end task :first_task, downstream: [:second_task], store_result: true do greetings = "Hi from first_task" Logger.info(greetings) # The return value must be a map when `store_result` is true. %{result: greetings} end task :second_task, ctx: %{run_id: run_id} do task = Flows.get_task_by_name_run("first_task", run_id) Logger.info(task.result) end end ## Parameters * `schedule` - A valid cron expression string. * `on_finished_callback` - The name of the function to be called. """ defmacro __using__(dag_options) do quote do import unquote(__MODULE__), only: [task: 2, task: 3] Module.register_attribute(__MODULE__, :dag_tasks, accumulate: true) def __dag_options__, do: unquote(dag_options) @before_compile unquote(__MODULE__) end end defmacro __before_compile__(_env) do quote do def __dag_tasks__, do: @dag_tasks end end @doc """ Defines a task in the DAG. ## Parameters * `name` — The name of the task (atom). * `opts_and_ctx` — A keyword list of options and an optional context pattern. * `block` — The code block executed when the task runs. ## Task Options * `:downstream` — A list of task names (atoms) to run after this task completes. * `:store_result` — When true, the task's return value will be persisted. * Note: If enabled, the return value **must be a map**. * `:ctx` — A pattern that will be matched against the context passed to the task. * Defaults to: `%{run_id: run_id}`. ## Example task :my_task, ctx: %{run_id: run_id} do IO.inspect(run_id) end task :first, downstream: [:second] do :ok end task :persist_result, store_result: true do %{result: :ok} end When using `store_result: true`, the return value **must** be a map so it can be merged into the overall DAG results. """ defmacro task(name, opts_and_ctx, do: block) do {ctx_pattern, opts} = Keyword.pop(opts_and_ctx, :ctx) ctx_pattern = ctx_pattern || quote do: %{run_id: run_id} quote do @dag_tasks {unquote(name), unquote(opts)} def unquote(name)(ctx) do unquote(ctx_pattern) = ctx unquote(block) end end end @doc """ Defines a task in the DAG without options or explicit context matching. ## Parameters * `name` - The name of the task (atom). * `block` - The code block to execute for the task. ## Example task :simple_task do IO.puts "Hello" end """ defmacro task(name, do: block) do quote do @dag_tasks {unquote(name), []} def unquote(name)(ctx) do unquote(block) end end end end