Merge pull request #413 from Shopify/tasks-with-params-active-model

Support Tasks with custom parameters

README.md

@@ -152,6 +152,41 @@ Tasks can define multiple throttle conditions. Throttle conditions are inherited
 by descendants, and new conditions will be appended without impacting existing
 conditions.
 
+### Custom Task Parameters
+
+Tasks may need additional information, supplied via parameters, to run.
+Parameters can be defined as Active Model Attributes in a Task, and then
+become accessible to any of Task's methods: `#collection`, `#count`, or
+`#process`.
+
+```ruby
+# app/tasks/maintenance/update_posts_via_params_task.rb
+module Maintenance
+  class UpdatePostsViaParamsTask < MaintenanceTasks::Task
+    attribute :updated_content, :string
+    validates :updated_content, presence: true
+
+    def collection
+      Post.all
+    end
+
+    def count
+      collection.count
+    end
+
+    def process(post)
+      post.update!(content: updated_content)
+    end
+  end
+end
+```
+
+Tasks can leverage Active Model Validations when defining parameters. Arguments
+supplied to a Task accepting parameters will be validated before the Task starts
+to run. Since arguments are specified in the user interface via text area
+inputs, it's important to check that they conform to the format your Task
+expects, and to sanitize any inputs if necessary.
+
 ### Considerations when writing Tasks
 
 MaintenanceTasks relies on the queue adapter configured for your application to

app/controllers/maintenance_tasks/tasks_controller.rb

@@ -27,17 +27,27 @@ def show
     def run
       task = Runner.run(
         name: params.fetch(:id),
-        csv_file: params[:csv_file]
+        csv_file: params[:csv_file],
+        arguments: task_arguments,
       )
       redirect_to(task_path(task))
     rescue ActiveRecord::RecordInvalid => error
       redirect_to(task_path(error.record.task_name), alert: error.message)
+    rescue ActiveRecord::ValueTooLong => error
+      task_name = params.fetch(:id)
+      redirect_to(task_path(task_name), alert: error.message)
     rescue Runner::EnqueuingError => error
       redirect_to(task_path(error.run.task_name), alert: error.message)
     end
 
     private
 
+    def task_arguments
+      return {} unless params[:task_arguments].present?
+      task_attributes = Task.named(params[:id]).attribute_names
+      params.require(:task_arguments).permit(*task_attributes).to_h
+    end
+
     def set_refresh
       @refresh = 3
     end

app/jobs/concerns/maintenance_tasks/task_job_concern.rb

@@ -71,7 +71,7 @@ def task_iteration(input)
 
     def before_perform
       @run = arguments.first
-      @task = Task.named(@run.task_name).new
+      @task = @run.task
       if @task.respond_to?(:csv_content=)
         @task.csv_content = @run.csv_file.download
       end

app/models/maintenance_tasks/run.rb

@@ -34,10 +34,12 @@ class Run < ApplicationRecord
       Task.available_tasks.map(&:to_s)
     } }
     validate :csv_attachment_presence, on: :create
+    validate :validate_task_arguments, on: :create
 
     attr_readonly :task_name
 
     serialize :backtrace
+    serialize :arguments, JSON
 
     scope :active, -> { where(status: ACTIVE_STATUSES) }
 
@@ -206,6 +208,40 @@ def csv_attachment_presence
       nil
     end
 
+    # Support iterating over ActiveModel::Errors in Rails 6.0 and Rails 6.1+.
+    # To be removed when Rails 6.0 is no longer supported.
+    if Rails::VERSION::STRING.match?(/^6.0/)
+      # Performs validation on the arguments to use for the Task. If the Task is
+      # invalid, the errors are added to the Run.
+      def validate_task_arguments
+        if task.invalid?
+          error_messages = task.errors
+            .map { |attribute, message| "#{attribute.inspect} #{message}" }
+          errors.add(
+            :arguments,
+            "are invalid: #{error_messages.join("; ")}"
+          )
+        end
+      rescue Task::NotFoundError
+        nil
+      end
+    else
+      # Performs validation on the arguments to use for the Task. If the Task is
+      # invalid, the errors are added to the Run.
+      def validate_task_arguments
+        if task.invalid?
+          error_messages = task.errors
+            .map { |error| "#{error.attribute.inspect} #{error.message}" }
+          errors.add(
+            :arguments,
+            "are invalid: #{error_messages.join("; ")}"
+          )
+        end
+      rescue Task::NotFoundError
+        nil
+      end
+    end
+
     # Fetches the attached ActiveStorage CSV file for the run. Checks first
     # whether the ActiveStorage::Attachment table exists so that we are
     # compatible with apps that are not using ActiveStorage.
