def expand_action()

in elasticsearch_serverless/helpers/actions.py [0:0]
50 lines of code
10 McCabe index (conditional complexity)

def expand_action(data: _TYPE_BULK_ACTION) -> _TYPE_BULK_ACTION_HEADER_AND_BODY:
    """
    From one document or action definition passed in by the user extract the
    action/data lines needed for elasticsearch's
    :meth:`~elasticsearch.Elasticsearch.bulk` api.
    """
    # when given a string, assume user wants to index raw json
    if isinstance(data, (bytes, str)):
        return {"index": {}}, to_bytes(data, "utf-8")

    # make sure we don't alter the action
    data = data.copy()
    op_type: str = data.pop("_op_type", "index")
    action: Dict[str, Any] = {op_type: {}}

    # If '_source' is a dict use it for source
    # otherwise if op_type == 'update' then
    # '_source' should be in the metadata.
    if (
        op_type == "update"
        and "_source" in data
        and not isinstance(data["_source"], Mapping)
    ):
        action[op_type]["_source"] = data.pop("_source")

    for key in (
        "_id",
        "_index",
        "_if_seq_no",
        "_if_primary_term",
        "_parent",
        "_percolate",
        "_retry_on_conflict",
        "_routing",
        "_timestamp",
        "_type",
        "_version",
        "_version_type",
        "if_seq_no",
        "if_primary_term",
        "parent",
        "pipeline",
        "retry_on_conflict",
        "routing",
        "version",
        "version_type",
    ):
        if key in data:
            if key in {
                "_if_seq_no",
                "_if_primary_term",
                "_parent",
                "_retry_on_conflict",
                "_routing",
                "_version",
                "_version_type",
            }:
                action[op_type][key[1:]] = data.pop(key)
            else:
                action[op_type][key] = data.pop(key)

    # no data payload for delete
    if op_type == "delete":
        return action, None

    return action, data.get("_source", data)