Merge branch 'master' into feature/retry-requests

lib/commands/unipept.rb

@@ -1,9 +1,11 @@
 require 'typhoeus'
 
-require_relative '../formatters'
-require_relative '../configuration'
 require_relative '../batch_order'
 require_relative '../batch_iterator'
+require_relative '../configuration'
+require_relative '../formatters'
+require_relative '../output_writer'
+require_relative '../server_message'
 require_relative '../version'
 
 require_relative 'unipept/config'
@@ -50,6 +52,8 @@ def create_root_command
         flag :v, :version, 'displays the version'
         flag :q, :quiet, 'disable service messages'
         option :i, :input, 'read input from file', argument: :required
+        option nil, :batch, 'specify the batch size', argument: :required, hidden: true
+        option nil, :parallel, 'specify the number of parallel requests', argument: :required, hidden: true
         option :o, :output, 'write output to file', argument: :required
         option :f, :format, "define the output format (available: #{Unipept::Formatter.available.join ', ' }) (default: #{Unipept::Formatter.default})", argument: :required
 

lib/commands/unipept/api_runner.rb

@@ -6,8 +6,6 @@ class Commands::ApiRunner < Cri::CommandRunner
 
     attr_reader :url
 
-    attr_reader :message_url
-
     attr_reader :user_agent
 
     def initialize(args, opts, cmd)
@@ -16,7 +14,6 @@ def initialize(args, opts, cmd)
       set_configuration
 
       @url = "#{@host}/api/v1/#{cmd.name}.json"
-      @message_url = "#{@host}/api/v1/messages.json"
     end
 
     # Sets the configurable options of the command line app:
@@ -56,66 +53,66 @@ def input_iterator
       $stdin.each_line
     end
 
-    # Returns the default batch_size of a command.
-    def batch_size
+    def output_writer
+      @output_writer ||= OutputWriter.new(options[:output])
+    end
+
+    # Returns the default default_batch_size of a command.
+    def default_batch_size
       100
     end
 
-    # Constructs a request body (a Hash) for set of input strings, using the
-    # options supplied by the user.
-    def construct_request_body(input)
-      names = selected_fields.empty? || selected_fields.any? { |f| f.to_s.include? 'name' }
-      { input: input,
-        equate_il: options[:equate] == true,
-        extra: options[:all] == true,
-        names: options[:all] == true && names
-      }
+    # returns the effective batch_size of a command
+    def batch_size
+      if options[:batch]
+        options[:batch].to_i
+      else
+        default_batch_size
+      end
     end
 
-    # Returns an array of regular expressions containing all the selected fields
-    def selected_fields
-      @selected_fields ||= [*options[:select]].map { |f| f.split(',') }.flatten.map { |f| glob_to_regex(f) }
+    # Returns a new batch_iterator based on the batch_size
+    def batch_iterator
+      Unipept::BatchIterator.new(batch_size)
     end
 
-    # Returns a formatter, based on the format specified in the options
-    def formatter
-      @formatter ||= Unipept::Formatter.new_for_format(options[:format])
+    def concurrent_requests
+      if options[:parallel]
+        options[:parallel].to_i
+      else
+        10
+      end
     end
 
-    # Checks if the server has a message and prints it if not empty.
-    # We will only check this once a day and won't print anything if the quiet
-    # option is set or if we output to a file.
-    def print_server_message
-      return if options[:quiet]
-      return unless $stdout.tty?
-      return if recently_fetched?
-      @configuration['last_fetch_date'] = Time.now
-      @configuration.save
-      resp = fetch_server_message
-      puts resp unless resp.empty?
+    def queue_size
+      concurrent_requests * 20
     end
 
-    # Fetches a message from the server and returns it
-    def fetch_server_message
-      Typhoeus.get(@message_url, params: { version: Unipept::VERSION }).body.chomp
+    # Returns an array of regular expressions containing all the selected fields
+    def selected_fields
+      @selected_fields ||= [*options[:select]].map { |f| f.split(',') }.flatten.map { |f| glob_to_regex(f) }
     end
 
-    # Returns true if the last check for a server message was less than a day
-    # ago.
-    def recently_fetched?
-      last_fetched = @configuration['last_fetch_date']
-      !last_fetched.nil? && (last_fetched + 60 * 60 * 24) > Time.now
+    # Returns a formatter, based on the format specified in the options
+    def formatter
+      @formatter ||= Unipept::Formatter.new_for_format(options[:format])
     end
 
-    # Returns a new batch_iterator based on the batch_size
-    def batch_iterator
-      Unipept::BatchIterator.new(batch_size)
+    # Constructs a request body (a Hash) for set of input strings, using the
+    # options supplied by the user.
+    def construct_request_body(input)
+      names = selected_fields.empty? || selected_fields.any? { |f| f.to_s.include? 'name' }
+      { input: input,
+        equate_il: options[:equate] == true,
+        extra: options[:all] == true,
+        names: options[:all] == true && names
+      }
     end
 
     # Runs the command
     def run
