Oban Pro

https://getoban.pro/docs/
Admin
Oban Pro is a premium job processing library for Elixir that provides advanced features like...
Tokens:93,670
Snippets:354
Trust Score:8.1
Update:1 year ago
Show doc for...
Tokens:
Raw
### Start Supervised Oban Instance

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing

Provides API documentation for `start_supervised_oban!`, a function to start a supervised Oban instance under the test supervisor. It accepts all valid Oban options and registers the instance with a unique reference, preventing conflicts.

```APIDOC
start_supervised_oban!(opts :: [Oban.option()]) :: Registry.key()

(since 0.11.0)

Starts a supervised Oban instance under the test supervisor, registering it with a unique reference. This prevents conflicts with other Oban instances or tests. It automatically adds sandbox allowances for plugins and queue producers, and the test process waits for clean shutdown.

**Running Jobs:** By default, the supervised instance does not process jobs (`stage_interval` is `:infinity`). Set `stage_interval` to a low value to process jobs normally.

**Options:** Accepts any option accepted by `Oban.start_link/1`, including shared repo-specific options.

**Examples:**

Start a basic supervised instance:
```elixir
name = start_supervised_oban!(repo: MyApp.Repo)
Oban.insert(name, MyWorker.new())
```

Start with a single queue and polling every 10ms:
```elixir
start_supervised_oban!(repo: MyApp.Repo, stage_interval: 10, queues: [alpha: 10])
```
```

--------------------------------

### Fetch Dependencies

Source: https://getoban.pro/docs/pro/1.6.2/installation

Fetches all project dependencies, including the newly added Oban Pro package, after updating `mix.exs`.

```elixir
$ mix deps.get
```

--------------------------------

### Add Oban Hex Repository

Source: https://getoban.pro/docs/pro/1.6.2/installation

Adds the private Oban Pro hex repository to your system. Requires your OBAN_LICENSE_KEY for authentication.

```elixir
mix hex.repo add oban https://repo.oban.pro \
  --fetch-public-key SHA256:4/OSKi0NRF91QVVXlGAhb/BIMLnK8NHcx/EWs+aIWPc \
  --auth-key $OBAN_LICENSE_KEY
```

--------------------------------

### Oban Pro Testing Setup

Source: https://getoban.pro/docs/pro/1.6.2/adoption

Demonstrates how to switch from Oban.Testing to Oban.Pro.Testing in an ExUnit test case. This change is necessary to leverage the Pro version's enhanced features for testing.

```Elixir
defmodule MyApp.Case do
  use ExUnit.CaseTemplate

  using do
    quote do
-      use Oban.Testing, repo: MyApp.Repo
+      use Oban.Pro.Testing, repo: MyApp.Repo
     end
   end
 end
```

--------------------------------

### Delegate Oban Pro Migrations

Source: https://getoban.pro/docs/pro/1.6.2/installation

Configures the generated migration file to delegate the up and down functions to Oban.Pro.Migration for proper database schema management.

```elixir
defmodule MyApp.Repo.Migrations.AddObanPro do
  use Ecto.Migration

  def up, do: Oban.Pro.Migration.up()

  def down, do: Oban.Pro.Migration.down()
end
```

--------------------------------

### Add Oban Pro Dependency

Source: https://getoban.pro/docs/pro/1.6.2/installation

Specifies Oban Pro as a dependency in your application's `mix.exs` file, pointing to the added 'oban' repository.

```elixir
{:oban_pro, "~> 1.6", repo: "oban"}
```

--------------------------------

### Generate Oban Pro Migration

Source: https://getoban.pro/docs/pro/1.6.2/installation

Generates a new Ecto migration file specifically for setting up Oban Pro's database tables and indexes.

```elixir
mix ecto.gen.migration add_oban_pro
```

--------------------------------

### Oban Pro Relay Usage Examples

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Relay

Demonstrates how to insert jobs for asynchronous execution using `async/1` and then await their results with `await/1` or `await_many/1`.

```elixir
relay = %{id: 123} |> MyApp.Worker.new() |> Oban.Pro.Relay.async()

{:ok, result} = Oban.Pro.Relay.await(relay)

{:ok, result} = Oban.Pro.Relay.await(relay, :timer.seconds(30))

relayed = 1..3 |> Enum.map(&DoubleWorker.new(%{int: &1})) |> Enum.map(&Oban.Pro.Relay.async/1) |> Oban.Pro.Relay.await_many()

#=> [{:ok, 2}, {:ok, 4}, {:ok, 6}]
```

