def __init__(self, args: Union[('Namespace', Dict)], k8s_namespace: Optional[str]=None, k8s_connection_pool: bool=True, k8s_deployments_addresses: Optional[Dict[(str, List[str])]]=None):
    assert (not (hasattr(args, 'external') and args.external))
    if (not validate_uses(args.uses)):
        raise NoContainerizedError(f'Executor "{args.uses}" is not valid to be used in K8s. You need to use a containerized Executor. You may check `jina hub --help` to see how Jina Hub can help you building containerized Executors.')
    self.k8s_namespace = k8s_namespace
    self.k8s_connection_pool = k8s_connection_pool
    self.k8s_deployments_addresses = k8s_deployments_addresses
    self.head_deployment = None
    self.args = copy.copy(args)
    if (k8s_namespace is not None):
        self.args.k8s_namespace = k8s_namespace
    self.args.k8s_connection_pool = k8s_connection_pool
    self.name = self.args.name
    self.deployment_args = self._get_deployment_args(self.args)
    if (self.deployment_args['head_deployment'] is not None):
        self.head_deployment = self._K8sDeployment(name=self.deployment_args['head_deployment'].name, version=get_base_executor_version(), shard_id=None, jina_deployment_name=self.name, common_args=self.args, deployment_args=self.deployment_args['head_deployment'], pod_type=PodRoleType.HEAD, k8s_namespace=self.k8s_namespace, k8s_connection_pool=self.k8s_connection_pool, k8s_deployments_addresses=self.k8s_deployments_addresses)
    self.worker_deployments = []
    deployment_args = self.deployment_args['deployments']
    for (i, args) in enumerate(deployment_args):
        name = (f'{self.name}-{i}' if (len(deployment_args) > 1) else f'{self.name}')
        self.worker_deployments.append(self._K8sDeployment(name=name, version=get_base_executor_version(), shard_id=i, common_args=self.args, deployment_args=args, pod_type=(PodRoleType.WORKER if (name != 'gateway') else PodRoleType.GATEWAY), jina_deployment_name=self.name, k8s_namespace=self.k8s_namespace, k8s_connection_pool=self.k8s_connection_pool, k8s_deployments_addresses=(self.k8s_deployments_addresses if (name == 'gateway') else None)))
def __init__(self, args: Union[('Namespace', Dict)], k8s_namespace: Optional[str]=None, k8s_connection_pool: bool=True, k8s_deployments_addresses: Optional[Dict[(str, List[str])]]=None):
    assert (not (hasattr(args, 'external') and args.external))
    if (not validate_uses(args.uses)):
        raise NoContainerizedError(f'Executor "{args.uses}" is not valid to be used in K8s. You need to use a containerized Executor. You may check `jina hub --help` to see how Jina Hub can help you building containerized Executors.')
    self.k8s_namespace = k8s_namespace
    self.k8s_connection_pool = k8s_connection_pool
    self.k8s_deployments_addresses = k8s_deployments_addresses
    self.head_deployment = None
    self.args = copy.copy(args)
    if (k8s_namespace is not None):
        self.args.k8s_namespace = k8s_namespace
    self.args.k8s_connection_pool = k8s_connection_pool
    self.name = self.args.name
    self.deployment_args = self._get_deployment_args(self.args)
    if (self.deployment_args['head_deployment'] is not None):
        self.head_deployment = self._K8sDeployment(name=self.deployment_args['head_deployment'].name, version=get_base_executor_version(), shard_id=None, jina_deployment_name=self.name, common_args=self.args, deployment_args=self.deployment_args['head_deployment'], pod_type=PodRoleType.HEAD, k8s_namespace=self.k8s_namespace, k8s_connection_pool=self.k8s_connection_pool, k8s_deployments_addresses=self.k8s_deployments_addresses)
    self.worker_deployments = []
    deployment_args = self.deployment_args['deployments']
    for (i, args) in enumerate(deployment_args):
        name = (f'{self.name}-{i}' if (len(deployment_args) > 1) else f'{self.name}')
        self.worker_deployments.append(self._K8sDeployment(name=name, version=get_base_executor_version(), shard_id=i, common_args=self.args, deployment_args=args, pod_type=(PodRoleType.WORKER if (name != 'gateway') else PodRoleType.GATEWAY), jina_deployment_name=self.name, k8s_namespace=self.k8s_namespace, k8s_connection_pool=self.k8s_connection_pool, k8s_deployments_addresses=(self.k8s_deployments_addresses if (name == 'gateway') else None)))