# `Gust.Flows` [🔗](https://github.com/marciok/gust/blob/v0.1.40/lib/gust/flows.ex#L1) The Flows context. This module serves as the boundary for accessing and manipulating Flow-related data, such as DAGs, Runs, Tasks, and Secrets. # `attach_task_error_reference` Adds an external reporter reference to a failed task's error. A non-failed task is treated as stale so a delayed reporter response cannot overwrite the state of a task that has moved on. # `cancel_waiting_task` Cancels a waiting task by clearing its wait markers and marking it failed in one write. # `change_secret` Returns an `%Ecto.Changeset{}` for tracking secret changes. ## Examples iex> change_secret(secret) %Ecto.Changeset{data: %Secret{}} # `count_runs_on_dag` Returns the number of runs associated with a given DAG. ## Parameters * `dag_id` - The identifier of the DAG whose runs should be counted. * `status` - Optional status to filter by. * `params_search` - Optional case-insensitive parameter key or value search. ## Returns * The integer count of runs associated with the specified DAG. # `create_dag` Creates a DAG. ## Examples iex> create_dag(%{field: value}) {:ok, %Dag{}} iex> create_dag(%{field: bad_value}) {:error, %Ecto.Changeset{}} # `create_log` Creates a log. # `create_run` Creates a run. ## Examples iex> create_run(%{field: value}) {:ok, %Run{}} iex> create_run(%{field: bad_value}) {:error, %Ecto.Changeset{}} # `create_secret` Creates a secret. ## Examples iex> create_secret(%{field: value}) {:ok, %Secret{}} iex> create_secret(%{field: bad_value}) {:error, %Ecto.Changeset{}} # `create_task` Creates a task. # `create_test_run` Creates a test run. # `create_test_task` Creates a test task. # `delete_dag!` Deletes a DAG. # `delete_not_found_ids` Deletes all DAGs whose IDs are not in the provided list. # `delete_run` Deletes the given run from the database. The `run` must be a persisted `%Run{}` struct. This function delegates to `Repo.delete/1` and returns `{:ok, %Run{}}` if the run is successfully deleted, or `{:error, %Ecto.Changeset{}}` if the delete operation fails. ## Examples iex> delete_run(run) {:ok, %Run{}} iex> delete_run(invalid_run) {:error, %Ecto.Changeset{}} # `delete_runs_on_dag` Deletes runs that belong to the given DAG. Returns the deleted run structs so callers can update LiveView streams. # `delete_secret` Deletes a secret. # `delete_task!` # `get_dag!` Gets a single DAG. Raises `Ecto.NoResultsError` if the Dag does not exist. # `get_dag_by_name` Gets a DAG by name. # `get_dag_by_name_with_runs!` Gets a DAG by name with its runs preloaded, with pagination for runs. The DAG is looked up by its `name`. The associated runs are ordered by descending `inserted_at` and paginated using the required `:limit` and `:offset` keyword options. ## Parameters * `name` - The name of the DAG to retrieve. ## Options * `:limit` - Required. The maximum number of runs to preload. * `:offset` - Required. The number of runs to skip before starting to preload. * `:status` - Optional. Filters preloaded runs by status when present. * `:params_search` - Optional. Case-insensitively filters runs by parameter keys or values. ## Returns Returns the `%Dag{}` struct with its `:runs` association preloaded according to the given pagination options. Raises `Ecto.NoResultsError` if no DAG with the given name exists. Raises `KeyError` if `:limit` or `:offset` is missing from `opts`. # `get_dag_with_recent_runs` Gets a DAG by name with its most recent runs preloaded. # `get_dag_with_runs!` Gets a DAG with its runs preloaded. # `get_dag_with_runs_and_tasks!` Gets a DAG with its runs and tasks preloaded, with pagination for runs. Tasks are loaded with only the fields required by the run-history grid. Use the task or run detail functions when payload fields are needed. The `:limit` option is required. Pass `:offset` for normal pagination or `:pinned_run_id` to load a window containing the pinned followed by older runs. The pinned must belong to the named DAG. # `get_dags_with_recent_runs` Gets the requested DAGs with their most recent runs preloaded. Runs are ordered from oldest to newest so they can be displayed as a chronological history. A lateral join limits the runs per DAG while keeping the operation to one database query. # `get_log!` Gets a single log. Raises `Ecto.NoResultsError` if the Log does not exist. # `get_logs` Gets logs for a task, optionally filtered by level. Logs are ordered by their insertion timestamp in ascending order. # `get_run!` Gets a single run. Raises `Ecto.NoResultsError` if the Run does not exist. ## Examples iex> get_run!(123) %Run{} iex> get_run!(456) ** (Ecto.NoResultsError) # `get_run_with_tasks!` Gets a single run with its tasks preloaded. # `get_running_runs_by_dag` Gets runs for a list of DAG IDs with the given statuses. ## Parameters - `dag_ids`: List of DAG IDs to filter runs by. - `statuses`: List of statuses (atoms) to filter runs by (e.g., [:running, :queued]). ## Examples iex> get_running_runs_by_dag([1, 2, 3], [:running, :retrying]) [%Run{}, ...] # `get_runs_on_dag` Gets runs by ID that belong to the given DAG. # `get_secret!` Gets a single secret. Raises `Ecto.NoResultsError` if the Secret does not exist. # `get_secret_by_name` Gets a single secret by name. # `get_secret_by_name!` Gets a single secret by name. Raises `Ecto.NoResultsError` if the Secret does not exist. # `get_task` # `get_task!` Gets a single task. Raises `Ecto.NoResultsError` if the Task does not exist. # `get_task_by_name` Gets a task by name, run ID, and map index. # `get_task_by_name_run` Gets a task by name and run ID. # `get_task_by_name_run!` Gets a task by name and run ID. Raises `Ecto.NoResultsError` if the Task does not exist. # `get_task_by_name_run_with_logs` Gets a task by name and run ID, with logs preloaded. Accepts an optional `log_level` argument: * When `log_level` is `nil` (the default), all logs for the task are preloaded. * When `log_level` is provided, only logs with the matching level are preloaded. Logs are ordered by their `inserted_at` timestamp in ascending order. # `get_task_result_by_name_run!` Gets the result of a task by name and run ID. The result is an empty map when the task did not save one. Raises `Ecto.NoResultsError` if the Task does not exist. # `get_task_statuses` Gets task statuses for the given task IDs. # `get_task_statuses_by_name` Gets all task statuses by name and run ID. # `get_task_with_logs!` Gets a task with its logs preloaded. # `get_tasks_by_name` Gets all task instances by name and run ID. # `list_dags` Lists all DAGs. # `list_secrets` Lists all secrets. # `prepare_task_restart` Clears execution state before manually restarting a terminal task instance. Mapping identity and persisted params are intentionally preserved so a mapped task can be restarted without rebuilding or collapsing its sibling instances. # `reconcile_run_tasks` # `resume_waiting_task` Updates a resumed waiting task's params, wait state, and status in one write. # `schedule_task_retry` Marks a task as retrying and persists when its next attempt starts. # `toggle_enabled` Toggles the enabled status of a DAG. # `update_run_status` Updates a run status. # `update_runs_status` Updates the status of multiple runs in one database statement. # `update_secret` Updates a secret. # `update_task_attempt` Updates a task attempt count. # `update_task_error` Updates a task error. # `update_task_mapping` Updates a task map index and its persisted parameters. # `update_task_params` Updates a task's persisted parameters. # `update_task_result` Updates a task result. # `update_task_result_error` Updates a task result and error. # `update_task_status` Updates a task status. # `update_task_wait_state` Updates a task's persisted wait state. # `update_tasks_status` Updates the status of multiple tasks in one database statement. Mirrors the `retry_at` clearing behavior of `update_task_status/2`: any status other than `:retrying` clears a previously scheduled retry. --- *Consult [api-reference.md](api-reference.md) for complete listing*