--------------------------------

### Setup Oban Pro Testing Helpers (Individual Test)

Source: https://getoban.pro/docs/pro/1.6.2/testing

Imports `Oban.Pro.Testing` directly into an individual test module. This approach is useful if you prefer not to include helpers in every test case.

```Elixir
defmodule MyApp.WorkerTest do
  use MyApp.Case, async: true

  use Oban.Pro.Testing, repo: MyApp.Repo
end
```

--------------------------------

### Run Database Migrations

Source: https://getoban.pro/docs/pro/1.6.2/installation

Executes all pending database migrations, including the one generated for Oban Pro, to apply schema changes.

```elixir
mix ecto.migrate
```

--------------------------------

### DynamicScaler Configuration Examples

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler

Demonstrates how to configure the DynamicScaler plugin with custom cloud modules, either as a tuple with options or by referencing the module directly if it pulls configuration from the environment.

```elixir
{DynamicScaler, scalers: [range: 1..3, cloud: {MyApp.ASG, asg: "my-asg-name"}]}
```

```elixir
{DynamicScaler, scalers: [range: 1..3, cloud: MyApp.ASG]}
```

--------------------------------

### Setup Oban Pro Testing Helpers (Module Use)

Source: https://getoban.pro/docs/pro/1.6.2/testing

Imports `Oban.Pro.Testing` into a custom test case module, providing access to advanced testing functions. This is a convenient way to include helpers in all tests within a module.

```Elixir
defmodule MyApp.Case do
  use ExUnit.CaseTemplate

  using do
    quote do
      use Oban.Pro.Testing, repo: MyApp.Repo
    end
  end
end
```

--------------------------------

### Add DynamicPrioritizer Plugin

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPrioritizer

Demonstrates how to add the DynamicPrioritizer plugin to your Oban configuration in `config.exs`. This is the basic setup to enable the plugin's functionality.

```elixir
config :my_app, Oban,
  plugins: [Oban.Pro.Plugins.DynamicPrioritizer]
  ...
```

--------------------------------

### Dynamic Queue Configuration Examples

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicQueues

Illustrates how queue definitions in Oban's configuration affect runtime persistence. Options not explicitly defined in the configuration will be retained when queues are updated at runtime.

```elixir
queues: [
  default: [limit: 10],
  ...
]
```

```elixir
# pause, global_limit, etc. will persist, but limit won't
default: [limit: 10]
```

```elixir
# pause will persist, but limit and global_limit won't
default: [limit: 10, global_limit: 20]
```

```elixir
# neither limits nor pausing will persist
default: [limit: 10, global_limit: 20, paused: true]
```

--------------------------------

### Initial DynamicCron Crontab Setup

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicCron

Demonstrates how to provide an initial list of cron entries to the DynamicCron plugin using the `:crontab` option. This includes cron expressions, worker modules, and optional arguments like queue or max_attempts.

```elixir
plugins: [{
  Oban.Pro.Plugins.DynamicCron,
  crontab: [
    {"* * * * *", MyApp.MinuteJob},
    {"0 * * * *", MyApp.HourlyJob, queue: :scheduled},
    {"0 0 * * *", MyApp.DailyJob, max_attempts: 1},
    {"0 12 * * MON", MyApp.MondayWorker, tags: ["scheduled"]},
    {"@daily", MyApp.AnotherDailyWorker}
  ]
}]
```

--------------------------------

### Oban Pro Workflow Basic Usage Example

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Demonstrates creating a workflow with sequential and parallel job execution using Oban Pro's Workflow behaviour. It shows how to add jobs with dependencies and insert the workflow into Oban.

```Elixir
defmodule EchoWorkflow do
  use Oban.Pro.Decorator

  alias Oban.Pro.Workflow

  def insert do
    Workflow.new()
    |> Workflow.add(:a, new_echo(1))
    |> Workflow.add(:b, new_echo(2), deps: :a)
    |> Workflow.add(:c, new_echo(3), deps: :b)
    |> Workflow.add(:d, new_echo(4), deps: :b)
    |> Workflow.add(:e, new_echo(5), deps: [:c, :d])
    |> Oban.insert_all()
  end

  @job true
  def echo(value), do: IO.inspect(value)
end
```