@@ -216,5 +252,21 @@ def csv_file
       return unless ActiveStorage::Attachment.table_exists?
       super
     end
+
+    # Returns a Task instance for this Run. Assigns any attributes to the Task
+    # based on the Run's parameters. Note that the Task instance is not supplied
+    # with :csv_content yet if it's a CSV Task. This is done in the job, since
+    # downloading the CSV file can take some time.
+    #
+    # @return [Task] a Task instance.
+    def task
+      @task ||= begin
+        task = Task.named(task_name).new
+        if task.attribute_names.any? && arguments.present?
+          task.assign_attributes(arguments)
+        end
+        task
+      end
+    end
   end
 end

app/models/maintenance_tasks/runner.rb

@@ -37,14 +37,19 @@ def initialize(run)
     #   for the Task to iterate over when running, in the form of an attachable
     #   (see https://edgeapi.rubyonrails.org/classes/ActiveStorage/Attached/One.html#method-i-attach).
     #   Value is nil if the Task does not use CSV iteration.
+    # @param arguments [Hash] the arguments to persist to the Run and to make
+    #   accessible to the Task.
     #
     # @return [Task] the Task that was run.
     #
     # @raise [EnqueuingError] if an error occurs while enqueuing the Run.
     # @raise [ActiveRecord::RecordInvalid] if validation errors occur while
     #   creating the Run.
-    def run(name:, csv_file: nil)
-      run = Run.active.find_by(task_name: name) || Run.new(task_name: name)
+    # @raise [ActiveRecord::ValueTooLong] if the creation of the Run fails due
+    #   to a value being too long for the column type.
+    def run(name:, csv_file: nil, arguments: {})
+      run = Run.active.find_by(task_name: name) ||
+        Run.new(task_name: name, arguments: arguments)
       run.csv_file.attach(csv_file) if csv_file
 
       run.enqueued!

app/models/maintenance_tasks/task_data.rb

@@ -132,6 +132,15 @@ def csv_task?
       !deleted? && Task.named(name) < CsvCollection
     end
 
+    # @return [Array<String>] the names of parameters the Task accepts.
+    def parameter_names
+      if deleted?
+        []
+      else
+        Task.named(name).attribute_names
+      end
+    end
+
     private
 
     def runs

app/tasks/maintenance_tasks/task.rb

@@ -3,6 +3,9 @@ module MaintenanceTasks
   # Base class that is inherited by the host application's task classes.
   class Task
     extend ActiveSupport::DescendantsTracker
+    include ActiveModel::Attributes
+    include ActiveModel::AttributeAssignment
+    include ActiveModel::Validations
 
     class NotFoundError < NameError; end
 

app/views/maintenance_tasks/tasks/show.html.erb

@@ -16,6 +16,16 @@
           <%= form.file_field :csv_file %>
         </div>
       <% end %>
+      <% if @task.parameter_names.any? %>
+        <div class="block">
+          <%= form.fields_for :task_arguments do |ff| %>
+            <% @task.parameter_names.each do |parameter| %>
+              <%= ff.label parameter, "#{parameter}: ", class: "label" %>
+              <%= ff.text_area parameter, class: "textarea" %>
+            <% end %>
+          <% end %>
+        </div>
+      <% end %>
       <div class="block">
         <%= form.submit 'Run', class: "button is-success", disabled: @task.deleted? %>
       </div>

db/migrate/20210517131953_add_arguments_to_maintenance_tasks_runs.rb

@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+class AddArgumentsToMaintenanceTasksRuns < ActiveRecord::Migration[6.0]
+  def change
+    add_column(:maintenance_tasks_runs, :arguments, :text)
+  end
+end

test/application_system_test_case.rb

@@ -19,12 +19,10 @@ class ApplicationSystemTestCase < ActionDispatch::SystemTestCase
 
   setup do
     travel_to Time.zone.local(2020, 1, 9, 9, 41, 44)
-    Maintenance::UpdatePostsTask.fast_task = false
   end
 
   teardown do
     assert_empty page.driver.browser.manage.logs.get(:browser)
-    Maintenance::UpdatePostsTask.fast_task = true
     FileUtils.rm_rf("test/dummy/tmp/downloads")
   end
 end

test/dummy/app/tasks/maintenance/params_task.rb

