original_docstring: Deserializes an operator from a JSON object.
docstring: Deserializes an operator from a JSON object.
docstring_tokens: Deserializes an operator from a JSON object .
code: def deserialize_operator(cls, encoded_op: Dict[str, Any]) -> BaseOperator:
          """Deserializes an operator from a JSON object."""
          from airflow import plugins_manager
  
          plugins_manager.initialize_extra_operators_links_plugins()
  
          if plugins_manager.operator_extra_links is None:
              raise AirflowException("Cnn't load plugins")
          op = SerializedBaseOperator(task_id=encoded_op['task_id'])
  
          # Extra Operator Links defined in Plugins
          op_extra_links_from_plugin = {}
  
          if "label" not in encoded_op:
              # Handle deserialization of old data before the introduction of TaskGroup
              encoded_op["label"] = encoded_op["task_id"]
  
          for ope in plugins_manager.operator_extra_links:
              for operator in ope.operators:
                  if (
                      operator.__name__ == encoded_op["_task_type"]
                      and operator.__module__ == encoded_op["_task_module"]
                  ):
                      op_extra_links_from_plugin.update({ope.name: ope})
  
          # If OperatorLinks are defined in Plugins but not in the Operator that is being Serialized
          # set the Operator links attribute
          # The case for "If OperatorLinks are defined in the operator that is being Serialized"
          # is handled in the deserialization loop where it matches k == "_operator_extra_links"
          if op_extra_links_from_plugin and "_operator_extra_links" not in encoded_op:
              setattr(op, "operator_extra_links", list(op_extra_links_from_plugin.values()))
  
          for k, v in encoded_op.items():
  
              if k == "_downstream_task_ids":
                  v = set(v)
              elif k == "subdag":
                  v = SerializedDAG.deserialize_dag(v)
              elif k in {"retry_delay", "execution_timeout"}:
                  v = cls._deserialize_timedelta(v)
              elif k in encoded_op["template_fields"]:
                  pass
              elif k.endswith("_date"):
                  v = cls._deserialize_datetime(v)
              elif k == "_operator_extra_links":
                  op_predefined_extra_links = cls._deserialize_operator_extra_links(v)
  
                  # If OperatorLinks with the same name exists, Links via Plugin have higher precedence
                  op_predefined_extra_links.update(op_extra_links_from_plugin)
  
                  v = list(op_predefined_extra_links.values())
                  k = "operator_extra_links"
              elif k in cls._decorated_fields or k not in op.get_serialized_fields():
                  v = cls._deserialize(v)
              # else use v as it is
  
              setattr(op, k, v)
  
          for k in op.get_serialized_fields() - encoded_op.keys() - cls._CONSTRUCTOR_PARAMS.keys():
              setattr(op, k, None)
  
          # Set all the template_field to None that were not present in Serialized JSON
          for field in op.template_fields:
              if not hasattr(op, field):
                  setattr(op, field, None)
  
          return op
code_tokens: def deserialize_operator ( cls , encoded_op : Dict [ str , Any ] ) -> BaseOperator : from airflow import plugins_manager plugins_manager . initialize_extra_operators_links_plugins ( ) if plugins_manager . operator_extra_links is None : raise AirflowException ( "Cnn't load plugins" ) op = SerializedBaseOperator ( task_id = encoded_op [ 'task_id' ] ) op_extra_links_from_plugin = { } if "label" not in encoded_op : encoded_op [ "label" ] = encoded_op [ "task_id" ] for ope in plugins_manager . operator_extra_links : for operator in ope . operators : if ( operator . __name__ == encoded_op [ "_task_type" ] and operator . __module__ == encoded_op [ "_task_module" ] ) : op_extra_links_from_plugin . update ( { ope . name : ope } ) if op_extra_links_from_plugin and "_operator_extra_links" not in encoded_op : setattr ( op , "operator_extra_links" , list ( op_extra_links_from_plugin . values ( ) ) ) for k , v in encoded_op . items ( ) : if k == "_downstream_task_ids" : v = set ( v ) elif k == "subdag" : v = SerializedDAG . deserialize_dag ( v ) elif k in { "retry_delay" , "execution_timeout" } : v = cls . _deserialize_timedelta ( v ) elif k in encoded_op [ "template_fields" ] : pass elif k . endswith ( "_date" ) : v = cls . _deserialize_datetime ( v ) elif k == "_operator_extra_links" : op_predefined_extra_links = cls . _deserialize_operator_extra_links ( v ) op_predefined_extra_links . update ( op_extra_links_from_plugin ) v = list ( op_predefined_extra_links . values ( ) ) k = "operator_extra_links" elif k in cls . _decorated_fields or k not in op . get_serialized_fields ( ) : v = cls . _deserialize ( v ) setattr ( op , k , v ) for k in op . get_serialized_fields ( ) - encoded_op . keys ( ) - cls . _CONSTRUCTOR_PARAMS . keys ( ) : setattr ( op , k , None ) for field in op . template_fields : if not hasattr ( op , field ) : setattr ( op , field , None ) return op
short_docstring: Deserializes an operator from a JSON object.
short_docstring_tokens: Deserializes an operator from a JSON object .
comment: """Deserializes an operator from a JSON object."""$SEP$# Extra Operator Links defined in Plugins$SEP$# Handle deserialization of old data before the introduction of TaskGroup$SEP$# If OperatorLinks are defined in Plugins but not in the Operator that is being Serialized$SEP$# set the Operator links attribute$SEP$# The case for "If OperatorLinks are defined in the operator that is being Serialized"$SEP$# is handled in the deserialization loop where it matches k == "_operator_extra_links"$SEP$# If OperatorLinks with the same name exists, Links via Plugin have higher precedence$SEP$# else use v as it is$SEP$# Set all the template_field to None that were not present in Serialized JSON