--------------------------------

### Oban Pro Workflow Sub-Workflows Example

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Demonstrates nesting workflows by adding one workflow as a dependency of another using `add_workflow/4`. This allows for modular composition of complex pipelines.

```Elixir
alias MyApp.{WorkerA, WorkerB, WorkerC}
alias Oban.Pro.Workflow

extr_flow = 
  Workflow.new(workflow_name: "extract")
  |> Workflow.add(:extract, WorkerA.new(%{source: "database"}))
  |> Workflow.add(:transform, WorkerB.new(%{type: "normalize"}), deps: :extract)

note_flow = 
  Workflow.new(workflow_name: "notify")
  |> Workflow.add(:prepare, WorkerA.new(%{template: "report"}))
  |> Workflow.add(:send, WorkerB.new(%{method: "email"}), deps: :prepare)

# Example of adding a sub-workflow (assuming 'main_workflow' is defined elsewhere)
# main_workflow = 
#   Workflow.new(workflow_name: "main")
#   |> Workflow.add_workflow(:extract_stage, extr_flow, deps: :start_node)
#   |> Workflow.add_workflow(:notify_stage, note_flow, deps: :extract_stage)
#   |> Oban.insert_all()
```

--------------------------------

### Implement Oban Pro Batch Handler Callbacks

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Batch

Provides an example of implementing all optional batch handler callbacks: `batch_attempted/1`, `batch_cancelled/1`, `batch_completed/1`, `batch_discarded/1`, `batch_exhausted/1`, and `batch_retryable/1`. Each callback receives an Oban.Job struct and should return `:ok`.

```elixir
defmodule MyApp.BatchWorker do
  use Oban.Pro.Worker

  @behaviour Oban.Pro.Batch

  @impl Oban.Pro.Worker
  def process(_job), do: :ok

  @impl Oban.Pro.Batch
  def batch_attempted(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:attempted,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_cancelled(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:cancelled,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_completed(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:completed,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_discarded(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:discarded,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_exhausted(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:exhausted,  batch_id})
    :ok
  end

  @impl Oban.Pro.Batch
  def batch_retryable(%Job{meta: %{"batch_id" => batch_id}}) do
    IO.inspect({:retryable,  batch_id})
    :ok
  end
end
```

--------------------------------

### Oban Pro Chains: Sequential Job Execution

Source: https://getoban.pro/docs/pro/1.6.2/composition

Chains ensure jobs for a specific resource execute in order, processing events exactly as they arrive or implementing sequential state transitions. This example shows how to configure chaining by account ID and a worker that processes payment webhooks sequentially.

```elixir
use Oban.Pro.Worker, chain: [by: [:worker, args: :account_id]]

defmodule MyApp.PaymentWebhookWorker do
  use Oban.Pro.Worker, queue: :webhooks, chain: [by: [args: :account_id]]

  alias MyApp.{Accounts, Payments, Transactions}

  @impl Oban.Pro.Worker
  def process(%Job{args: %{"account_id" => account_id, "event" => event, "data" => data}}) do
    with {:ok, account} <- Accounts.get(account_id) do
      case event do
        "payment.succeeded" -> Transactions.record_payment(account, data)
        "payment.refunded" -> Transactions.record_refund(account, data)
        "subscription.renewed" -> Payments.renew_subscription(account, data)
      end
    end
  end
end
```

--------------------------------

### Configure Oban Pro Testing Helpers with Options

Source: https://getoban.pro/docs/pro/1.6.2/testing

Configures `Oban.Pro.Testing` with additional options like `repo`, `prefix`, and `log`. The `repo` option is mandatory as it's injected into many generated helpers.

```Elixir
using do
  quote do
    use Oban.Pro.Testing, repo: MyApp.Repo, prefix: "private", log: :debug
  end
end
```

--------------------------------

### Configure DynamicQueues Plugin

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicQueues

Adds the DynamicQueues plugin to Oban's configuration and defines initial queues. This setup persists queue definitions to the database and supervises them across all connected nodes.

```Elixir
config :my_app, Oban,
  plugins: [Oban.Pro.Plugins.DynamicQueues]
  ...
```

```Elixir
config :my_app, Oban,
  plugins: [{
    Oban.Pro.Plugins.DynamicQueues,
    queues: [
      default: 20,
      mailers: [global_limit: 20],
      events: [local_limit: 30, rate_limit: [allowed: 100, period: 60]]
    ]
  }]
  ...
```

