def __init__()

in eventdata/parameter_sources/elasticlogs_bulk_source.py [0:0]

• 32 lines of code
• 7 McCabe index (conditional complexity)

    def __init__(self, track, params, **kwargs):
        self.infinite = False
        self.orig_args = [track, params, kwargs]
        self._indices = track.indices
        self._params = params
        # we could also do `kwargs.get("random_event", RandomEvent(params))` but that would call the constructor eagerly
        # which we want to avoid because this can cause significant overhead.
        if "random_event" in kwargs:
            self._randomevent = kwargs["random_event"]
        else:
            self._randomevent = RandomEvent(params)

        self._bulk_size = params["bulk-size"]
        self.seq_id = 0

        self._id_type = params.get("id_type", "auto")
        if self._id_type not in ["auto", "seq"]:
            raise AssertionError("The value [{}] is invalid for the parameter [id_type]".format(self._id_type))

        if self._id_type == "seq":
            self._id_seq_probability = float(params.get("id_seq_probability", 0.0))
            self._low_id_bias = str(params.get('id_seq_low_id_bias', False)).lower() == "true"
            if self._low_id_bias:
                logger.info("Will use low id bias for updates")
            else:
                logger.info("Will use uniform distribution for updates")

        self._default_index = False
        if "index" not in params.keys():
            index_name = self._indices[0].name
            if len(self._indices) > 1:
                logger.debug("[bulk] More than one index specified in track configuration. Will use the first one ({})".format(index_name))
            else:
                logger.debug("[bulk] Using index specified in track configuration ({})".format(index_name))

            self._params["index"] = index_name
            self._default_index = True
        else:
            logger.debug("[bulk] Index pattern specified in parameters ({}) will be used".format(params["index"]))