class_docstr: :class:`JAMLCompatible` is a mixin class designed to be used with multiple inheritance.
  
  It will add :meth:`to_yaml` and :meth:`from_yaml` to the target class,
  making that class JAML-friendly.
  
  .. warning::
      For the sake of cooperative multiple inheritance, do NOT implement :meth:`__init__` for this class
docstr: A high-level interface for loading configuration with features
  of loading extra py_modules, substitute env & context variables. Any class that
  implements :class:`JAMLCompatible` mixin can enjoy this feature, e.g. :class:`BaseFlow`,
  :class:`BaseExecutor`, :class:`BaseGateway` and all their subclasses.
  
  Support substitutions in YAML:
      - Environment variables: ``${{ ENV.VAR }}`` (recommended), ``$VAR`` (deprecated).
      - Context dict (``context``): ``${{ CONTEXT.VAR }}``(recommended), ``${{ VAR }}``.
      - Internal reference via ``this`` and ``root``: ``${{this.same_level_key}}``, ``${{root.root_level_key}}``
  
  Substitutions are carried in the order and multiple passes to resolve variables with best effort.
  
  .. highlight:: yaml
  .. code-block:: yaml
  
      !BaseEncoder
      metas:
          name: ${{VAR_A}}  # env or context variables
          workspace: my-${{this.name}}  # internal reference
  
  .. highlight:: python
  .. code-block:: python
  
      # load Executor from yaml file
      BaseExecutor.load_config('a.yml')
  
      # load Executor from yaml file and substitute environment variables
      os.environ['VAR_A'] = 'hello-world'
      b = BaseExecutor.load_config('a.yml')
      assert b.name == 'hello-world'
  
      # load Executor from yaml file and substitute variables from a dict
      b = BaseExecutor.load_config('a.yml', context={'VAR_A': 'hello-world'})
      assert b.name == 'hello-world'
  
      # disable substitute
      b = BaseExecutor.load_config('a.yml', substitute=False)
  
  
  .. # noqa: DAR401
  :param source: the multi-kind source of the configs.
  :param allow_py_modules: allow importing plugins specified by ``py_modules`` in YAML at any levels
  :param substitute: substitute environment, internal reference and context variables.
  :param context: context replacement variables in a dict, the value of the dict is the replacement.
  :param uses_with: dictionary of parameters to overwrite from the default config's with field
  :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
  :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
  :param extra_search_paths: extra paths used when looking for executor yaml files
  :param py_modules: Optional py_module from which the object need to be loaded
  :param runtime_args: Optional dictionary of parameters runtime_args to be directly passed without being parsed into a yaml config
  :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's dynamic_batching field
  :param needs: the name of the Deployment(s) that this Deployment receives data from. One can also use "gateway" to indicate the connection with the gateway.
  :param include_gateway: Defines if the gateway deployment should be included, defaults to True
  :param noblock_on_start: If set, starting a Pod/Deployment does not block the thread/process. It then relies on '
      '`wait_start_success` at outer function for the postpone check.
  :param kwargs: kwargs for parse_config_source
  :return: :class:`JAMLCompatible` object