--------------------------------

### Retrieve Recorded Job Results

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker

Fetch the recorded output of a job using the `fetch_recorded/1` function. This example shows how to retrieve the result or handle cases where no result is recorded.

```elixir
job = MyApp.Repo.get(Oban.Job, job_id)

case MyApp.RecordedWorker.fetch_recorded(job) do
  {:ok, result} ->
    # Use the result

  {:error, :missing} ->
    # Nothing recorded yet
end
```

--------------------------------

### Oban Pro Cloud Callback: init

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Cloud

Callback executed once at runtime to gather, normalize, and transform options for cloud hosting providers. It returns the initial configuration.

```APIDOC
@callback init(opts()) :: conf()

Executed once at runtime to gather, normalize, and transform options.
```

--------------------------------

### Define Recorded Worker with Custom Limit

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker

Configure a recorded worker with a specific output size limit. This example sets the limit to 128MB, overriding the default 64MB.

```elixir
use Oban.Pro.Worker, recorded: [limit: 128_000_000]
```

--------------------------------

### Run Basic Workflow

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing

Demonstrates how to create and run a basic Oban workflow with sequential job dependencies. The workflow is built using a pipe operator, adding jobs `:a`, `:b`, and `:c` with specified dependencies.

```Elixir
MyFlow.new_workflow()
|> MyFlow.add(:a, MyFlow.new(%{id: 1}))
|> MyFlow.add(:b, MyFlow.new(%{id: 2}), deps: [:a])
|> MyFlow.add(:c, MyFlow.new(%{id: 3}), deps: [:b])
|> run_workflow()
```

--------------------------------

### Drain Jobs API

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing

API documentation for the `drain_jobs` function, detailing its options for controlling queue draining behavior, concurrency, recursion, and result summarization. Includes usage examples for various scenarios.

```APIDOC
drain_jobs(opts \\ [])

  Drains jobs from the Oban queue.

  ## Options

  * `:queue` - an atom specifying the queue to drain, or `:all` to drain jobs across all queues at once. Defaults to `:all` when no queue is provided.
  * `:with_limit` — the maximum number of jobs to fetch for draining at once. The limit only impacts how many jobs are fetched at once, *not* concurrency. When recursion is enabled this is how many jobs are processed per-iteration, and it defaults to `1`. Otherwise, there isn't a limit and all available jobs are fetched.
  * `:with_recursion` — whether to draining jobs recursively, or all in a single pass. Either way, jobs are processed sequentially, one at a time. Recursion is required when jobs insert other jobs (e.g. batches), or depend on the execution of other jobs (e.g. workflows). Defaults to `true`.
  * `:with_safety` — whether to silently catch errors when draining. When `false`, raised exceptions or unhandled exits are reraised (unhandled exits are wrapped in [`Oban.CrashError`](https://hexdocs.pm/oban/2.19.2/Oban.CrashError.html)). Defaults to `false`.
  * `:with_scheduled` — whether to include scheduled or retryable jobs when draining. In recursive mode, which is the default, this will include snoozed jobs, and may lead to an infinite loop if the job snoozes repeatedly. Defaults to `true`.
  * `:with_summary` — whether to summarize execution results with a map of counts by state, or return a list of each job that was drained. Defaults to `true`, which returns a summary map.

  See [shared options](#module-shared-repo-options) for additional repo-specific options.

  ## Examples

  Drain all available jobs across all queues:

  ```elixir
  assert %{completed: 3} = drain_jobs(queue: :all)
  ```

  Drain and return all executed jobs without a count summary:

  ```elixir
  assert [%Oban.Job{}, %Oban.Job{}] = drain_jobs(with_summary: false)
  ```

  Drain including a job that you expect to raise:

  ```elixir
  assert_raise RuntimeError, fn -> drain_jobs(queue: :risky) end
  ```

  Drain without recursion to identify snoozed jobs:

  ```elixir
  assert %{scheduled: 3} = drain_jobs(with_recursion: false)
  ```

  Drain without staging scheduled jobs:

  ```elixir
  assert %{completed: 1, scheduled: 0} = drain_jobs(with_scheduled: false)
  ```

  Drain within a custom prefix:

  ```elixir
  assert %{completed: 3} = drain_jobs(queue: :default, prefix: "private")
  ```

  Drain using a specific repo (necessary when calling this function directly):

  ```elixir
  assert %{completed: 3} = drain_jobs(queue: :default, repo: MyApp.Repo)
  ```
```