@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+module Maintenance
+  class ParamsTask < MaintenanceTasks::Task
+    attribute :post_ids, :string
+
+    validates :post_ids,
+      presence: true,
+      format: { with: /\A(\s?\d+(,\s?\d+\s?)*)\z/, allow_blank: true }
+
+    class << self
+      attr_accessor :fast_task
+    end
+
+    def collection
+      Post.where(id: post_ids_array)
+    end
+
+    def count
+      collection.count
+    end
+
+    def process(post)
+      sleep(1) unless self.class.fast_task
+
+      post.update!(content: "New content added on #{Time.now.utc}")
+    end
+
+    private
+
+    def post_ids_array
+      post_ids.split(",")
+    end
+  end
+end

test/dummy/db/schema.rb

@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 2021_02_25_152418) do
+ActiveRecord::Schema.define(version: 2021_05_17_131953) do
 
   create_table "active_storage_attachments", force: :cascade do |t|
     t.string "name", null: false
@@ -55,7 +55,8 @@
     t.text "backtrace"
     t.datetime "created_at", precision: 6, null: false
     t.datetime "updated_at", precision: 6, null: false
-    t.index ["task_name", "created_at"], name: "index_maintenance_tasks_runs_on_task_name_and_created_at", order: { created_at: :desc }
+    t.text "arguments"
+    t.index ["task_name", "created_at"], name: "index_maintenance_tasks_runs_on_task_name_and_created_at"
   end
 
   create_table "posts", force: :cascade do |t|

test/jobs/maintenance_tasks/task_job_test.rb

@@ -355,5 +355,18 @@ class << self
 
       assert_predicate run.reload, :succeeded?
     end
+
+    test ".perform_now makes arguments supplied for Task parameters available" do
+      post = Post.last
+      Maintenance::ParamsTask.any_instance.expects(:process).once.with(post)
+
+      run = Run.create!(
+        task_name: "Maintenance::ParamsTask",
+        arguments: { post_ids: post.id.to_s }
+      )
+      TaskJob.perform_now(run)
+
+      assert_predicate run.reload, :succeeded?
+    end
   end
 end

test/models/maintenance_tasks/run_test.rb

@@ -5,19 +5,27 @@ module MaintenanceTasks
   class RunTest < ActiveSupport::TestCase
     test "invalid if the task doesn't exist" do
       run = Run.new(task_name: "Maintenance::DoesNotExist")
-      refute run.valid?
+      refute_predicate run, :valid?
     end
 
     test "invalid if associated with CSV Task and no attachment" do
       run = Run.new(task_name: "Maintenance::ImportPostsTask")
-      refute run.valid?
+      refute_predicate run, :valid?
     end
 
     test "invalid if unassociated with CSV Task and attachment" do
       run = Run.new(task_name: "Maintenance::UpdatePostsTask")
       csv = Rack::Test::UploadedFile.new(file_fixture("sample.csv"), "text/csv")
       run.csv_file.attach(csv)
-      refute run.valid?
+      refute_predicate run, :valid?
+    end
+
+    test "invalid if associated Task has parameters and they are invalid" do
+      run = Run.new(
+        task_name: "Maintenance::ParamsTask",
+        arguments: { post_ids: "xyz" }
+      )
+      refute_predicate run, :valid?
     end
 
     test "#persist_progress persists increments to tick count and time_running" do
@@ -276,6 +284,22 @@ class RunTest < ActiveSupport::TestCase
       end
     end
 
+    test "#task returns Task instance for Run" do
+      run = Run.new(task_name: "Maintenance::UpdatePostsTask")
+      assert_kind_of Maintenance::UpdatePostsTask, run.task
+    end
+
+    test "#validate_task_arguments instantiates Task and assigns arguments if Task has parameters" do
+      run = Run.new(
+        task_name: "Maintenance::ParamsTask",
+        arguments: { post_ids: "1,2,3" }
+      )
+      run.validate_task_arguments
+
+      assert_predicate run, :valid?
+      assert_equal "1,2,3", run.task.post_ids
+    end
+
     private
 
     def count_uncached_queries(&block)

test/models/maintenance_tasks/runner_test.rb

@@ -83,6 +83,16 @@ class RunnerTest < ActiveSupport::TestCase
       end
     end
 
+    test "#run raises ActiveRecord::ValueTooLong error if arguments input is too long" do
+      Run.any_instance.expects(:enqueued!).raises(ActiveRecord::ValueTooLong)
+      assert_raises(ActiveRecord::ValueTooLong) do
+        @runner.run(
+          name: "Maintenance::ParamsTask",
+          arguments: { post_ids: "123" }
+        )
+      end
+    end
+
     test "#run attaches CSV file to Run if one is provided" do
       @runner.run(name: "Maintenance::ImportPostsTask", csv_file: csv_io)