<commit_before>module Resque
  module Plugins
    class Pertry
      JOB_HASH = :_pertry

      include Resque::Pertry::Persistence
      include Resque::Pertry::Retry

      class << self

        # Enqueue a job
        def enqueue(args = {})
          raise ArgumentError, "Invalid arguments, expecting a Hash but got: #{args.inspect}" unless Hash === args

          args.symbolize_keys!
          args = check_arguments(args)
          raise ArgumentError, "Invalid arguments, #{JOB_HASH} is a reserved argument!" if args.key?(JOB_HASH)

          Resque.enqueue(self, args)
        end

        # Perform a job
        def perform(args = {})
          raise ArgumentError, "Invalid arguments, expecting a Hash but got: #{args.inspect}" unless Hash === args

          instance(args).perform
        end

        def instance(args = {})
          args.symbolize_keys!
          raise ArgumentError, "Job is not supported, missing key #{JOB_HASH} from payload #{args.inspect}" unless args.key?(JOB_HASH)

          new(check_arguments(args), args[JOB_HASH])
        end

        # Specificy job queue
        def in_queue(queue)
          @queue = queue.to_sym
        end
        
        # Get job queue
        def queue
          @queue or raise ArgumentError, "No queue defined for job #{self.name}!"
        end

        # Define required job attributes
        def needs(*arguments)
          arguments.each do |argument|
            if Hash === argument
              argument.each do |key, default|
                self.required_arguments << { :name => key, :default => default }
              end
            else
              self.required_arguments << { :name => argument }
            end
          end
        end

        # List of required attributes
        def required_arguments
          @required_arguments ||= []
        end

        private

        # Check that job arguments match required arguments
        def check_arguments(provided_arguments)
          required_arguments.inject({}) do |checked_arguments, argument|
            raise ArgumentError, "#{self} is missing required argument #{argument[:name]} from #{provided_arguments.inspect}" unless provided_arguments.member?(argument[:name]) || argument.member?(:default)

            provided_argument = provided_arguments[argument[:name]] || argument[:default]
            # TODO check that provided_argument is serializable as json

            checked_arguments[argument[:name]] = provided_argument
            checked_arguments
          end
        end

      end

      def initialize(arguments, job_properties)
        set_job_arguments(arguments)
        set_job_properties(job_properties)
      end

      # Perform method needs to be overridden in job classes
      def perform
        raise NoMethodError, "No method #{self.class.name}#perform defined!"
      end

      def arguments
        @_arguments
      end

      def payload
        @_arguments.merge(JOB_HASH => @_job_properties)
      end

      private

      def set_job_properties(hash)
        @_job_properties ||= {}
        @_job_properties.merge!(hash)
      end

      def set_job_arguments(hash)
        @_arguments = hash
        arguments.each do |key, val|
          instance_variable_set("@#{key}", val)
        end
      end

    end
  end
end<commit_msg>allow jobs to handle raised exceptions before Resque gets them, allowing a job to fail or complete themselves before using up retries<commit_after>module Resque
  module Plugins
    class Pertry
      JOB_HASH = :_pertry

      include Resque::Pertry::Persistence
      include Resque::Pertry::Retry

      attr_reader :exception_handled

      class << self

        # Enqueue a job
        def enqueue(args = {})
          raise ArgumentError, "Invalid arguments, expecting a Hash but got: #{args.inspect}" unless Hash === args

          args.symbolize_keys!
          args = check_arguments(args)
          raise ArgumentError, "Invalid arguments, #{JOB_HASH} is a reserved argument!" if args.key?(JOB_HASH)

          Resque.enqueue(self, args)
        end

        # Perform a job
        def perform(args = {})
          raise ArgumentError, "Invalid arguments, expecting a Hash but got: #{args.inspect}" unless Hash === args

          begin
            job = instance(args)
            job.perform
          rescue => e
            job.handle_exception(e)
            raise unless job.exception_handled
          end
        end

        def instance(args = {})
          args.symbolize_keys!
          raise ArgumentError, "Job is not supported, missing key #{JOB_HASH} from payload #{args.inspect}" unless args.key?(JOB_HASH)

          new(check_arguments(args), args[JOB_HASH])
        end

        # Specificy job queue
        def in_queue(queue)
          @queue = queue.to_sym
        end
        
        # Get job queue
        def queue
          @queue or raise ArgumentError, "No queue defined for job #{self.name}!"
        end

        # Define required job attributes
        def needs(*arguments)
          arguments.each do |argument|
            if Hash === argument
              argument.each do |key, default|
                self.required_arguments << { :name => key, :default => default }
              end
            else
              self.required_arguments << { :name => argument }
            end
          end
        end

        # List of required attributes
        def required_arguments
          @required_arguments ||= []
        end

        private

        # Check that job arguments match required arguments
        def check_arguments(provided_arguments)
          required_arguments.inject({}) do |checked_arguments, argument|
            raise ArgumentError, "#{self} is missing required argument #{argument[:name]} from #{provided_arguments.inspect}" unless provided_arguments.member?(argument[:name]) || argument.member?(:default)

            provided_argument = provided_arguments[argument[:name]] || argument[:default]
            # TODO check that provided_argument is serializable as json

            checked_arguments[argument[:name]] = provided_argument
            checked_arguments
          end
        end

      end

      def initialize(arguments, job_properties)
        set_job_arguments(arguments)
        set_job_properties(job_properties)
      end

      # Perform method needs to be overridden in job classes
      def perform
        raise NoMethodError, "No method #{self.class.name}#perform defined!"
      end

      def handle_exception(exception)
        # we don't handle exceptions by default, leave it for resque
        # when overridding this method, if you decide you want to completely
        # handle it, and not pass it back to resque, you'll need to call
        # exception_handled!
      end

      def exception_handled
        @exception_handled ||= false
      end

      def exception_handled!
        @exception_handled = true
      end

      # mark job as failed, won't be retried
      def fail!
        Resque::Pertry::ResquePertryPersistence.fail_job(self.class, payload)
      end

      # mark job as complete
      def complete!
        Resque::Pertry::ResquePertryPersistence.finnish_job(self.class, payload)
      end

      def arguments
        @_arguments
      end

      def payload
        @_arguments.merge(JOB_HASH => @_job_properties)
      end

      private

      def set_job_properties(hash)
        @_job_properties ||= {}
        @_job_properties.merge!(hash)
      end

      def set_job_arguments(hash)
        @_arguments = hash
        arguments.each do |key, val|
          instance_variable_set("@#{key}", val)
        end
      end

    end
  end
end
{'fraction_non_alphanumeric': 0.07845934379457917, 'fraction_numerical': 0.0, 'mean_word_length': 2.0959454034524287, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '95171', 'n_tokens_mistral': 2014, 'n_tokens_neox': 1972, 'n_words': 597}