--------------------------------

### Run Workflow with Job Matching

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing

Shows how to run a workflow and match on the returned jobs. It highlights that the execution order of jobs might differ from the insertion order, especially when dependencies are involved.

```Elixir
workflow = \
  MyFlow.new_workflow()
  |> MyFlow.add(:a, MyFlow.new(%{id: 1}))
  |> MyFlow.add(:b, MyFlow.new(%{id: 2}), deps: [:a])
  |> MyFlow.add(:c, MyFlow.new(%{id: 3}), deps: [:b])

[_job_a, _job_b, _job_c] = run_workflow(workflow, with_summary: false)
```

--------------------------------

### Get Specific Named Jobs

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Retrieve specific jobs from a workflow by their names or by providing a workflow ID. This allows targeted access to jobs within a workflow structure.

```Elixir
[:job_a, :job_b] = Workflow.all_jobs(job, names: [:a, :b])

jobs = Workflow.all_jobs("some-uuid-1234-5678")
```

--------------------------------

### Configure Oban Pro for Testing

Source: https://getoban.pro/docs/pro/1.6.2/testing

Sets Oban Pro to `:manual` testing mode in `test.exs`. This is crucial for testing Pro features like `Batch` and `Workflow` workers, especially when not using `:inline` mode.

```Elixir
# test.exs
config :my_app, Oban, testing: :manual
```

--------------------------------

### Define Typespec for Structured Worker

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker

Manually define a typespec for a structured worker to provide type hints and documentation. This example shows a basic worker with an ID and an active status.

```elixir
defmodule MyApp.StructuredWorker do
  use Oban.Pro.Worker

  @type t :: %__MODULE__{id: integer(), active: boolean()}

```

--------------------------------

### Oban Pro DynamicPartitioner API Types and Functions

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner

API documentation for Oban Pro's DynamicPartitioner, including type definitions for options and retention periods, and the signature and description for the `backfill_jobs/2` function.

```APIDOC
Types:

option() ::
  {:conf, Oban.Config.t()} |
  {:name, GenServer.name()} |
  {:retention, retention()} |
  {:schedule, String.t()} |
  {:timeout, timeout()} |
  {:timezone, String.t()}

retention() :: [
  completed: pos_integer(),
  cancelled: pos_integer(),
  discarded: pos_integer()
]

Functions:

backfill_jobs(conf_or_name, opts \ [])
  Backfill jobs from a standard table into a newly partitioned table.
  Parameters:
    conf_or_name: Configuration or name of the Oban instance.
    opts: Options for backfilling, including `:new_prefix`, `:old_prefix`, `:states`, `:batch_size`, `:sleep_ms`.
```

--------------------------------

### Build and Insert Basic Workflow

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Demonstrates creating a new workflow, adding sequential jobs with dependencies, and inserting the entire workflow into Oban. It shows a simple linear execution flow.

```elixir
alias MyApp.WorkerA
alias MyApp.WorkerB
alias MyApp.WorkerC

Workflow.new()
|> Workflow.add(:setup, WorkerC.new(%{mode: "initialize"}))
|> Workflow.add_workflow(:extract, extr_flow, deps: :setup)
|> Workflow.add_workflow(:notify, note_flow, deps: :extract)
|> Workflow.add(:finalize, WorkerC.new(%{mode: "cleanup"}), deps: :notify)
|> Oban.insert_all()
```

--------------------------------

### Oban Pro Batch Worker Example

Source: https://getoban.pro/docs/pro/1.6.2/composition

Demonstrates how to define an Oban Pro Batch worker in Elixir, including processing individual jobs and defining callbacks for batch completion or failure.

```elixir
defmodule MyApp.ImageProcessor do
  use Oban.Pro.Worker, queue: :media

  @behaviour Oban.Pro.Batch

  alias MyApp.{Media, Notifications}

  @impl Oban.Pro.Worker
  def process(%Job{args: %{"image_id" => id, "operations" => operations}}) do
    with {:ok, image} <- Media.get_image(id),
      Media.process(image, operations)
    end
  end

  @impl Oban.Pro.Batch
  def batch_completed(%Job{meta: %{"batch_id" => batch_id, "account_id" => account_id}}) do
    with {:ok, images} <- Media.get_batch_images(batch_id),
         {:ok, archive_path} <- Media.create_archive(images),
         {:ok, url} <- Media.store_archive(archive_path, account_id) do
      Notifications.image_processing_complete(account_id, url)
    end
  end

  @impl Oban.Pro.Batch
  def batch_discarded(%Job{meta: %{"batch_id" => batch_id, "account_id" => account_id}}) do
    Notifications.processing_failed(account_id, batch_id)
  end
end
```

