Coding, A Data-Driven Guide to Measuring, Visualizing, and Enforcing Cognitive Complexity in Python Projects Using complexipy
In this tutorial, we build a complex end-to-end analytics workflow using complexipy. We start by measuring the complexity directly from raw code strings, then measure the same analysis on individual files and the entire project directory. Along the way, we generate machine-readable reports, normalize them into structured DataFrames, and visualize complex distributions to understand how the depth of decision stacks up across operations. By treating cognitive complexity as a measurable engineering signal, we show how it can be naturally integrated into everyday Python development and quality testing. Check it out FULL CODES here.
!pip -q install complexipy pandas matplotlib
import os
import json
import textwrap
import subprocess
from pathlib import Path
import pandas as pd
import matplotlib.pyplot as plt
from complexipy import code_complexity, file_complexity
print("✅ Installed complexipy and dependencies")We set up the environment by installing the required libraries and importing all the dependencies needed for analysis and visualization. We ensure that the notebook is fully self-contained and ready to run on Google Colab without any external setup. It forms the backbone of all subsequent executions.
snippet = """
def score_orders(orders):
total = 0
for o in orders:
if o.get("valid"):
if o.get("priority"):
if o.get("amount", 0) > 100:
total += 3
else:
total += 2
else:
if o.get("amount", 0) > 100:
total += 2
else:
total += 1
else:
total -= 1
return total
"""
res = code_complexity(snippet)
print("=== Code string complexity ===")
print("Overall complexity:", res.complexity)
print("Functions:")
for f in res.functions:
print(f" - {f.name}: {f.complexity} (lines {f.line_start}-{f.line_end})")We start by analyzing raw Python code strings to understand the complexity of the task-level understanding. We specifically examine how nested conditions and control flow contribute to complexity. It helps us to verify the critical behavior of complexipy before scaling on real files.
root = Path("toy_project")
src = root / "src"
tests = root / "tests"
src.mkdir(parents=True, exist_ok=True)
tests.mkdir(parents=True, exist_ok=True)
(src / "__init__.py").write_text("")
(tests / "__init__.py").write_text("")
(src / "simple.py").write_text(textwrap.dedent("""
def add(a, b):
return a + b
def safe_div(a, b):
if b == 0:
return None
return a / b
""").strip() + "n")
(src / "legacy_adapter.py").write_text(textwrap.dedent("""
def legacy_adapter(x, y):
if x and y:
if x > 0:
if y > 0:
return x + y
else:
return x - y
else:
if y > 0:
return y - x
else:
return -(x + y)
return 0
""").strip() + "n")
(src / "engine.py").write_text(textwrap.dedent("""
def route_event(event):
kind = event.get("kind")
payload = event.get("payload", {})
if kind == "A":
if payload.get("x") and payload.get("y"):
return _handle_a(payload)
return None
elif kind == "B":
if payload.get("flags"):
return _handle_b(payload)
else:
return None
elif kind == "C":
for item in payload.get("items", []):
if item.get("enabled"):
if item.get("mode") == "fast":
_do_fast(item)
else:
_do_safe(item)
return True
else:
return None
def _handle_a(p):
total = 0
for v in p.get("vals", []):
if v > 10:
total += 2
else:
total += 1
return total
def _handle_b(p):
score = 0
for f in p.get("flags", []):
if f == "x":
score += 1
elif f == "y":
score += 2
else:
score -= 1
return score
def _do_fast(item):
return item.get("id")
def _do_safe(item):
if item.get("id") is None:
return None
return item.get("id")
""").strip() + "n")
(tests / "test_engine.py").write_text(textwrap.dedent("""
from src.engine import route_event
def test_route_event_smoke():
assert route_event({"kind": "A", "payload": {"x": 1, "y": 2, "vals": [1, 20]}}) == 3
""").strip() + "n")
print(f"✅ Created project at: {root.resolve()}")We are programming a small but practical Python project with many modules and test files. We intentionally include control flow patterns to make meaningful differences in complexity. Check it out FULL CODES here.
engine_path = src / "engine.py"
file_res = file_complexity(str(engine_path))
print("n=== File complexity (Python API) ===")
print("Path:", file_res.path)
print("File complexity:", file_res.complexity)
for f in file_res.functions:
print(f" - {f.name}: {f.complexity} (lines {f.line_start}-{f.line_end})")
MAX_ALLOWED = 8
def run_complexipy_cli(project_dir: Path, max_allowed: int = 8):
cmd = [
"complexipy",
".",
"--max-complexity-allowed", str(max_allowed),
"--output-json",
"--output-csv",
]
proc = subprocess.run(cmd, cwd=str(project_dir), capture_output=True, text=True)
preferred_csv = project_dir / "complexipy.csv"
preferred_json = project_dir / "complexipy.json"
csv_candidates = []
json_candidates = []
if preferred_csv.exists():
csv_candidates.append(preferred_csv)
if preferred_json.exists():
json_candidates.append(preferred_json)
csv_candidates += list(project_dir.glob("*.csv")) + list(project_dir.glob("**/*.csv"))
json_candidates += list(project_dir.glob("*.json")) + list(project_dir.glob("**/*.json"))
def uniq(paths):
seen = set()
out = []
for p in paths:
p = p.resolve()
if p not in seen and p.is_file():
seen.add(p)
out.append(p)
return out
csv_candidates = uniq(csv_candidates)
json_candidates = uniq(json_candidates)
def pick_best(paths):
if not paths:
return None
paths = sorted(paths, key=lambda p: p.stat().st_mtime, reverse=True)
return paths[0]
return proc.returncode, pick_best(csv_candidates), pick_best(json_candidates)
rc, csv_report, json_report = run_complexipy_cli(root, MAX_ALLOWED)We parse the original source file using the Python API, and use the complexipy CLI throughout the project. We use the CLI from the right directory to generate reliable reports. This step covers the local use of the API with a production-style static analysis workflow.
df = None
if csv_report and csv_report.exists():
df = pd.read_csv(csv_report)
elif json_report and json_report.exists():
data = json.loads(json_report.read_text())
if isinstance(data, list):
df = pd.DataFrame(data)
elif isinstance(data, dict):
if "files" in data and isinstance(data["files"], list):
df = pd.DataFrame(data["files"])
elif "results" in data and isinstance(data["results"], list):
df = pd.DataFrame(data["results"])
else:
df = pd.json_normalize(data)
if df is None:
raise RuntimeError("No report produced")
def explode_functions_table(df_in):
if "functions" in df_in.columns:
tmp = df_in.explode("functions", ignore_index=True)
if tmp["functions"].notna().any() and isinstance(tmp["functions"].dropna().iloc[0], dict):
fn = pd.json_normalize(tmp["functions"])
base = tmp.drop(columns=["functions"])
return pd.concat([base.reset_index(drop=True), fn.reset_index(drop=True)], axis=1)
return tmp
return df_in
fn_df = explode_functions_table(df)
col_map = {}
for c in fn_df.columns:
lc = c.lower()
if lc in ("path", "file", "filename", "module"):
col_map[c] = "path"
if ("function" in lc and "name" in lc) or lc in ("function", "func", "function_name"):
col_map[c] = "function"
if lc == "name" and "function" not in fn_df.columns:
col_map[c] = "function"
if "complexity" in lc and "allowed" not in lc and "max" not in lc:
col_map[c] = "complexity"
if lc in ("line_start", "linestart", "start_line", "startline"):
col_map[c] = "line_start"
if lc in ("line_end", "lineend", "end_line", "endline"):
col_map[c] = "line_end"
fn_df = fn_df.rename(columns=col_map)We load the complex generated reports into pandas and normalize them into a job level table. We handle multiple possible reporting schemas to keep workflows robust. This systematic representation allows us to think about complexity using standard data analysis tools.
if "complexity" in fn_df.columns:
fn_df["complexity"] = pd.to_numeric(fn_df["complexity"], errors="coerce")
plt.figure()
fn_df["complexity"].dropna().plot(kind="hist", bins=20)
plt.title("Cognitive Complexity Distribution (functions)")
plt.xlabel("complexity")
plt.ylabel("count")
plt.show()
def refactor_hints(complexity):
if complexity >= 20:
return [
"Split into smaller pure functions",
"Replace deep nesting with guard clauses",
"Extract complex boolean predicates"
]
if complexity >= 12:
return [
"Extract inner logic into helpers",
"Flatten conditionals",
"Use dispatch tables"
]
if complexity >= 8:
return [
"Reduce nesting",
"Early returns"
]
return ["Acceptable complexity"]
if "complexity" in fn_df.columns and "function" in fn_df.columns:
for _, r in fn_df.sort_values("complexity", ascending=False).head(8).iterrows():
cx = float(r["complexity"]) if pd.notna(r["complexity"]) else None
if cx is None:
continue
print(r["function"], cx, refactor_hints(cx))
print("✅ Tutorial complete.")We visualize the distribution of cognitive difficulties and derive guidance for reconstruction from the numerical limit. We translate complex scores into concrete engineering actions. It closes the loop by connecting measurement directly to retention decisions.
In conclusion, we have presented an efficient, reproducible pipeline for evaluating cognitive complexity in Python projects using complexipy. We showed how we can move from ad-hoc testing to data-driven assumptions about code structure, identify high-risk operations, and provide actionable refactoring guidelines based on calculated constraints. Workflows allow us to think about maintenance ahead of time, use complexity budgets consistently, and change code bases with clarity and confidence, rather than relying solely on intuition.
Check it out FULL CODES here. Also, feel free to follow us Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to Our newspaper. Wait! are you on telegram? now you can join us on telegram too.
