"""Miscellaneous enums used in Jina.To use these enums in YAML config, following the example below:.. highlight:: yaml.. code-block:: yaml chunk_idx: uses: index/chunk.yml parallel: ${{PARALLEL}} parallel_type: !PollingType ANY # or parallel_type: ANY # or parallel_type: any"""fromenumimportEnum,EnumMeta,Flag,IntEnum
[docs]classEnumType(EnumMeta):"""The metaclass for BetterEnum."""def__new__(cls,*args,**kwargs):"""Register a new EnumType :param args: args passed to super() :param kwargs: kwargs passed to super() :return: the registry class """_cls=super().__new__(cls,*args,**kwargs)returncls.register_class(_cls)
[docs]@staticmethoddefregister_class(cls):""" Register the class for dumping loading. :param cls: Target class. :return: Registered class. """reg_cls_set=getattr(cls,'_registered_class',set())ifcls.__name__notinreg_cls_set:reg_cls_set.add(cls.__name__)setattr(cls,'_registered_class',reg_cls_set)fromjina.jamlimportJAMLJAML.register(cls)returncls
[docs]classBetterEnum(IntEnum,metaclass=EnumType):"""The base class of Enum used in Jina."""def__str__(self):returnself.namedef__format__(self,format_spec):# noqa""" override format method for python 3.7 :parameter format_spec: format_spec :return: format using actual value type unless __str__ has been overridden. """# credit python 3.9 : https://github.com/python/cpython/blob/612019e60e3a5340542122dabbc7ce5a27a8c635/Lib/enum.py#L755# fix to enum BetterEnum not correctly formatedstr_overridden=type(self).__str__notin(Enum.__str__,Flag.__str__)ifself._member_type_isobjectorstr_overridden:cls=strval=str(self)# mix-in branchelse:cls=self._member_type_val=self._value_returncls.__format__(val,format_spec)
[docs]@classmethoddeffrom_string(cls,s:str):""" Parse the enum from a string. :param s: string representation of the enum value :return: enum value """try:returncls[s.upper()]exceptKeyError:raiseValueError(f'{s.upper()} is not a valid enum for {cls!r}, must be one of {list(cls)}')
@classmethoddef_to_yaml(cls,representer,data):"""Required by :mod:`pyyaml`. .. note:: In principle, this should inherit from :class:`JAMLCompatible` directly, however, this method is too simple and thus replaced the parent method. :param representer: pyyaml representer :param data: enum value :return: yaml representation """returnrepresenter.represent_scalar('tag:yaml.org,2002:str',str(data),style='"')@classmethoddef_from_yaml(cls,constructor,node):"""Required by :mod:`pyyaml`. .. note:: In principle, this should inherit from :class:`JAMLCompatible` directly, however, this method is too simple and thus replaced the parent method. :param constructor: unused :param node: node to derive the enum value from :return: enum value """returncls.from_string(node.value)
[docs]classPollingType(BetterEnum):"""The enum for representing the parallel type of pods in a deployment."""ANY=1#: one of the shards will receive the messageALL=2#: all shards will receive the message, blocked until all done with the messageALL_ASYNC=3#: (reserved) all replica will receive the message, but any one of them can return, useful in backup@propertydefis_push(self)->bool:""" Check if :class:`PollingType` is using `push` protocol. :return: True if this :class:`PollingType` is using `push` protocol else False. """returnself.value==1@propertydefis_block(self)->bool:""" Check if :class:`PollingType` is using `block` protocol. :return: True if this :class:`PollingType` is requiring `block` protocol else False. """returnself.value==2
[docs]classLogVerbosity(BetterEnum):"""Verbosity level of the logger."""DEBUG=10INFO=20SUCCESS=25WARNING=30ERROR=40CRITICAL=50
[docs]classFlowBuildLevel(BetterEnum):""" The enum for representing a flow's build level. Some :class:`jina.orchestrate.flow.Flow` class functions require certain build level to run. """EMPTY=0#: Nothing is builtGRAPH=1#: The underlying graph is built, you may visualize the flowRUNNING=2#: the graph is started and all deployment are running
[docs]classGatewayProtocolType(BetterEnum):""" Gateway communication protocol """GRPC=0HTTP=1WEBSOCKET=2
[docs]classPodRoleType(BetterEnum):"""The enum of a Pod role."""HEAD=0WORKER=1GATEWAY=2
[docs]classDeploymentRoleType(BetterEnum):"""The enum of a Deploymen role for visualization."""DEPLOYMENT=0JOIN=1INSPECT=2GATEWAY=3INSPECT_AUX_PASS=4JOIN_INSPECT=5@propertydefis_inspect(self)->bool:""" If the role is inspect deployment related. :return: True if the Deployment role is inspect related else False. """returnself.valuein{2,4}
[docs]classRequestType(BetterEnum):"""The enum of Client mode."""DATA=0CONTROL=1
[docs]classCompressAlgo(BetterEnum):""" The enum of Compress algorithms. .. note:: LZ4 requires additional package, to install it use pip install "jina[lz4]" .. seealso:: https://docs.python.org/3/library/archiving.html """NONE=0LZ4=1ZLIB=2GZIP=3BZ2=4LZMA=5
[docs]classOnErrorStrategy(BetterEnum):""" The level of error handling. .. warning:: In theory, all methods below do not 100% guarantee the success execution on the sequel flow. If something is wrong in the upstream, it is hard to CARRY this exception and moving forward without ANY side-effect. """IGNORE=(0#: Ignore it, keep running all Drivers & Executors logics in the sequel flow)SKIP_HANDLE=1#: Skip all Executors in the sequel, only `pre_hook` and `post_hook` are calledTHROW_EARLY=2#: Immediately throw the exception, the sequel flow will not be running at all
[docs]classFlowInspectType(BetterEnum):"""Inspect strategy in the flow."""HANG=0# keep them hanging thereREMOVE=1# remove them in the buildCOLLECT=2# spawn a new deployment and collect them before build@propertydefis_keep(self)->bool:""" Check if the target is inspected. :return: True if the target is inspected else False. """returnself.valuein{0,2}
[docs]classDataInputType(BetterEnum):"""Data input type in the request generator."""AUTO=0# auto inference the input type from data (!WARN: could be slow as it relies on try-execept)DOCUMENT=1# the input is a full documentCONTENT=2# the input is just the content of the documentDICT=3# the input is a dictionary representing a Document, needed while pydantic model not available
[docs]classWebsocketSubProtocols(str,Enum):"""Subprotocol supported with Websocket Gateway"""JSON='json'BYTES='bytes'
[docs]defreplace_enum_to_str(obj):""" Transform BetterEnum type into string. :param obj: Target obj. :return: Transformed obj with string type values. """fork,vinobj.items():ifisinstance(v,dict):obj[k]=replace_enum_to_str(v)elifisinstance(v,BetterEnum):obj[k]=str(v)returnobj