--------------------------------

### Update SmartEngine Namespace

Source: https://getoban.pro/docs/pro/1.6.2/v1-0

The SmartEngine has been moved to an `Engines` namespace for consistency with Oban's engines. Update your configuration to reference the new path.

```Elixir
- engine: Oban.Pro.Queue.SmartEngine
+ engine: Oban.Pro.Engines.Smart
```

--------------------------------

### Get All Workflow Jobs

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Retrieves all jobs associated with a workflow, with options to filter by upstream dependencies, specific names, or include sub-workflow jobs. Can be used with a job object or a workflow ID.

```Elixir
def process(job) do
  job
  |> Workflow.all_jobs()
  |> do_things_with_jobs()

  :ok
end
```

```Elixir
jobs = Workflow.all_jobs(job, only_deps: true)
```

```Elixir
[job_a, job_b] = Workflow.all_jobs(job, names: [:a, :b])
```

```Elixir
[sub_a, sub_b] = Workflow.all_jobs(job, names: [:sub], with_subs: true)
```

```Elixir
jobs = Workflow.all_jobs("some-uuid-1234-5678")
```

```Elixir
[job_a, job_b] = Workflow.all_jobs("some-uuid-1234-5678", deps: [:a, :b])
```

```Elixir
jobs = Workflow.all_jobs(MyApp.Oban, "some-uuid-1234-5678")
```

--------------------------------

### Configure DynamicPartitioner Schedule and Buffer

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner

Configuration snippet to customize the `DynamicPartitioner` plugin's behavior. Allows setting a custom buffer period for pre-creating partitions and defining a specific schedule and timezone for partition management.

```Config
plugins: [{
  Oban.Pro.Plugins.DynamicPartitioner,
  buffer: 3,
  schedule: "0 5 * * *",
  timezone: "Europe/Paris"
}]
```

--------------------------------

### Force Job Cancellation on Deadline Expiry

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker

Shows how to configure a worker to forcefully cancel itself if it starts processing but cannot complete before its deadline. This is achieved by setting the `force: true` option within the `deadline` configuration.

```Elixir
defmodule FirmDeadlinedWorker do
  use Oban.Pro.Worker, deadline: [in: {10, :minutes}, force: true]

  # ... worker implementation ...
end
```

--------------------------------

### Update Batch Creation from new_batch to Oban.Pro.Batch.new

Source: https://getoban.pro/docs/pro/1.6.2/v1-5

Illustrates the change in how new batches are created. The old `MyApp.BatchWorker.new_batch(jobs)` syntax is replaced by the more direct `Oban.Pro.Batch.new(jobs)` call, simplifying batch instantiation.

```elixir
-MyApp.BatchWorker.new_batch(jobs)
+Oban.Pro.Batch.new(jobs)
```

--------------------------------

### Scaler Configuration: Cooldown Period

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler

Sets a custom cooldown period between scaling events. This example configures a 5-minute (300 seconds) cooldown to prevent rapid, successive scaling actions.

```Elixir
scalers: [range: 1..3, cloud: MyApp.Cloud, cooldown: 300]
```

--------------------------------

### Partitioning Schemes

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart

Demonstrates different ways to define partition schemes for Oban Pro jobs. Partitions can be based on worker identity, arguments, or a combination thereof, helping to manage concurrency and distribute workloads effectively.

```ruby
# Partition by worker alone
partition: :worker

# Partition by the `id` and `account_id` from args, ignoring the worker
partition: [args: [:id, :account_id]]

# Partition by worker and the `account_id` key from args
partition: [:worker, args: :account_id]
```

--------------------------------

### Oban Smart Engine Snooze Handling

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart

Provides an example of how to check a job's meta for snooze information within the `process/1` function. This helps differentiate between actual attempts and snoozes.

