class_docstr: Metrics manager and batch writer 
class_docstr_tok: ["Metrics","manager","and","batch","writer"]
docstr: Sends an iterable of events as a series of batch operations. note: metric send does not raise on error
docstr_tok: ["Sends","an","iterable","of","events","as","a","series","of","batch","operations",".","note",":","metric","send","does","not","raise","on","error"]
code: def _do_write_events(self, events, storage_uri=None):
      """Sends an iterable of events as a series of batch operations. note: metric send does not raise on error"""
      assert isinstance(events, (list, tuple))
      assert all(isinstance(x, MetricsEventAdapter) for x in events)
  
      # def event_key(ev):
      #     return (ev.metric, ev.variant)
      #
      # events = sorted(events, key=event_key)
      # multiple_events_for = [k for k, v in groupby(events, key=event_key) if len(list(v)) > 1]
      # if multiple_events_for:
      #     log.warning(
      #         'More than one metrics event sent for these metric/variant combinations in a report: %s' %
      #         ', '.join('%s/%s' % k for k in multiple_events_for))
  
      storage_uri = storage_uri or self._storage_uri
  
      def update_and_get_file_entry(ev):
          entry = ev.get_file_entry()
          kwargs = {}
          if entry:
              key, url = ev.get_target_full_upload_uri(
                  storage_uri, self.storage_key_prefix, quote_uri=False
              )
              kwargs[entry.key_prop] = key
              kwargs[entry.url_prop] = url
              if not entry.stream:
                  # if entry has no stream, we won't upload it
                  entry = None
              else:
                  if not isinstance(entry.stream, Path) and not hasattr(
                      entry.stream, "read"
                  ):
                      raise ValueError(
                          "Invalid file object %s" % entry.stream
                      )
                  entry.url = url
          ev.update(
              task=self._task_id,
              iter_offset=self._task_iteration_offset,
              **kwargs,
          )
          return entry
  
      # prepare event needing file upload
      entries = []
      for ev in events[:]:
          try:
              e = update_and_get_file_entry(ev)
              if e:
                  entries.append(e)
          except Exception as ex:
              self._get_logger().warning(str(ex))
              events.remove(ev)
  
      # upload the needed files
      if entries:
          # upload files
          def upload(e):
              upload_uri = e.upload_uri or storage_uri
  
              try:
                  storage = self._get_storage(upload_uri)
                  retries = (
                      getattr(e, "retries", None)
                      or self._file_upload_retries
                  )
                  if isinstance(e.stream, Path):
                      url = storage.upload(
                          e.stream.as_posix(),
                          e.url,
                          retries=retries,
                          return_canonized=False,
                      )
                  else:
                      url = storage.upload_from_stream(
                          e.stream,
                          e.url,
                          retries=retries,
                          return_canonized=False,
                      )
                  e.event.update(url=url)
              except Exception as exp:
                  self._get_logger().warning(
                      "Failed uploading to {} ({})".format(
                          upload_uri
                          if upload_uri
                          else "(Could not calculate upload uri)",
                          exp,
                      )
                  )
  
                  e.set_exception(exp)
              if not isinstance(e.stream, Path):
                  e.stream.close()
              if e.delete_local_file:
                  # noinspection PyBroadException
                  try:
                      Path(e.delete_local_file).unlink()
                  except Exception:
                      pass
  
          self._initialize_upload_pools()
          res = self._file_upload_pool.map_async(upload, entries)
          res.wait()
  
          # remember the last time we uploaded a file
          self._file_upload_time = time()
  
      t_f, t_u, t_ref = (
          self._file_related_event_time,
          self._file_upload_time,
          self._file_upload_starvation_warning_sec,
      )
      if t_f and t_u and t_ref and (t_f - t_u) > t_ref:
          self._get_logger().warning(
              "Possible metrics file upload starvation: "
              "files were not uploaded for {} seconds".format(t_ref)
          )
  
      # send the events in a batched request
      good_events = [ev for ev in events if ev.upload_exception is None]
      error_events = [
          ev for ev in events if ev.upload_exception is not None
      ]
  
      if error_events:
          self._get_logger().error(
              "Not uploading {}/{} events because the data upload failed".format(
                  len(error_events),
                  len(events),
              )
          )
  
      if good_events:
          _events = [ev.get_api_event() for ev in good_events]
          additional_kwargs = {}
          if Session.check_min_api_version("2.23"):
              additional_kwargs["model_event"] = self._for_model
          batched_requests = [
              api_events.AddRequest(event=ev, **additional_kwargs)
              for ev in _events
              if ev
          ]
          if batched_requests:
              if self._offline_mode:
                  with open(
                      self._offline_log_filename.as_posix(), "at"
                  ) as f:
                      f.write(
                          json.dumps(
                              [b.to_dict() for b in batched_requests]
                          )
                          + "\n"
                      )
                  return
  
              req = api_events.AddBatchRequest(
                  requests=batched_requests
              )
              return self.send(req, raise_on_errors=False)
  
      return None
  
imports: import json
  import os
  from functools import partial
  from logging import warning
  from multiprocessing.pool import ThreadPool
  from time import time
  from pathlib2 import Path
  from ...backend_api import Session
  from ...backend_api.services import events
  from ..base import InterfaceBase
  from ...config import config, deferred_config
  from ...debugging import get_logger
  from ...storage.helper import StorageHelper
  from .events import MetricsEventAdapter
  from ...utilities.process.mp import SingletonLock
label_desc: Use image node_id 9 for calling the Metrics obj's underlying member method code with example usage: obj._do_write_events(events, storage_uri) and returns: None, entry, self