class_docstr: Object to encapsulate the code related to handle the data requests passing to executor and its returned values
class_docstr_tok: ["Object","to","encapsulate","the","code","related","to","handle","the","data","requests","passing","to","executor","and","its","returned","values"]
docstr: Initialize private parameters and execute private loading functions.
  
  :param args: args from CLI
  :param logger: the logger provided by the user
  :param metrics_registry: optional metrics registry for prometheus used if we need to expose metrics from the executor of from the data request handler
  :param tracer_provider: Optional tracer_provider that will be provided to the executor for tracing
  :param meter_provider: Optional meter_provider that will be provided to the executor for metrics
  :param meter: meter object from runtime
  :param tracer: tracer object from runtime
  :param deployment_name: name of the deployment to use as Executor name to set in requests
  :param kwargs: extra keyword arguments
docstr_tok: ["Initialize","private","parameters","and","execute","private","loading","functions",".",":","param","args",":","args","from","CLI",":","param","logger",":","the","logger","provided","by","the","user",":","param","metrics_registry",":","optional","metrics","registry","for","prometheus","used","if","we","need","to","expose","metrics","from","the","executor","of","from","the","data","request","handler",":","param","tracer_provider",":","Optional","tracer_provider","that","will","be","provided","to","the","executor","for","tracing",":","param","meter_provider",":","Optional","meter_provider","that","will","be","provided","to","the","executor","for","metrics",":","param","meter",":","meter","object","from","runtime",":","param","tracer",":","tracer","object","from","runtime",":","param","deployment_name",":","name","of","the","deployment","to","use","as","Executor","name","to","set","in","requests",":","param","kwargs",":","extra","keyword","arguments"]
code: def __init__(
      self,
      args: "argparse.Namespace",
      logger: "JinaLogger",
      metrics_registry: Optional["CollectorRegistry"] = None,
      tracer_provider: Optional["trace.TracerProvider"] = None,
      meter_provider: Optional["metrics.MeterProvider"] = None,
      meter=None,
      tracer=None,
      deployment_name: str = "",
      **kwargs,
  ):
      """Initialize private parameters and execute private loading functions.
  
      :param args: args from CLI
      :param logger: the logger provided by the user
      :param metrics_registry: optional metrics registry for prometheus used if we need to expose metrics from the executor of from the data request handler
      :param tracer_provider: Optional tracer_provider that will be provided to the executor for tracing
      :param meter_provider: Optional meter_provider that will be provided to the executor for metrics
      :param meter: meter object from runtime
      :param tracer: tracer object from runtime
      :param deployment_name: name of the deployment to use as Executor name to set in requests
      :param kwargs: extra keyword arguments
      """
      super().__init__()
      self.meter = meter
      self.metrics_registry = metrics_registry
      self.tracer = tracer
      self.args = args
      self.logger = logger
      self._is_closed = False
      if self.metrics_registry:
          with ImportExtensions(
              required=True,
              help_text="You need to install the `prometheus_client` to use the montitoring functionality of jina",
          ):
              from prometheus_client import Counter, Summary
  
          self._summary = Summary(
              "receiving_request_seconds",
              "Time spent processing request",
              registry=self.metrics_registry,
              namespace="jina",
              labelnames=("runtime_name",),
          ).labels(self.args.name)
  
          self._failed_requests_metrics = Counter(
              "failed_requests",
              "Number of failed requests",
              registry=self.metrics_registry,
              namespace="jina",
              labelnames=("runtime_name",),
          ).labels(self.args.name)
  
          self._successful_requests_metrics = Counter(
              "successful_requests",
              "Number of successful requests",
              registry=self.metrics_registry,
              namespace="jina",
              labelnames=("runtime_name",),
          ).labels(self.args.name)
  
      else:
          self._summary = None
          self._failed_requests_metrics = None
          self._successful_requests_metrics = None
  
      if self.meter:
          self._receiving_request_seconds = self.meter.create_histogram(
              name="jina_receiving_request_seconds",
              description="Time spent processing request",
          )
          self._failed_requests_counter = self.meter.create_counter(
              name="jina_failed_requests",
              description="Number of failed requests",
          )
  
          self._successful_requests_counter = self.meter.create_counter(
              name="jina_successful_requests",
              description="Number of successful requests",
          )
      else:
          self._receiving_request_seconds = None
          self._failed_requests_counter = None
          self._successful_requests_counter = None
      self._metric_attributes = {"runtime_name": self.args.name}
      self._load_executor(
          metrics_registry=metrics_registry,
          tracer_provider=tracer_provider,
          meter_provider=meter_provider,
      )
      meter = (
          meter_provider.get_meter(self.__class__.__name__)
          if meter_provider
          else None
      )
      self._init_monitoring(metrics_registry, meter)
      self.deployment_name = deployment_name
      # In order to support batching parameters separately, we have to lazily create batch queues
      # So we store the config for each endpoint in the initialization
      self._batchqueue_config: Dict[str, Dict] = {}
      # the below is of "shape" exec_endpoint_name -> parameters_key -> batch_queue
      self._batchqueue_instances: Dict[str, Dict[str, BatchQueue]] = {}
      self._init_batchqueue_dict()
      self._snapshot = None
      self._did_snapshot_raise_exception = None
      self._restore = None
      self._did_restore_raise_exception = None
      self._snapshot_thread = None
      self._restore_thread = None
      self._snapshot_parent_directory = tempfile.mkdtemp()
      self._hot_reload_task = None
      if self.args.reload:
          self._hot_reload_task = asyncio.create_task(
              self._hot_reload()
          )
  
imports: import argparse
  import asyncio
  import functools
  import json
  import os
  import tempfile
  import threading
  import uuid
  import warnings
  from typing import TYPE_CHECKING, AsyncIterator, Dict, Generator, List, Optional, Tuple, Union
  from google.protobuf.struct_pb2 import Struct
  from jina._docarray import DocumentArray, docarray_v2
  from jina.constants import __default_endpoint__
  from jina.excepts import BadConfigSource, RuntimeTerminated
  from jina.helper import get_full_version
  from jina.importer import ImportExtensions
  from jina.proto import jina_pb2
  from jina.serve.executors import BaseExecutor
  from jina.serve.instrumentation import MetricsTimer
  from jina.serve.runtimes.worker.batch_queue import BatchQueue
  from jina.types.request.data import DataRequest, SingleDocumentRequest
label_desc: Use image node_id 1 to create a new WorkerRequestHandler object  with example: obj = WorkerRequestHandler(args, logger, metrics_registry, tracer_provider, meter_provider, meter, tracer, deployment_name)