docstr_tok: ["A","high-level","interface","for","loading","configuration","with","features","of","loading","extra","py_modules",",","substitute","env","&","context","variables",".","Any","class","that","implements",":","class",":","`","JAMLCompatible","`","mixin","can","enjoy","this","feature",",","e.g",".",":","class",":","`","BaseFlow","`",",",":","class",":","`","BaseExecutor","`",",",":","class",":","`","BaseGateway","`","and","all","their","subclasses",".","Support","substitutions","in","YAML",":","-","Environment","variables",":","``","$","{","{","ENV.VAR","}","}","``","(","recommended",")",",","``","$","VAR","``","(","deprecated",")",".","-","Context","dict","(","``","context","``",")",":","``","$","{","{","CONTEXT.VAR","}","}","``","(","recommended",")",",","``","$","{","{","VAR","}","}","``",".","-","Internal","reference","via","``","this","``","and","``","root","``",":","``","$","{","{","this.same_level_key","}","}","``",",","``","$","{","{","root.root_level_key","}","}","``","Substitutions","are","carried","in","the","order","and","multiple","passes","to","resolve","variables","with","best","effort",".","..","highlight",":",":","yaml","..","code-block",":",":","yaml","!","BaseEncoder","metas",":","name",":","$","{","{","VAR_A","}","}","#","env","or","context","variables","workspace",":","my-","$","{","{","this.name","}","}","#","internal","reference","..","highlight",":",":","python","..","code-block",":",":","python","#","load","Executor","from","yaml","file","BaseExecutor.load_config","(","'","a.yml","'",")","#","load","Executor","from","yaml","file","and","substitute","environment","variables","os.environ","[","'VAR_A","'","]","=","'hello-world'","b","=","BaseExecutor.load_config","(","'","a.yml","'",")","assert","b.name","==","'hello-world'","#","load","Executor","from","yaml","file","and","substitute","variables","from","a","dict","b","=","BaseExecutor.load_config","(","'","a.yml","'",",","context=","{","'VAR_A","'",":","'hello-world","'","}",")","assert","b.name","==","'hello-world'","#","disable","substitute","b","=","BaseExecutor.load_config","(","'","a.yml","'",",","substitute=False",")","..","#","noqa",":","DAR401",":","param","source",":","the","multi-kind","source","of","the","configs",".",":","param","allow_py_modules",":","allow","importing","plugins","specified","by","``","py_modules","``","in","YAML","at","any","levels",":","param","substitute",":","substitute","environment",",","internal","reference","and","context","variables",".",":","param","context",":","context","replacement","variables","in","a","dict",",","the","value","of","the","dict","is","the","replacement",".",":","param","uses_with",":","dictionary","of","parameters","to","overwrite","from","the","default","config","'s","with","field",":","param","uses_metas",":","dictionary","of","parameters","to","overwrite","from","the","default","config","'s","metas","field",":","param","uses_requests",":","dictionary","of","parameters","to","overwrite","from","the","default","config","'s","requests","field",":","param","extra_search_paths",":","extra","paths","used","when","looking","for","executor","yaml","files",":","param","py_modules",":","Optional","py_module","from","which","the","object","need","to","be","loaded",":","param","runtime_args",":","Optional","dictionary","of","parameters","runtime_args","to","be","directly","passed","without","being","parsed","into","a","yaml","config",":","param","uses_dynamic_batching",":","dictionary","of","parameters","to","overwrite","from","the","default","config","'s","dynamic_batching","field",":","param","needs",":","the","name","of","the","Deployment","(","s",")","that","this","Deployment","receives","data","from",".","One","can","also","use","``","gateway","''","to","indicate","the","connection","with","the","gateway",".",":","param","include_gateway",":","Defines","if","the","gateway","deployment","should","be","included",",","defaults","to","True",":","param","noblock_on_start",":","If","set",",","starting","a","Pod\/Deployment","does","not","block","the","thread\/process",".","It","then","relies","on","'","'","`","wait_start_success","`","at","outer","function","for","the","postpone","check",".",":","param","kwargs",":","kwargs","for","parse_config_source",":","return",":",":","class",":","`","JAMLCompatible","`","object"]
code: def load_config(
      cls,
      source: Union[str, TextIO, Dict],
      *,
      allow_py_modules: bool = True,
      substitute: bool = True,
      context: Optional[Dict[str, Any]] = None,
      uses_with: Optional[Dict] = None,
      uses_metas: Optional[Dict] = None,
      uses_requests: Optional[Dict] = None,
      extra_search_paths: Optional[List[str]] = None,
      py_modules: Optional[str] = None,
      runtime_args: Optional[Dict[str, Any]] = None,
      uses_dynamic_batching: Optional[Dict] = None,
      needs: Optional[Set[str]] = None,
      include_gateway: bool = True,
      noblock_on_start: bool = False,
      **kwargs,
  ) -> "JAMLCompatible":
      """A high-level interface for loading configuration with features
      of loading extra py_modules, substitute env & context variables. Any class that
      implements :class:`JAMLCompatible` mixin can enjoy this feature, e.g. :class:`BaseFlow`,
      :class:`BaseExecutor`, :class:`BaseGateway` and all their subclasses.
  
      Support substitutions in YAML:
          - Environment variables: ``${{ ENV.VAR }}`` (recommended), ``$VAR`` (deprecated).
          - Context dict (``context``): ``${{ CONTEXT.VAR }}``(recommended), ``${{ VAR }}``.
          - Internal reference via ``this`` and ``root``: ``${{this.same_level_key}}``, ``${{root.root_level_key}}``
  
      Substitutions are carried in the order and multiple passes to resolve variables with best effort.
  
      .. highlight:: yaml
      .. code-block:: yaml
  
          !BaseEncoder
          metas:
              name: ${{VAR_A}}  # env or context variables
              workspace: my-${{this.name}}  # internal reference
  
      .. highlight:: python
      .. code-block:: python
  
          # load Executor from yaml file
          BaseExecutor.load_config('a.yml')
  
          # load Executor from yaml file and substitute environment variables
          os.environ['VAR_A'] = 'hello-world'
          b = BaseExecutor.load_config('a.yml')
          assert b.name == 'hello-world'
  
          # load Executor from yaml file and substitute variables from a dict
          b = BaseExecutor.load_config('a.yml', context={'VAR_A': 'hello-world'})
          assert b.name == 'hello-world'
  
          # disable substitute
          b = BaseExecutor.load_config('a.yml', substitute=False)
  
  
      .. # noqa: DAR401
      :param source: the multi-kind source of the configs.
      :param allow_py_modules: allow importing plugins specified by ``py_modules`` in YAML at any levels
      :param substitute: substitute environment, internal reference and context variables.
      :param context: context replacement variables in a dict, the value of the dict is the replacement.
      :param uses_with: dictionary of parameters to overwrite from the default config's with field
      :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
      :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
      :param extra_search_paths: extra paths used when looking for executor yaml files
      :param py_modules: Optional py_module from which the object need to be loaded
      :param runtime_args: Optional dictionary of parameters runtime_args to be directly passed without being parsed into a yaml config
      :param uses_dynamic_batching: dictionary of parameters to overwrite from the default config's dynamic_batching field
      :param needs: the name of the Deployment(s) that this Deployment receives data from. One can also use "gateway" to indicate the connection with the gateway.
      :param include_gateway: Defines if the gateway deployment should be included, defaults to True
      :param noblock_on_start: If set, starting a Pod/Deployment does not block the thread/process. It then relies on '
          '`wait_start_success` at outer function for the postpone check.
      :param kwargs: kwargs for parse_config_source
      :return: :class:`JAMLCompatible` object
      """
      if runtime_args:
          kwargs[
              "runtimes_args"
          ] = (
              dict()
          )  # when we have runtime args it is needed to have an empty runtime args session in the yam config
  
      if py_modules:
          kwargs["runtimes_args"]["py_modules"] = py_modules
  
      if isinstance(source, str) and os.path.exists(source):
          extra_search_paths = (extra_search_paths or []) + [
              os.path.dirname(source)
          ]
  
      stream, s_path = parse_config_source(
          source, extra_search_paths=extra_search_paths, **kwargs
      )
      with stream as fp:
          # first load yml with no tag
          no_tag_yml = JAML.load_no_tags(fp)
          if no_tag_yml:
              no_tag_yml.update(**kwargs)
  
              # if there is `override_with` u should make sure that `uses_with` does not remain in the yaml
              def _delitem(
                  obj,
                  key,
              ):
                  value = obj.get(key, None)
                  if value:
                      del obj[key]
                      return
                  for k, v in obj.items():
                      if isinstance(v, dict):
                          _delitem(v, key)
  
              if uses_with is not None:
                  _delitem(no_tag_yml, key="uses_with")
              if uses_metas is not None:
                  _delitem(no_tag_yml, key="uses_metas")
              if uses_requests is not None:
                  _delitem(no_tag_yml, key="uses_requests")
              if uses_dynamic_batching is not None:
                  _delitem(no_tag_yml, key="uses_dynamic_batching")
              cls._override_yml_params(no_tag_yml, "with", uses_with)
              cls._override_yml_params(no_tag_yml, "metas", uses_metas)
              cls._override_yml_params(
                  no_tag_yml, "requests", uses_requests
              )
              cls._override_yml_params(
                  no_tag_yml, "dynamic_batching", uses_dynamic_batching
              )
  
          else:
              raise BadConfigSource(
                  f"can not construct {cls} from an empty {source}. nothing to read from there"
              )
          if substitute:
              # expand variables
              no_tag_yml = JAML.expand_dict(no_tag_yml, context)
  
          if allow_py_modules:
              _extra_search_paths = extra_search_paths or []
              load_py_modules(
                  no_tag_yml,
                  extra_search_paths=(
                      _extra_search_paths + [os.path.dirname(s_path)]
                  )
                  if s_path
                  else _extra_search_paths,
              )
  
          from jina.enums import DeploymentRoleType
          from jina.orchestrate.deployments import Deployment
          from jina.orchestrate.flow.base import Flow
  
          if issubclass(cls, Flow):
              no_tag_yml_copy = copy.copy(no_tag_yml)
              # only needed for Flow
              if no_tag_yml_copy.get("with") is None:
                  no_tag_yml_copy["with"] = {}
              no_tag_yml_copy["with"]["extra_search_paths"] = (
                  no_tag_yml_copy["with"].get("extra_search_paths")
                  or []
              ) + (extra_search_paths or [])
  
              if cls.is_valid_jaml(no_tag_yml_copy):
                  no_tag_yml = no_tag_yml_copy
  
              tag_yml = JAML.unescape(
                  JAML.dump(no_tag_yml),
                  include_unknown_tags=False,
                  jtype_whitelist=("Flow",),
              )
          elif issubclass(cls, Deployment):
              no_tag_yml["with"]["extra_search_paths"] = (
                  no_tag_yml["with"].get("extra_search_paths") or []
              ) + (extra_search_paths or [])
              no_tag_yml["with"]["include_gateway"] = (
                  no_tag_yml["with"].get("include_gateway")
                  or include_gateway
              )
              no_tag_yml["with"]["noblock_on_start"] = noblock_on_start
              no_tag_yml["with"][
                  "deployment_role"
              ] = DeploymentRoleType.DEPLOYMENT
  
              if needs:
                  no_tag_yml["needs"] = list(needs)
  
              tag_yml = JAML.unescape(
                  JAML.dump(no_tag_yml),
                  include_unknown_tags=False,
                  jtype_whitelist=("Deployment",),
              )
          else:
              # revert yaml's tag and load again, this time with substitution
              tag_yml = JAML.unescape(JAML.dump(no_tag_yml))
          # load into object, no more substitute
          obj = JAML.load(
              tag_yml, substitute=False, runtime_args=runtime_args
          )
          if not isinstance(obj, cls):
              raise BadConfigSource(
                  f"Can not construct {cls} object from {source}. Source might be an invalid configuration."
              )
  
          if type(source) == str:
              obj._config_loaded = source
          return obj
  
imports: import copy
  import os
  import re
  import string
  import tempfile
  import warnings
  from types import SimpleNamespace
  from typing import Any, Dict, List, Optional, Set, TextIO, Tuple, Union
  import yaml
  from yaml.constructor import FullConstructor
  from jina.jaml.helper import JinaLoader, JinaResolver, get_jina_loader_with_runtime, load_py_modules, parse_config_source
  from jina.excepts import BadConfigSource
label_desc: Use image node_id 4 for calling the JAMLCompatible obj's underlying member method code with example usage: obj.load_config(cls, source) and returns: obj