elastic / pipes-py - Conditional Complexity

Intro

Conditional complexity (also called cyclomatic complexity) is a term used to measure the complexity of software. The term refers to the number of possible paths through a program function. A higher value ofter means higher maintenance and testing costs (infosecinstitute.com).
Conditional complexity is calculated by counting all conditions in the program that can affect the execution path (e.g. if statement, loops, switches, and/or operators, try and catch blocks...).
Conditional complexity is measured at the unit level (methods, functions...).
Units are classified in four categories based on the measured McCabe index: 1-5 (simple units), 6-10 (medium complex units), 11-25 (complex units), 26+ (very complex units).

Learn more...

Conditional Complexity Overall

There are 71 units with 812 lines of code in units (66.9% of code).

0 very complex units (0 lines of code)
1 complex units (67 lines of code)
6 medium complex units (196 lines of code)
9 simple units (193 lines of code)
55 very simple units (356 lines of code)

Legend:

51+

26-50

11-25

6-10

1-5

Alternative Visuals

3D view of all units

Conditional Complexity per Extension

51+

26-50

11-25

6-10

1-5

Conditional Complexity per Logical Component

primary logical decomposition

51+

26-50

11-25

6-10

1-5

Most Complex Units

Top 50 most complex units

Unit	# lines	McCabe index	# params
def help_message() in core/standalone.py	67	26	1
def handle_param() in core/__init__.py	42	18	5
def run() in core/__init__.py	37	15	6
def handle_param() in core/__init__.py	39	13	5
def main() in core/import.py	24	12	3
def __init__() in hcp/vault/common.py	23	11	1
def __init__() in core/kibana.py	31	11	7
def walk_params() in core/util.py	27	10	1
def get_pipes() in core/__init__.py	25	10	1
def shell_expand() in core/shelllib.py	13	10	1
def get_node() in core/util.py	22	9	6
def run() in core/standalone.py	35	8	1
def bind() in core/__init__.py	23	8	5
def main() in core/export.py	20	7	3
def set_node() in core/util.py	15	6	3
def sync_logger_config() in core/__init__.py	13	6	2
def walk_contexts() in core/util.py	13	5	1
def parse_runtime_arguments() in core/runner.py	18	5	1
def create_detection_engine_rules() in core/kibana.py	10	4	2
def split_path() in core/util.py	9	4	1
def has_node() in core/util.py	10	4	2
def serialize() in core/util.py	9	4	4
def deserialize() in core/util.py	10	4	3
def _repl_env_var() in core/shelllib.py	8	4	1
def find_detection_engine_rules() in core/kibana.py	11	3	3
def receive_state_from_unix_pipe() in core/standalone.py	13	3	2
def get_es_client() in core/util.py	14	3	1
def get_kb_client() in core/util.py	14	3	1
def serialize_yaml() in core/util.py	7	3	2
def deserialize_yaml() in core/util.py	7	3	1
def warn_interactive() in core/util.py	6	3	1
def setup_logging() in core/util.py	15	3	1
def validate_logging_config() in core/__init__.py	5	3	2
def __init__() in core/__init__.py	6	3	5
def _repl_subshell() in core/shelllib.py	9	3	1
def ping() in core/kibana.py	6	2	1
def batched() in core/util.py	4	2	2
def serialize_ndjson() in core/util.py	4	2	2
def deserialize_ndjson() in core/util.py	3	2	1
def is_mutable() in core/util.py	7	2	1
def __call__() in core/__init__.py	9	2	2
def main() in hcp/vault/read.py	13	1	4
def main() in hcp/vault/write.py	13	1	4
def close() in core/kibana.py	2	1	1
def status() in core/kibana.py	5	1	1
def create_siem_index() in core/kibana.py	5	1	1
def get_siem_index() in core/kibana.py	5	1	1
def create_detection_engine_rule() in core/kibana.py	5	1	2
def get_detection_engine_rule() in core/kibana.py	5	1	2
def delete_detection_engine_rule() in core/kibana.py	5	1	2