```Elixir
def process(job) do
  case job.meta do
    %{"orig_scheduled_at" => unix_microseconds, "snoozed" => snoozed} ->
      IO.inspect({snoozed, unix_microseconds}, label: "Times snoozed since")

    _ ->
      # This job has never snoozed before
  end
end
```

--------------------------------

### Get Workflow Status

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Retrieve detailed status information for a workflow, including its overall state, job counts, execution duration, and sub-workflow details. The output is a map containing comprehensive metrics.

```Elixir
status = Workflow.status(job)
```

--------------------------------

### Authorize Oban Pro Hex Repo in CI/CD

Source: https://getoban.pro/docs/pro/1.6.2/ci_cd

This snippet demonstrates how to add the Oban Pro Hex repository to your CI/CD environment. It uses the `mix hex.repo add` command to register the repository with its public key and an authentication key, typically stored as a secret in the CI service. This ensures secure access to Oban Pro packages during builds.

```shell
- name: Authorize Oban
  run: |
    mix hex.repo add oban https://repo.oban.pro \
      --fetch-public-key SHA256:4/OSKi0NRF91QVVXlGAhb/BIMLnK8NHcx/EWs+aIWPc \
      --auth-key ${{secrets.oban_license_key}}
```

--------------------------------

### Generate Flowchart with Mermaid

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

This example demonstrates how to generate a flowchart representation of a workflow using the `to_mermaid` function from the Oban Pro library. It takes a workflow object as input and outputs a string formatted for Mermaid.js.

```Elixir
Workflow.to_mermaid(workflow)
```

--------------------------------

### Run Ecto Migration

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner

Executes the pending Ecto migrations, including the one created for the partitioned oban_jobs table.

```bash
mix ecto.migrate
```

--------------------------------

### Basic Oban Pro Migration Implementation

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Migration

Shows how to implement the generated migration by delegating the `up/0` and `down/0` functions to Oban.Pro.Migration. This applies all necessary migrations for the database.

```elixir
defmodule MyApp.Repo.Migrations.AddObanPro do
  use Ecto.Migration

  def up, do: Oban.Pro.Migration.up()

  def down, do: Oban.Pro.Migration.down()
end
```

--------------------------------

### Implement Workflow Cancellation Callback

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Provides an example of implementing the `after_cancelled/2` callback for workflows. This callback is invoked when a workflow job is cancelled due to upstream dependency changes, allowing for custom cleanup logic.

```elixir
defmodule MyApp.Workflow do
  use Oban.Pro.Worker

  @behaviour Oban.Pro.Workflow

  require Logger

  @impl Oban.Pro.Workflow
  def after_cancelled(reason, job) do
    Logger.warn("Workflow job #{job.id} cancelled because a dependency was #{reason}")

    # Perform any cleanup required
    notify_monitoring_system(job, reason)

    :ok
  end
end
```

--------------------------------

### Elixir Cloud Module for AWS EC2 Scaling

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler

Example implementation of an Oban.Pro.Cloud behaviour for scaling AWS EC2 Auto Scaling Groups. It uses the ex_aws package to interact with the AWS API.

```elixir
defmodule MyApp.ASG do
  @behaviour Oban.Pro.Cloud

  @impl Oban.Pro.Cloud
  def init(opts), do: Map.new(opts)

  @impl Oban.Pro.Cloud
  def scale(desired, conf) do
    params = %{
      "Action" => "SetDesiredCapacity",
      "AutoScalingGroupName" => conf.asg,
      "DesiredCapacity" => desired,
      "Version" => "2011-01-01"
    }

    query = %ExAws.Operation.Query{
      path: "",
      params: params,
      service: :autoscaling,
      action: :set_desired_capacity
    }

    with {:ok, _} <- ExAws.request(query), do: {:ok, conf}
  end
end
```

--------------------------------

### run_workflow: Oban Pro Test Helper

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Testing

Inserts and executes a workflow synchronously within the test process. This helper augments the workflow with options optimized for testing but respects all standard workflow options. It accepts options for drain_jobs/1 and repo-specific options.

```APIDOC
run_workflow(workflow, opts \ [])

(since 0.11.0)

@spec run_workflow(
  Oban.Pro.Workflow.t(),
  Oban.Pro.Workflow.new_opts() | drain_option()
) ::
  drain_result()

Insert and execute a workflow synchronously, within the test process.

This helper augments the workflow with options optimized for testing, but it will still respect
all standard workflow options.

## Options

Accepts all options for [`drain_jobs/1`](#drain_jobs/1), including the repo-specific options listed in
[shared-options](#module-shared-repo-options).

```

