def create_detection_engine_rules()

in core/kibana.py [0:0]

• 10 lines of code
• 4 McCabe index (conditional complexity)

    def create_detection_engine_rules(self, rules):
        body = "\n".join(json.dumps(rule) for rule in rules)
        files = {"file": ("rules.ndjson", body, "application/octet-stream")}
        url = f"{self.url}/api/detection_engine/rules/_import"
        res = self.session.post(url, files=files, headers={"Content-Type": None})
        res.raise_for_status()
        ret = res.json()
        if ret["errors"]:
            raise ValueError("Could not create rule(s):\n  " + "\n  ".join(str(x) for x in ret["errors"]))
        return ret