class_docstr: An async implementation of a gRPC server.
code: def __init__(
      self,
      bento_service: Service,
      bind_address: str,
      max_message_length: int
      | None = Provide[BentoMLContainer.grpc.max_message_length],
      maximum_concurrent_rpcs: int
      | None = Provide[BentoMLContainer.grpc.maximum_concurrent_rpcs],
      enable_reflection: bool = False,
      enable_channelz: bool = False,
      max_concurrent_streams: int | None = None,
      migration_thread_pool_workers: int = 1,
      ssl_certfile: str | None = None,
      ssl_keyfile: str | None = None,
      ssl_ca_certs: str | None = None,
      graceful_shutdown_timeout: float | None = None,
      compression: grpc.Compression | None = None,
      protocol_version: str = LATEST_PROTOCOL_VERSION,
  ):
      pb, _ = import_generated_stubs(protocol_version)
  
      self.bento_service = bento_service
      self.servicer = bento_service.get_grpc_servicer(protocol_version)
  
      # options
      self.max_message_length = max_message_length
      self.max_concurrent_streams = max_concurrent_streams
      self.bind_address = bind_address
      self.enable_reflection = enable_reflection
      self.enable_channelz = enable_channelz
      self.graceful_shutdown_timeout = graceful_shutdown_timeout
      self.ssl_certfile = ssl_certfile
      self.ssl_keyfile = ssl_keyfile
      self.ssl_ca_certs = ssl_ca_certs
      self.protocol_version = protocol_version
  
      # Create a health check servicer. We use the non-blocking implementation
      # to avoid thread starvation.
      self.health_servicer = health.aio.HealthServicer()
  
      self.mount_servicers = self.bento_service.mount_servicers
  
      self.service_names = tuple(
          service.full_name
          for service in pb.DESCRIPTOR.services_by_name.values()
      ) + (health.SERVICE_NAME,)
  
      super().__init__(
          # Note that the max_workers are used inside ThreadPoolExecutor.
          # This ThreadPoolExecutor are used by aio.Server() to execute non-AsyncIO RPC handlers.
          # Setting it to 1 makes it thread-safe for sync APIs.
          thread_pool=ThreadPoolExecutor(
              max_workers=migration_thread_pool_workers
          ),
          generic_handlers=()
          if self.handlers is None
          else self.handlers,
          interceptors=list(map(lambda x: x(), self.interceptors)),
          options=self.options,
          # maximum_concurrent_rpcs defines the maximum number of concurrent RPCs this server
          # will service before returning RESOURCE_EXHAUSTED status.
          # Set to None will indicate no limit.
          maximum_concurrent_rpcs=maximum_concurrent_rpcs,
          compression=compression,
      )
  
imports: from __future__ import annotations
  import asyncio
  import inspect
  import logging
  import os
  import sys
  import typing
  from concurrent.futures import ThreadPoolExecutor
  from functools import cached_property
  from functools import partial
  from typing import TYPE_CHECKING
  from simple_di import Provide
  from simple_di import inject
  from ...grpc.utils import LATEST_PROTOCOL_VERSION
  from ...grpc.utils import import_generated_stubs
  from ...grpc.utils import import_grpc
  from ...grpc.utils import load_from_file
  from ..configuration.containers import BentoMLContainer
  from ..context import ServiceContext
  from ..utils import LazyLoader
label_desc: Use image node_id 1 to create a new Server object from inherited base classes: aio._server with example: obj = Server(bento_service, bind_address, max_message_length, maximum_concurrent_rpcs, enable_reflection, enable_channelz, max_concurrent_streams, migration_thread_pool_workers, ssl_certfile, ssl_keyfile, ssl_ca_certs, graceful_shutdown_timeout, compression, protocol_version)