--------------------------------

### Generate Partitioned Oban Jobs Migration

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner

Generates a new Ecto migration file to create a partitioned oban_jobs table. This is the first step before configuring the DynamicPartitioner plugin.

```bash
mix ecto.gen.migration add_partitioned_oban_jobs
```

--------------------------------

### Put and Get Workflow Context

Source: https://getoban.pro/docs/pro/1.6.2/changelog

Illustrates using `Workflow.put_context/3` to share common data across jobs in a workflow and `Workflow.get_context/1` within a worker to access this shared data.

```elixir
workflow = 
  Workflow.new()
  |> Workflow.put_context(%{user_id: 123, app_version: "1.2.3"})
  |> Workflow.add(:job_a, WorkerA.new(%{}))
  |> Workflow.add(:job_b, WorkerB.new(%{}))

# Later in a worker:
def process(job) do
  context = Workflow.get_context(job)
  # Use context map...
end
```

--------------------------------

### Define before_new Hook to Augment Job Arguments

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Worker

Shows how to define a before_new/2 hook to modify or augment the arguments and options used when creating new jobs. This example injects a span_id into the job's meta.

```elixir
defmodule MyApp.SpanHook do
  def before_new(args, opts) do
    opts = 
      opts
      |> Keyword.put_new(:meta, %{})
      |> put_in([:meta, :span_id], MyApp.Telemetry.gen_span_id())

    {:ok, args, opts}
  end
end
```

--------------------------------

### Rate Limit with Partitioning

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Engines.Smart

Shows how to configure rate limits that are applied per partition. This allows for granular control over how frequently jobs within specific partitions can be executed within a given time period.

```ruby
# Allow one job per worker, every ten seconds, across the cluster
local_limit: 10, rate_limit: [allowed: 1, period: 10, partition: :worker]
```

--------------------------------

### Basic Scaler Configuration

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicScaler

Shows a fundamental configuration for a DynamicScaler, specifying the node range and the cloud integration module. This sets up the minimum and maximum number of nodes and how to interact with the cloud provider.

```Elixir
{DynamicScaler, scalers: [range: 1..5, cloud: MyApp.Cloud]}
```

--------------------------------

### Get All Recorded Workflow Job Outputs

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Workflow

Retrieves recorded outputs for workflow jobs, optionally filtered by name or relationship. Returns a map where keys are job names and values are their recorded outputs, or nil if unrecorded.

```Elixir
def process(job) do
  %{"a" => job_a_return, "b" => job_b_return} = Workflow.all_recorded(job)

  :ok
end
```

```Elixir
recorded = Workflow.all_recorded(job, only_deps: true)
```

```Elixir
%{"a" => _, "b" => _} = Workflow.all_recorded(job, names: [:a, :b])
```

```Elixir
%{"a" => _, "b" => _} = Workflow.all_recorded("some-uuid-1234-5678", names: [:a, :b])
```

--------------------------------

### Configure Two Oban Instances for Transition

Source: https://getoban.pro/docs/pro/1.6.2/Oban.Pro.Plugins.DynamicPartitioner

Sets up two separate Oban instances during a migration: one for the original table and one for the new partitioned table. This allows processing of existing jobs while new jobs use the partitioned table.

```elixir
config :my_app, Oban.Original,
  queues: [
    default: 10,
    other_queue: 10,
    and_another: 10
  ]

config :my_app, Oban,
  prefix: "partitioned",
  queues: [
    default: 10,
    other_queue: 10,
    and_another: 10
  ],
  plugins: [
    ...
  ]
```

--------------------------------

### Oban Pro Batch Creation and Insertion

Source: https://getoban.pro/docs/pro/1.6.2/composition

Shows how to create a new Oban Pro Batch and insert multiple jobs into it using Elixir. This includes setting callback options and inserting the batch into Oban.

```elixir
alias Oban.Pro.Batch
alias MyApp.ImageProcessor

image_ids
|> Enum.map(&ImageProcessor.new(%{image_id: &1,  operations: ~w(resize optimize watermark)}))
|> Batch.new(callback_opts: [args: %{account_id: current_user.id}])
|> Oban.insert_all()
```
Tokens:
Raw