-      print_server_message
-      hydra = Typhoeus::Hydra.new(max_concurrency: 10)
+      ServerMessage.new(@host).print unless options[:quiet]
+      hydra = Typhoeus::Hydra.new(max_concurrency: concurrent_requests)
       batch_order = Unipept::BatchOrder.new
 
       batch_iterator.iterate(input_iterator) do |input_slice, batch_id, fasta_mapper|
@@ -133,7 +130,7 @@ def run
         end
 
         hydra.queue request
-        hydra.run if batch_id % 200 == 0
+        hydra.run if batch_id % queue_size == 0
       end
 
       hydra.run
@@ -148,16 +145,6 @@ def save_error(message)
       $stderr.puts "API request failed! log can be found in #{path}"
     end
 
-    # Write a string to the output defined by the command. If a file is given,
-    # write it to the file. If not, write to stdout
-    def write_to_output(string)
-      if options[:output]
-        File.open(options[:output], 'a') { |f| f.write string }
-      else
-        puts string
-      end
-    end
-
     private
 
     def error_file_path
@@ -172,8 +159,8 @@ def handle_response(response, batch_id, fasta_mapper)
 
         lambda do
           unless result.empty?
-            write_to_output formatter.header(result, fasta_mapper) if batch_id == 0
-            write_to_output formatter.format(result, fasta_mapper)
+            output_writer.write_line formatter.header(result, fasta_mapper) if batch_id == 0
+            output_writer.write_line formatter.format(result, fasta_mapper)
           end
         end
       elsif response.timed_out?

lib/commands/unipept/pept2lca.rb

@@ -1,7 +1,7 @@
 require_relative 'api_runner'
 module Unipept::Commands
   class Pept2lca < ApiRunner
-    def batch_size
+    def default_batch_size
       if options[:all]
         100
       else

lib/commands/unipept/pept2prot.rb

@@ -2,7 +2,7 @@
 
 module Unipept::Commands
   class Pept2prot < ApiRunner
-    def batch_size
+    def default_batch_size
       if options[:all]
         5
       else

lib/commands/unipept/pept2taxa.rb

@@ -1,7 +1,7 @@
 require_relative 'api_runner'
 module Unipept::Commands
   class Pept2taxa < ApiRunner
-    def batch_size
+    def default_batch_size
       if options[:all]
         5
       else

lib/commands/unipept/taxa2lca.rb

@@ -5,7 +5,7 @@ def batch_iterator
       SimpleBatchIterator.new
     end
 
-    def batch_size
+    def default_batch_size
       fail 'NOT NEEDED FOR TAXA2LCA'
     end
   end

lib/commands/unipept/taxonomy.rb

@@ -1,7 +1,7 @@
 require_relative 'api_runner'
 module Unipept::Commands
   class Taxonomy < ApiRunner
-    def batch_size
+    def default_batch_size
       100
     end
   end

lib/output_writer.rb

@@ -0,0 +1,13 @@
+module Unipept
+  class OutputWriter
+    attr_reader :output
+
+    def initialize(file)
+      @output = file ? File.open(file, 'a') : $stdout
+    end
+
+    def write_line(line)
+      @output.write line
+    end
+  end
+end

lib/server_message.rb

@@ -0,0 +1,45 @@
+require 'typhoeus'
+
+require_relative 'configuration'
+
+module Unipept
+  class ServerMessage
+    attr_reader :message_url
+
+    attr_reader :configuration
+
+    def initialize(host)
+      @message_url = "#{host}/api/v1/messages.json"
+      @configuration = Unipept::Configuration.new
+    end
+
+    # Checks if the server has a message and prints it if not empty.
+    # We will only check this once a day and won't print anything if the quiet
+    # option is set or if we output to a file.
+    def print
+      return unless $stdout.tty?
+      return if recently_fetched?
+      resp = fetch_server_message
+      update_fetched
+      puts resp unless resp.empty?
+    end
+
+    # Fetches a message from the server and returns it
+    def fetch_server_message
+      Typhoeus.get(@message_url, params: { version: Unipept::VERSION }).body.chomp
+    end
+
+    # Returns true if the last check for a server message was less than a day
+    # ago.
+    def recently_fetched?
+      last_fetched = @configuration['last_fetch_date']
+      !last_fetched.nil? && (last_fetched + 60 * 60 * 24) > Time.now
+    end
+
+    # Updates the last checked timestamp
+    def update_fetched
+      @configuration['last_fetch_date'] = Time.now
+      @configuration.save
+    end
+  end
+end