Python Refactor

Python Refactor

Transform complex Python into clear, maintainable code while preserving correctness. Phased workflow with safety-by-design and continuous validation. For deep references (anti-patterns, OOP principles, cognitive complexity, regression prevention), see the references/ directory.

When to invoke

• Explicit "human", "readable", "maintainable", "clean", or "refactor" request
• Code review flags comprehension or maintainability issues
• Legacy code modernization
• Onboarding / educational contexts
• Complexity metrics exceed thresholds
• Red flags: file > 500 lines with scattered functions and global state, multiple global statements, no clear module/class organization, configuration mixed with business logic

Do NOT invoke when

• Code is performance-critical and profiling shows perf optimization is needed first
• Code is scheduled for deletion or replacement
• External dependencies require upstream contributions instead
• User explicitly requested perf optimization over readability

Core principles (priority order)

1. Prefer structured OOP for complex code -- shared state, multiple concerns, scattered globals = restructure into classes/modules. (But: simple modules with pure functions, click/argparse CLIs, and functional pipelines DON'T need to be forced into classes.) 2. Clarity over cleverness -- explicit beats implicit 3. Preserve correctness -- all tests pass, behavior identical 4. Single Responsibility -- one thing per class/function (SOLID) 5. Self-documenting structure -- code = what, comments = why 6. Progressive disclosure -- reveal complexity in layers 7. Reasonable performance -- never sacrifice >2× without explicit approval

Hard constraints

• SAFETY BY DESIGN -- mandatory migration checklists for destructive changes. CREATE → SEARCH → MIGRATE → VERIFY → only then REMOVE. NEVER remove before 100% migration verified.
• STATIC ANALYSIS FIRST -- flake8 --select=F821,E0602 (or ruff check --select=F821) BEFORE tests. Catches NameErrors immediately.
• PRESERVE BEHAVIOR -- all existing tests pass after.
• NO PERF REGRESSION -- never degrade > 2× without explicit approval.
• NO API CHANGES -- public APIs unchanged unless explicitly requested + documented.
• NO OVER-ENGINEERING -- simple stays simple.
• NO MAGIC -- no framework magic, no metaprogramming unless absolutely necessary.
• VALIDATE CONTINUOUSLY -- static analysis + tests after each logical change.

Regression prevention (MANDATORY)

Refactoring must NEVER introduce regressions. Read references/REGRESSION_PREVENTION.md before any session.

Before each session:

• Test suite passes 100%
• Coverage ≥ 80% on target code (write tests FIRST if not)
• Golden outputs captured for critical edge cases
• Static analysis baseline saved

After EACH micro-change (not at the end -- every single one):

• flake8 --select=F821,E999 → 0 errors
• pytest -x → all passing
• Spot check 1 edge case for unchanged behavior

If ANY check fails: STOP → REVERT → ANALYZE → FIX APPROACH → RETRY.

ANY REGRESSION = TOTAL FAILURE.

Workflow (4 phases)

Phase 1: Analysis

1. Read the entire codebase section. 2. Identify readability issues using references/anti-patterns.md (script-like/global-state, God Objects, nested conditionals, long functions, magic numbers, cryptic names). 3. Assess architecture against references/oop_principles.md (proper classes/modules, encapsulated state, separated responsibilities, SOLID, DI vs hard-coded deps). 4. Measure current metrics with scripts/measure_complexity.py or scripts/analyze_multi_metrics.py. 5. Run linting analysis (see Tooling below). 6. Check test coverage; identify gaps to fill BEFORE refactoring. 7. Document with assets/templates/analysis_template.md.

Output: prioritized list of issues by impact and risk.

Phase 2: Planning

1. Classify each change:

• Non-destructive (rename, docs, type hints) → low risk
• Destructive (remove globals, delete functions, replace APIs) → high risk

2. For DESTRUCTIVE changes -- migration plan is MANDATORY:

• Search ALL usages of each element to be removed
• Document every usage (file, line, type)
• No complete migration plan = cannot proceed with the destructive change

3. Risk assessment per change (Low/Medium/High) 4. Dependency map -- what depends on this code? 5. Test strategy -- what tests are needed? what might break? 6. Order changes safest → riskiest 7. Document expected metric improvements

Output: refactoring plan, sequenced changes, migration plans, test strategy, rollback plan.

Phase 3: Execution

Non-destructive (safe anytime)

1. Rename for clarity 2. Extract magic numbers/strings to named constants 3. Add/improve docs and type hints 4. Add guard clauses to reduce nesting

Destructive (STRICT PROTOCOL)

1. CREATE new structure (no removal) -- write new classes/functions + tests 2. SEARCH for ALL usages of the element being removed 3. CREATE migration checklist documenting every found usage 4. MIGRATE one usage at a time, checking off the list, running static analysis + tests after each 5. VERIFY complete migration -- re-run searches, should find zero old references 6. REMOVE old code only after 100% migration verified

Execution rules

• NEVER skip the migration checklist for destructive changes
• Run static analysis BEFORE tests
• One pattern at a time -- never mix multiple refactoring patterns in a single change
• Atomic commits -- each migration step gets its own commit
• Stop on ANY error (static analysis OR test failure) → immediate fix/revert

Recommended order

1. Transform script-like code to proper architecture (references/examples/script_to_oop_transformation.md) 2. Rename for clarity 3. Extract magic numbers/strings to constants/enums 4. Improve docs + type hints 5. Extract methods to reduce function length 6. Simplify conditionals with guard clauses 7. Reduce nesting depth 8. Final review: separation of concerns

Phase 4: Validation

1. Static analysis FIRST:

   flake8 <file> --select=F821,E0602          # undefined names/variables -- MUST be 0
   flake8 <file> --select=F401                # unused imports
   flake8 <file>                              # full quality check

2. Full test suite → 100% pass required. 3. Architecture validation: global state eliminated/encapsulated, proper modules/classes, separated responsibilities, SOLID compliance. 4. Before/after metrics with scripts/measure_complexity.py or scripts/analyze_multi_metrics.py. 5. Performance regression check with scripts/benchmark_changes.py for hot paths. 6. Summary report using assets/templates/summary_template.md. 7. Flag for human review if: perf degraded > 10%, public API signatures changed, test coverage decreased, significant architectural changes.

Refactoring patterns (catalog summary)

Full catalog with examples in references/patterns.md. Key patterns:

• Guard Clauses -- early returns instead of nested conditionals
• Extract Method -- split large functions into focused units (resets the nesting counter -- most powerful for cognitive complexity)
• Dictionary Dispatch -- replace if-elif chains with lookup tables
• Match Statement (Py 3.10+) -- counts as +1 total, not per branch
• Named Boolean Conditions -- extract complex booleans into named variables
• Encapsulate Global State -- move globals into classes with proper encapsulation
• Group Related Functions -- organize scattered functions into classes by responsibility
• Create Domain Models -- replace primitive dicts with dataclasses + enums
• Apply Dependency Injection -- replace hard-coded deps with injected ones

For cognitive complexity calculation rules and reduction strategies, see references/cognitive_complexity_guide.md.

Naming conventions

• Variables: descriptive, booleans as is_active / has_permission / can_edit, collections as plurals
• Functions: verb + object (calculate_total, validate_email); boolean queries as is_valid() / has_items()
• Constants: UPPERCASE_WITH_UNDERSCORES; replace magic numbers/strings
• Classes: PascalCase nouns (UserAccount, PaymentProcessor)

Anti-patterns to fix (priority order)

Full catalog: references/anti-patterns.md.

• Critical: script-like / procedural code with global state; God Object / God Class
• High: complex nested conditionals (> 3 levels), long functions (> 30 lines), magic numbers, cryptic names, missing type hints, missing docstrings
• Medium: duplicate code, primitive obsession, long parameter lists (> 5)
• Low: inconsistent naming, redundant comments, unused imports

Tooling

Primary stack: Ruff + Complexipy (recommended for new projects)

uv tool install ruff complexipy radon wily

ruff check src/                                 # fast linting (Rust, replaces flake8+plugins)
complexipy src/ --max-complexity-allowed 15     # cognitive complexity (Rust)
radon mi src/ -s                                # maintainability index

Full configuration (pyproject.toml, pre-commit, GitHub Actions): references/cognitive_complexity_guide.md.

Alternative: flake8 + curated plugins

For projects already on flake8, see references/flake8_plugins_guide.md (curated 16-plugin selector list).

Multi-metric analysis

scripts/analyze_multi_metrics.py combines complexipy + radon + maintainability index in a single report.

Metric targets

• Cyclomatic complexity: < 10 per function (warning 15, error 20)
• Cognitive complexity: < 15 per function (SonarQube default; warning 20)
• Function length: < 30 lines (warning 50)
• Nesting depth: ≤ 3 levels
• Docstring coverage: > 80% for public functions
• Type-hint coverage: > 90% for public APIs

Historical tracking with Wily

Trends matter, not just thresholds. Setup + CI integration: references/cognitive_complexity_guide.md.

Common refactoring mistakes

Full guide: references/REGRESSION_PREVENTION.md. Key traps:

1. Incomplete migration -- removing old code before ALL usages migrated (causes NameErrors). 2. Partial pattern application -- applying refactoring to some functions but not others. 3. Breaking public APIs -- changing signatures used by external code. 4. Assuming tests cover everything -- tests pass but runtime errors occur (run static analysis!).

When to reach for which tool

• clean-code (cross-language plugin) -- multi-language cosmetic cleanup; renames local vars, improves comments, simplifies structure. Lowest regression risk. Use for "make this readable", "clean up naming."
• python-refactor (this skill) -- Python-only deep restructuring. OOP transformation, SOLID, complexity metrics, migration checklists, benchmark validation. Use for "refactor this module", "reduce complexity", "transform to OOP."

Escalation path: clean-code → python-refactor (safest to most thorough).

Integration

• python-tdd -- set up tests before refactoring, validate coverage after
• python-performance-optimization -- deep profiling before/after
• python-packaging -- handle pyproject.toml + distribution if refactoring a library
• uv-package-manager -- uv run ruff, uv run complexipy for tool execution
• async-python-patterns -- reference async patterns when refactoring async code

When NOT to refactor

Perf-critical optimized code (profile first), code scheduled for deletion, external deps (contribute upstream), stable legacy code nobody needs to modify.

Limitations

Cannot improve algorithmic complexity (that's an algorithm change). Cannot add domain knowledge not in code/comments. Cannot guarantee correctness without tests. Style preferences vary -- adjust to team conventions.

Examples

references/examples/:

• script_to_oop_transformation.md -- script → clean OOP architecture (flagship case study)
• python_complexity_reduction.md -- nested conditionals and long functions
• typescript_naming_improvements.md -- naming patterns (cross-language reference)

Success criteria

1. Zero regressions -- all tests pass, behavior unchanged 2. Golden master match for documented critical cases 3. Complexity metrics improved (documented in summary) 4. No perf regression > 10% (or explicit approval) 5. Documentation coverage improved 6. Code easier for humans to understand 7. No new security vulnerabilities 8. Atomic, well-documented git history 9. Wily trend -- complexity not increased vs previous commit 10. Static analysis shows improvement

Refactoring Analysis

File: path/to/file.ext Date: YYYY-MM-DD Analyst: [Your Name/Tool]

---

Executive Summary

[Brief 2-3 sentence overview of the code quality and recommended refactoring approach]

---

Current State Metrics

Complexity Metrics

Documentation Metrics

---

Identified Issues

High Priority

1. [Issue Name] - Line XXX

• Anti-Pattern: [Name from anti-patterns.md]
• Impact: [Readability/Maintainability/Performance]
• Description: [What makes this problematic]
• Recommended Fix: [Pattern from patterns.md]

2. [Issue Name] - Line XXX

• Anti-Pattern: [Name]
• Impact: [Impact type]
• Description: [Details]
• Recommended Fix: [Pattern]

Medium Priority

3. [Issue Name] - Line XXX

• Anti-Pattern: [Name]
• Impact: [Impact type]
• Description: [Details]
• Recommended Fix: [Pattern]

Low Priority

4. [Issue Name] - Line XXX

• Anti-Pattern: [Name]
• Impact: [Impact type]
• Description: [Details]
• Recommended Fix: [Pattern]

---

Risk Assessment

Refactoring Complexity: [Low/Medium/High]

Factors:

• Test Coverage: [Percentage] - [Good/Poor]
• Dependencies: [Number] external dependencies affected
• Public API Changes: [None/Minor/Major]
• Business Logic Complexity: [Low/Medium/High]

Risk Level by Issue

---

Recommended Refactoring Plan

Phase 1: Safe Changes (Low Risk)

Estimated Time: [Duration]

1. Rename variables and functions for clarity

• Lines: XXX, YYY, ZZZ
• Risk: Very Low
• Impact: Readability improved

2. Extract magic numbers to constants

• Lines: XXX, YYY
• Risk: Very Low
• Impact: Maintainability improved

3. Add/improve documentation

• All public functions
• Risk: None
• Impact: Documentation coverage to >80%

Phase 2: Structural Changes (Medium Risk)

Estimated Time: [Duration]

4. Apply guard clauses to reduce nesting

• Function: function_name() (Line XXX)
• Risk: Low (preserves behavior)
• Impact: Nesting reduced from X to Y levels

5. Extract methods to reduce function length

• Function: long_function() (Line XXX)
• Extract: 3-4 helper functions
• Risk: Medium (requires careful testing)
• Impact: Complexity reduced from XX to Y

Phase 3: Advanced Refactoring (Higher Risk)

Estimated Time: [Duration]

6. Separate concerns into distinct layers

• Mixed data/logic/presentation
• Risk: Medium-High
• Impact: Major architectural improvement

7. Replace primitive obsession with domain objects

• Risk: High (API changes)
• Impact: Type safety and validation improved

---

Expected Outcomes

Metrics Improvement Projections

Qualitative Improvements

• Readability: [Description of improvement]
• Maintainability: [Description of improvement]
• Testability: [Description of improvement]
• Onboarding: [Description of improvement]

---

Test Coverage Assessment

Current Coverage

• Overall Coverage: XX%
• Critical Paths Covered: Yes/No
• Edge Cases Tested: Yes/No
• Integration Tests: Yes/No

Recommended Test Additions

Before refactoring, add tests for:

1. [Test Description] - Current gap in coverage 2. [Test Description] - Needed before structural changes 3. [Test Description] - Critical path not tested

---

Dependencies Analysis

Internal Dependencies

Functions/modules that depend on code being refactored:

1. [Module/Function Name] - [Dependency Type]

• Impact: [Low/Medium/High]
• Required Changes: [Description]

External Dependencies

Third-party code or APIs affected:

1. [Library/API Name]

• Impact: [Low/Medium/High]
• Required Changes: [Description]

---

Alternatives Considered

Option A: [Approach Name]

• Pros: [List]
• Cons: [List]
• Effort: [Low/Medium/High]
• Recommendation: [Chosen/Not Chosen - Why]

Option B: [Approach Name]

• Pros: [List]
• Cons: [List]
• Effort: [Low/Medium/High]
• Recommendation: [Chosen/Not Chosen - Why]

---

Sign-Off Checklist

Before proceeding with refactoring:

• [ ] All high-priority issues identified
• [ ] Risk assessment completed
• [ ] Test coverage adequate (>80% recommended)
• [ ] Refactoring plan reviewed by team
• [ ] Time estimate approved
• [ ] Dependencies documented
• [ ] Rollback plan defined

---

Next Steps

1. Immediate: [Action item] 2. Short-term: [Action item] 3. Long-term: [Action item]

---

Notes

[Any additional context, concerns, or observations]

Flake8 Code Quality Report

Target: path/to/code Date: YYYY-MM-DD Analysis Type: [Before Refactoring / After Refactoring]

---

Executive Summary

Status: [PASSED / FAILED] Total Issues: XXX Risk Level: [Low / Medium / High]

[2-3 sentence summary of overall code quality based on flake8 analysis]

---

Installed Plugins

ESSENTIAL (Must-Have - Highest Impact)

✓/✗ flake8-bugbear: Finds likely bugs and design problems (B codes) ✓/✗ flake8-simplify: Suggests simpler, clearer code (SIM codes) ✓/✗ flake8-cognitive-complexity: Measures cognitive load (CCR codes) ✓/✗ pep8-naming: Enforces clear naming conventions (N codes) ✓/✗ flake8-docstrings: Ensures documentation (D codes)

RECOMMENDED (Strong Readability Impact)

✓/✗ flake8-comprehensions: Cleaner comprehensions (C4 codes) ✓/✗ flake8-expression-complexity: Prevents complex expressions (ECE codes) ✓/✗ flake8-functions: Simpler function signatures (CFQ codes) ✓/✗ flake8-variables-names: Better variable naming (VNE codes) ✓/✗ tryceratops: Clean exception handling (TC codes)

OPTIONAL (Nice to Have)

✓/✗ flake8-builtins: Prevents shadowing built-ins (A codes) ✓/✗ flake8-eradicate: Finds commented-out code (E800 codes) ✓/✗ flake8-unused-arguments: Flags unused parameters (U codes) ✓/✗ flake8-annotations: Validates type hints (ANN codes) ✓/✗ pydoclint: Complete docstrings (DOC codes) ✓/✗ flake8-spellcheck: Catches typos (SC codes)

Missing Plugins

Missing ESSENTIAL plugins (install these first): [List missing essential plugins]

Installation Command:

pip install flake8 [missing-essential-plugins]

Missing RECOMMENDED plugins: [List missing recommended plugins]

Installation Command:

pip install [missing-recommended-plugins]

Install all 16 plugins (full suite):

pip install flake8 flake8-bugbear flake8-simplify \
    flake8-cognitive-complexity pep8-naming flake8-docstrings \
    flake8-comprehensions flake8-expression-complexity \
    flake8-functions flake8-variables-names tryceratops \
    flake8-builtins flake8-eradicate flake8-unused-arguments \
    flake8-annotations pydoclint flake8-spellcheck

---

Issues by Severity

Severity Distribution

High   [████████░░] XX%
Medium [██████░░░░] XX%
Low    [███░░░░░░░] XX%

---

Issues by Category

---

Top Issue Types

---

Files with Most Issues

---

High Severity Issues

Potential Bugs (Bugbear - B codes)

B001 - Line XXX: Do not use bare except:

# File: path/to/file.py:XXX
try:
    dangerous_operation()
except:  # Too broad!
    pass

Impact: May catch and hide critical errors like KeyboardInterrupt Fix: Catch specific exceptions or use except Exception:

---

B006 - Line XXX: Do not use mutable data structures as default arguments

# File: path/to/file.py:XXX
def append_to_list(item, lst=[]):  # Dangerous!
    lst.append(item)
    return lst

Impact: Shared mutable default causes unexpected behavior Fix: Use lst=None and initialize inside function

---

PyFlakes Errors (F codes)

F401 - Lines XXX, YYY, ZZZ: Module imported but unused Impact: Clutters namespace, suggests dead code Fix: Remove unused imports or add to __all__

F841 - Lines XXX, YYY: Local variable assigned but never used Impact: Dead code, possible logic error Fix: Remove assignment or use the variable

---

Runtime Errors (E9 codes)

[List any E9xx codes - these are critical syntax/runtime errors]

---

Medium Severity Issues

Complexity (C codes)

C901 - Function too complex

Function: complex_function() (Line XXX)
Cyclomatic Complexity: XX (threshold: 10)

Impact: Hard to understand and test Recommendation: Apply Extract Method pattern (see patterns.md)

---

Naming Conventions (N codes)

N802 - Lines XXX, YYY: Function name should be lowercase

def CalculateTotal():  # Should be: calculate_total()
    pass

N803 - Lines XXX: Argument name should be lowercase N806 - Lines XXX: Variable should be lowercase

---

Annotations (A codes)

ANN201 - Lines XXX, YYY, ZZZ: Missing return type annotation

def process_data(data: str):  # Missing -> ReturnType
    return data.upper()

Impact: Reduced type safety, no IDE support Fix: Add return type annotation

---

Low Severity Issues

Style Errors (E/W codes)

E501 - Line too long (XX > 88 characters) E302 - Expected 2 blank lines, found 1 E303 - Too many blank lines W503 - Line break before binary operator

[List most common style violations with counts]

---

Docstring Issues (D codes)

D100 - Missing docstring in public module D103 - Missing docstring in public function D107 - Missing docstring in __init__

Functions Missing Docstrings: 1. function_name() - Line XXX 2. another_function() - Line YYY 3. [...]

---

Simplification Opportunities (S codes)

SIM105 - Lines XXX: Use contextlib.suppress() instead of try/except/pass SIM108 - Lines XXX: Use ternary operator instead of if/else SIM118 - Lines XXX: Use key in dict instead of key in dict.keys()

---

Detailed Issue List

Critical Issues (Must Fix)

1. File: path/to/file.py:XXX Code: BXXX Severity: High Message: [Full error message] Recommendation: [How to fix]

2. [Continue with all high severity issues...]

---

Important Issues (Should Fix)

[Medium severity issues with line numbers and recommendations]

---

Minor Issues (Nice to Fix)

[Low severity issues - can list first 50]

---

Recommendations

Immediate Actions (High Priority)

1. Fix all Bugbear issues (B codes) - These indicate likely bugs

• Estimated effort: [Duration]
• Files affected: [X files]

2. Address PyFlakes errors (F codes) - Dead code and logic issues

• Estimated effort: [Duration]
• Files affected: [X files]

3. Reduce function complexity (C901) - Apply Extract Method pattern

• Estimated effort: [Duration]
• Functions affected: [X functions]

Short-Term Actions (Medium Priority)

4. Add missing docstrings (D codes) - Improve documentation

• Target: >80% coverage
• Estimated effort: [Duration]

5. Add type annotations (ANN codes) - Improve type safety

• Target: >90% coverage
• Estimated effort: [Duration]

6. Fix naming conventions (N codes) - Follow PEP 8

• Estimated effort: [Duration]

Long-Term Actions (Low Priority)

7. Address style violations (E/W codes) - Improve consistency

• Consider using Black formatter
• Estimated effort: [Duration]

8. Apply simplifications (S codes) - Make code more Pythonic

• Estimated effort: [Duration]

---

Comparison with Targets

---

Integration with Refactoring Workflow

Issues Aligned with Anti-Patterns

[Map flake8 issues to anti-patterns.md categories]

Complex Nested Conditionals ← C901 (High complexity) God Functions ← C901 (Complexity), E501 (Too long) Magic Numbers ← No direct flake8 code (manual review needed) Cryptic Names ← N803, N806 (Naming violations) Missing Docstrings ← D100-D107 (Docstring codes) Missing Type Hints ← ANN codes Unclear Error Handling* ← B001 (Bare except)

Recommended Patterns to Apply

Based on flake8 results, apply these patterns from patterns.md:

1. Extract Method - For C901 complexity violations 2. Guard Clauses - For deeply nested code (manual + C901) 3. Add Documentation - For D codes 4. Add Type Hints - For ANN codes 5. Improve Naming - For N codes 6. Simplify Code - For S codes

---

Next Steps

1. Review High Severity Issues - Fix all B, F, and E9 codes first 2. Run Refactoring Workflow - Use patterns from patterns.md 3. Re-run flake8 - Measure improvements 4. Compare Reports - Use compare_flake8_reports.py

Command to Re-run:

python scripts/analyze_with_flake8.py <target> --output after_flake8.json --html after_flake8.html

Command to Compare:

python scripts/compare_flake8_reports.py before_flake8.json after_flake8.json --html comparison.html

---

Configuration Used

[flake8]
max-line-length = 88
max-complexity = 10
max-cognitive-complexity = 10
max-expression-complexity = 7
docstring-convention = google

Configuration File: assets/.flake8 [Copy this file to your project root for consistent analysis]

---

Additional Resources

• Flake8 Documentation: https://flake8.pycqa.org/
• Plugin Documentation:
• flake8-bugbear: https://github.com/PyCQA/flake8-bugbear
• flake8-docstrings: https://github.com/PyCQA/flake8-docstrings
• flake8-simplify: https://github.com/MartinThoma/flake8-simplify
• Anti-Patterns Reference: references/anti-patterns.md
• Refactoring Patterns: references/patterns.md

---

Report Metadata

Generated By: python-refactor skill Skill Version: 1.1.0 Timestamp: YYYY-MM-DD HH:MM:SS Flake8 Version: [Version from --version] Python Version: [Version]

---

Notes

[Any additional observations, context, or recommendations specific to this codebase]

Refactoring Summary

File: path/to/file.ext Date: YYYY-MM-DD Refactored By: [Your Name] Duration: [Time Spent] Risk Level: [Low/Medium/High]

---

Overview

[2-3 sentence summary of what was refactored and why]

---

Changes Made

1. [Change Category - e.g., "Extracted Methods for Complexity Reduction"]

Lines Affected: XXX-YYY

Description: [Detailed description of what was changed and why]

Rationale: [Business/technical justification for this change]

Pattern Applied: [Reference to pattern from patterns.md]

Risk Level: Low/Medium/High

Before:

[Code snippet before refactoring]

After:

[Code snippet after refactoring]

---

2. [Change Category]

Lines Affected: XXX-YYY

Description: [Detailed description]

Rationale: [Justification]

Pattern Applied: [Pattern reference]

Risk Level: Low/Medium/High

Before:

[Code snippet before]

After:

[Code snippet after]

---

3. [Change Category]

Lines Affected: XXX-YYY

Description: [Detailed description]

Rationale: [Justification]

Pattern Applied: [Pattern reference]

Risk Level: Low/Medium/High

---

Metrics Improvement

Complexity Metrics

Documentation Metrics

Code Size Metrics

---

Validation Results

Test Results

Test Suite: [Name] Tests Run: XXX Tests Passed: XXX ✓ Tests Failed: 0 Test Coverage: XX% (unchanged/improved)

Details:

• All existing tests pass without modification ✓
• No regression detected ✓
• Test execution time: [Duration] (±X% change)

Performance Validation

Benchmark Results:

Performance Assessment:

• No function shows >10% regression ✓
• Overall performance impact: [Negligible/Positive/Within acceptable range]

Code Quality Validation

Linter Results:

• Warnings Before: XX
• Warnings After: Y
• Reduction: Z warnings fixed

Type Checker Results:

• Type Errors Before: XX
• Type Errors After: 0 ✓
• New Type Coverage: YY%

---

Risk Assessment

Overall Risk: [Low/Medium/High]

Risk Factors:

✓ Low Risk Factors:

• All tests passing
• No public API changes
• Performance within acceptable range
• Changes are mechanical refactorings
• Full test coverage maintained

⚠ Medium Risk Factors:

• [If any, list here]

✗ High Risk Factors:

• [If any, list here]

Risk Mitigation

[If medium/high risk, describe mitigation steps taken]

---

Human Review Recommendations

Review Required: [Yes/No]

Review Focus Areas: 1. [Specific area if review needed - e.g., "Verify business logic in extracted payment_calculator() function"] 2. [Another area if applicable] 3. [Another area if applicable]

Review Checklist:

• [ ] Verify all tests pass
• [ ] Review extracted functions for correctness
• [ ] Confirm no unintended behavior changes
• [ ] Validate performance is acceptable
• [ ] Check documentation accuracy
• [ ] Verify error handling remains robust

---

Breaking Changes

API Changes: [None/Listed Below]

[If any breaking changes, list them here with migration guide]

Migration Guide: [If applicable, provide guidance for updating calling code]

---

Lessons Learned

What Went Well

1. [Positive outcome or discovery] 2. [Another positive aspect]

Challenges Encountered

1. [Challenge faced and how it was resolved] 2. [Another challenge]

Recommendations for Future

1. [Suggestion for similar refactorings] 2. [Process improvement suggestion]

---

Related Refactorings

Recommended Follow-Ups

Based on this refactoring, these related improvements are recommended:

1. [Related File/Function]

• Issue: [Similar problem]
• Estimated Effort: [Duration]
• Priority: [Low/Medium/High]

2. [Another Related Area]

• Issue: [Description]
• Estimated Effort: [Duration]
• Priority: [Low/Medium/High]

---

Commit Information

Branch: feature/refactor-[description] Commits: X commit(s)

Commit Messages:

• [commit-hash] - [Commit message]
• [commit-hash] - [Commit message]

Pull Request: #XXX (if applicable)

---

Sign-Off

Refactored By: [Your Name] Date: YYYY-MM-DD

Reviewed By: [Reviewer Name] (if applicable) Review Date: YYYY-MM-DD

Approved For Merge: [Yes/Pending/No]

---

Appendix

Files Modified

• path/to/file1.ext - [Description of changes]
• path/to/file2.ext - [Description of changes]

References

• [Link to relevant documentation]
• [Link to related issues/PRs]
• [Link to patterns used]

Additional Notes

[Any other relevant information, context, or observations]

Code Anti-Patterns Reference

This document catalogs common code anti-patterns that harm readability and maintainability, with detection criteria and refactoring guidance.

Table of Contents

1. High-Priority Anti-Patterns 2. Medium-Priority Anti-Patterns 3. Low-Priority Anti-Patterns

---

High-Priority Anti-Patterns

These anti-patterns significantly harm code comprehension and should be fixed first.

1. Script-Like / Procedural Code (Spaghetti Code)

Problem: Code organized as a script with scattered functions, global state, and no clear structure instead of proper OOP architecture.

Detection:

• Global variables for state management
• Functions scattered without cohesion
• No classes or poorly designed classes
• Everything in one file
• No clear separation of concerns
• Direct dependencies instead of dependency injection

Impact:

• Impossible to test in isolation
• Global state causes unpredictable behavior
• Difficult to reuse components
• No clear boundaries or responsibilities
• Changes in one area break unrelated functionality
• Cannot scale or extend easily

Example:

# BAD: Script-like code with global state
users_cache = {}  # Global state!
API_URL = "https://api.example.com"

def fetch_user(user_id):
    global users_cache  # Modifying global state
    if user_id in users_cache:
        return users_cache[user_id]

    response = requests.get(f"{API_URL}/users/{user_id}")
    data = response.json()
    users_cache[user_id] = data
    return data

def process_user(user_id):
    user = fetch_user(user_id)
    # ... scattered processing logic
    return processed_data

# Executing at module level
if __name__ == "__main__":
    result = process_user(123)

Fix: Transform to OOP architecture - see oop_principles.md

Good Example:

# GOOD: OOP-based with clear structure
# models/user.py
@dataclass
class User:
    id: int
    name: str
    email: str

# repositories/user_repository.py
class UserRepository:
    def __init__(self, api_client: APIClient):
        self._api_client = api_client
        self._cache: dict[int, User] = {}

    def get_by_id(self, user_id: int) -> Optional[User]:
        if user_id in self._cache:
            return self._cache[user_id]

        data = self._api_client.get(f"/users/{user_id}")
        if data:
            user = User(**data)
            self._cache[user_id] = user
            return user
        return None

# services/user_service.py
class UserService:
    def __init__(self, user_repository: UserRepository):
        self._user_repo = user_repository

    def process_user(self, user_id: int) -> dict:
        user = self._user_repo.get_by_id(user_id)
        if not user:
            return {'error': 'User not found'}
        return self._process(user)

    def _process(self, user: User) -> dict:
        # Processing logic
        pass

Related Patterns: Repository Pattern, Service Layer, Dependency Injection Related Reference: See references/oop_principles.md for complete guide

---

2. God Object / God Class

Problem: One class responsible for too many unrelated things.

Detection:

• Class > 500 lines
• Class has > 10 public methods
• Class name contains "Manager", "Helper", "Utility", "Handler"
• Methods handle unrelated responsibilities
• Difficult to describe class purpose in one sentence

Impact:

• Violates Single Responsibility Principle
• Impossible to maintain
• Changes risk breaking everything
• Cannot test in isolation
• Poor reusability

Example:

# BAD: God Object
class ApplicationManager:
    def __init__(self):
        self.users = []
        self.products = []
        self.orders = []

    def add_user(self): pass
    def delete_user(self): pass
    def validate_user(self): pass
    def send_email(self): pass
    def process_payment(self): pass
    def generate_report(self): pass
    def backup_database(self): pass
    def log_activity(self): pass
    def manage_sessions(self): pass
    # ... 50 more methods

Fix: Split into focused classes with single responsibilities

Good Example:

# GOOD: Separated responsibilities
class UserManager:
    """Manages user operations only."""
    pass

class EmailService:
    """Handles email operations only."""
    pass

class PaymentService:
    """Handles payment operations only."""
    pass

class ReportService:
    """Handles report generation only."""
    pass

Related Patterns: Single Responsibility Principle, Service Layer Related Reference: See references/oop_principles.md

---

3. Complex Nested Conditionals (Arrow Anti-Pattern)

Problem: Deeply nested if/else blocks create "arrow" shape that's hard to follow.

Detection:

• Nesting depth > 3 levels
• Visual "arrow" pointing to right edge of screen

Impact:

• High cognitive load to understand control flow
• Difficult to test all branches
• Easy to introduce bugs when modifying

Example:

# BAD: Arrow anti-pattern
def process_request(user, data, permissions):
    if user:
        if user.is_active:
            if data:
                if data.is_valid():
                    if permissions:
                        if 'write' in permissions:
                            # Finally do something
                            return save_data(data)

Fix: Use guard clauses (early returns) - see patterns.md

Related Patterns: Guard Clauses, Extract Method

---

4. God Functions (Too Long, Too Complex)

Problem: Single function doing too many unrelated things.

Detection:

• Function length > 50 lines
• Cyclomatic complexity > 15
• Function name contains "and" or multiple verbs
• Multiple levels of abstraction mixed

Impact:

• Impossible to understand without reading entire function
• Difficult to test individual logic pieces
• Changes risk breaking unrelated functionality
• Often violates Single Responsibility Principle

Example:

# BAD: God function
def process_user_and_save_to_database_and_send_email(user_data):
    # 20 lines of validation
    # 15 lines of data transformation
    # 10 lines of database logic
    # 12 lines of email composition
    # 8 lines of error handling
    # Total: 65 lines of mixed concerns

Fix: Extract Method - break into focused functions (see patterns.md)

Related Patterns: Extract Method, Separate Concerns

---

5. Magic Numbers and Strings

Problem: Unexplained literal values scattered through code.

Detection:

• Numeric literals (except 0, 1, -1) without explanation
• String literals used multiple times
• Hardcoded thresholds, limits, or configuration

Impact:

• Meaning unclear without context
• Changes require finding all occurrences
• Easy to use wrong value by mistake
• Difficult to configure or test with different values

Example:

# BAD: Magic numbers everywhere
def calculate_price(base_price, quantity):
    if quantity > 10:
        discount = 0.15
    elif quantity > 5:
        discount = 0.10
    else:
        discount = 0

    price = base_price * quantity * (1 - discount)

    if price > 100:
        price *= 1.08  # ???
    else:
        price += 5  # ???

    return price

Fix: Extract to named constants (see patterns.md)

Related Patterns: Extract Magic Numbers to Named Constants

---

6. Cryptic Variable Names

Problem: Single-letter or abbreviated names that don't convey meaning.

Detection:

• Single letters (except i, j, k for simple loops)
• Abbreviations without obvious meaning (tmp, d, val, obj, mgr)
• Generic names (data, info, item, thing)

Impact:

• Requires reading surrounding code to understand purpose
• Increases cognitive load
• Makes code review difficult
• Confusing months after writing

Example:

# BAD: Cryptic names
def calc(d, r, t, m):
    p = d
    for i in range(t * m):
        p = p * (1 + r/m)
    return p

Fix: Use meaningful names (see patterns.md)

Related Patterns: Meaningful Variable Names

---

7. Missing Type Hints (Python)

Problem: Function signatures without type information.

Detection:

• Function parameters without type annotations
• Return types not specified
• Generic types (dict, list) without element types

Impact:

• Unclear what types function expects/returns
• No IDE autocomplete or type checking
• Runtime type errors not caught early
• Difficult to refactor safely

Example:

# BAD: No type hints
def process_orders(orders, user):
    results = []
    for order in orders:
        if check_permission(user, order):
            result = process(order)
            results.append(result)
    return results

Fix: Add comprehensive type hints

# GOOD: Clear type hints
from typing import List

def process_orders(
    orders: List[Order],
    user: User
) -> List[ProcessResult]:
    results: List[ProcessResult] = []
    for order in orders:
        if check_permission(user, order):
            result = process(order)
            results.append(result)
    return results

---

8. Missing or Inadequate Docstrings

Problem: Functions without documentation of purpose, parameters, or behavior.

Detection:

• Public functions without docstrings
• Docstrings that just repeat function name
• Missing parameter descriptions
• Missing return value descriptions
• No exception documentation

Impact:

• Unclear how to use function correctly
• Users must read implementation to understand behavior
• Edge cases and exceptions not documented
• Difficult for new team members

Example:

# BAD: No docstring
def calculate_discount(user, order):
    if user.tier == 'gold':
        return order.total * 0.2
    elif user.tier == 'silver':
        return order.total * 0.1
    return 0

# BAD: Useless docstring
def calculate_discount(user, order):
    """Calculate discount."""  # Adds nothing!
    # ... implementation

Fix: Write comprehensive docstrings (see patterns.md)

Related Patterns: Comprehensive Function Docstrings

---

9. Unclear Error Handling

Problem: Errors silently swallowed or handled unclearly.

Detection:

• Bare except: clauses
• Empty exception handlers
• Generic exception catching without re-raising
• Errors converted to None without logging

Impact:

• Bugs hide silently
• Difficult to debug failures
• Unclear what errors can occur
• May mask serious problems

Example:

# BAD: Silent failure
def load_config():
    try:
        with open('config.json') as f:
            return json.load(f)
    except:  # What errors? Why silent?
        return {}

# BAD: Too broad
def process_data(data):
    try:
        # 50 lines of code
        return result
    except Exception:  # Catches everything!
        return None

Fix: Handle specific exceptions, log errors, provide context

# GOOD: Clear error handling
def load_config(config_path: str = 'config.json') -> Dict[str, Any]:
    """Load configuration from JSON file.

    Args:
        config_path: Path to config file

    Returns:
        Configuration dictionary

    Raises:
        ConfigNotFoundError: If config file doesn't exist
        ConfigParseError: If config file is invalid JSON
    """
    try:
        with open(config_path) as f:
            return json.load(f)
    except FileNotFoundError:
        log.error(f"Config file not found: {config_path}")
        raise ConfigNotFoundError(f"No config file at {config_path}")
    except json.JSONDecodeError as e:
        log.error(f"Invalid JSON in config: {e}")
        raise ConfigParseError(f"Config file has invalid JSON: {e}")

---

10. Mixed Abstraction Levels

Problem: High-level and low-level operations mixed in same function.

Detection:

• Function has both business logic and implementation details
• Some lines are conceptual, others are mechanical
• Reading requires switching between abstraction levels

Impact:

• Difficult to understand function purpose
• Can't see the big picture
• Implementation details obscure logic
• Hard to modify without affecting unrelated parts

Example:

# BAD: Mixed levels
def process_order(order):
    # High-level
    customer = get_customer(order.customer_id)

    # Low-level detail
    conn = psycopg2.connect(DATABASE_URL)
    cursor = conn.cursor()
    cursor.execute("UPDATE inventory SET stock = stock - %s WHERE id = %s",
                   (order.quantity, order.product_id))

    # High-level
    send_confirmation(customer.email)

    # Low-level detail
    cursor.execute("INSERT INTO logs (message) VALUES (%s)", ("Order processed",))
    conn.commit()

Fix: Consistent Abstraction Levels (see patterns.md)

Related Patterns: Consistent Abstraction Levels, Separate Concerns

---

Medium-Priority Anti-Patterns

These anti-patterns harm maintainability but are less critical than high-priority ones.

11. Duplicate Code (DRY Violations)

Problem: Same or similar code repeated multiple times.

Detection:

• Identical code blocks in multiple places
• Similar logic with minor variations
• Copy-pasted functions with small changes

Impact:

• Bug fixes must be applied in multiple places
• Inconsistencies introduced over time
• Increased code size and maintenance burden
• Changes are error-prone

Fix: Extract common logic to shared function

---

12. Primitive Obsession

Problem: Using primitive types (string, int, dict) instead of domain objects.

Detection:

• Dictionaries with fixed keys used as objects
• String constants representing enumerations
• Multiple primitive parameters that logically go together
• Validation logic scattered across codebase

Impact:

• No type safety
• Easy to pass wrong values
• Validation logic duplicated
• Domain concepts not explicitly modeled

Example:

# BAD: Primitives everywhere
def create_user(name: str, email: str, role: str, status: str):
    if '@' not in email:  # Validation scattered
        raise ValueError()
    if role not in ['admin', 'user', 'guest']:  # Validation scattered
        raise ValueError()
    return {'name': name, 'email': email, 'role': role, 'status': status}

# GOOD: Domain objects
@dataclass
class Email:
    """Value object for email addresses."""
    address: str

    def __post_init__(self):
        if '@' not in self.address:
            raise ValueError(f"Invalid email: {self.address}")

class UserRole(Enum):
    """User role enumeration."""
    ADMIN = 'admin'
    USER = 'user'
    GUEST = 'guest'

@dataclass
class User:
    """User entity."""
    name: str
    email: Email
    role: UserRole
    status: str

---

13. Long Parameter Lists

Problem: Functions with too many parameters.

Detection:

• > 5 parameters
• Many boolean flags
• Parameters that always passed together

Impact:

• Easy to pass arguments in wrong order
• Difficult to remember parameter order
• Function signature changes affect many callers
• Often indicates function does too much

Fix:

• Group related parameters into objects
• Use builder pattern for complex construction
• Split function if doing too much

Example:

# BAD: Too many parameters
def create_report(user_id, start_date, end_date, include_details,
                  include_summary, format, timezone, currency,
                  filter_by_status, filter_by_type):
    pass

# GOOD: Parameter object
@dataclass
class ReportOptions:
    """Configuration for report generation."""
    start_date: date
    end_date: date
    include_details: bool = True
    include_summary: bool = True
    format: str = 'pdf'
    timezone: str = 'UTC'
    currency: str = 'USD'
    filter_by_status: Optional[str] = None
    filter_by_type: Optional[str] = None

def create_report(user_id: int, options: ReportOptions):
    pass

---

14. Comments Explaining What Instead of Why

Problem: Comments that describe what code does instead of why it does it.

Detection:

• Comment restates code in English
• Comment documents obvious operation
• Comment becomes outdated when code changes
• Code needs comment to be understood

Impact:

• Comments add noise without value
• Comments become outdated and misleading
• Indicates code should be clearer
• Maintenance burden keeping comments synchronized

Example:

# BAD: Obvious comments
# Increment counter by 1
counter += 1

# Get user by ID
user = get_user(user_id)

# Check if user is active
if user.is_active:
    # Do something
    pass

# GOOD: Comments explain WHY
# Force cache clear to ensure fresh data after schema migration
cache.clear()

# Use exponential backoff to avoid overwhelming API during outages
time.sleep(2 ** retry_count)

Fix:

• Make code self-documenting through naming
• Only comment non-obvious reasoning
• Move "what" explanations to docstrings

---

Low-Priority Anti-Patterns

These anti-patterns are minor annoyances but should still be addressed.

15. Inconsistent Naming Conventions

Problem: Mixed naming styles within codebase.

Detection:

• camelCase mixed with snake_case
• Inconsistent capitalization
• Some booleans use is_/has_, others don't

Impact:

• Looks unprofessional
• Harder to search and navigate
• Cognitive load from switching conventions

Fix: Follow language conventions consistently

---

16. Redundant Comments

Problem: Comments that add no information beyond code itself.

Detection:

• Comment is exact translation of code
• Comment just repeats function name
• Outdated comments that don't match code

Example:

# BAD: Redundant
# Create a new user
create_user()

# Validate the email
if not is_valid_email(email):
    raise ValueError()

# GOOD: Only when adding value
# Bypass cache to ensure consistency after database migration
user = get_user(user_id, use_cache=False)

Fix: Delete redundant comments, keep only valuable ones

---

17. Unused Code

Problem: Commented-out code, unused imports, unused variables.

Detection:

• Code blocks commented out
• Imports not used in file
• Variables assigned but never read
• Functions never called

Impact:

• Clutter and noise
• Confusion about whether code should be there
• Maintenance burden
• Git history preserves old code better than comments

Fix: Delete completely (git history preserves it if needed)

---

Anti-Pattern Priority Matrix

Use this matrix to prioritize refactoring efforts:

Priority Formula: (Impact × 2 + Maintainability × 2 - Effort) = Priority Score

Focus refactoring on high-priority anti-patterns first for maximum improvement with minimal effort.

---

Detection Checklist

Use this checklist to scan code for anti-patterns:

Function-Level Checks

• [ ] Function > 30 lines? → Extract Method
• [ ] Nesting > 3 levels? → Guard Clauses
• [ ] Complexity > 10? → Simplify or Extract
• [ ] > 5 parameters? → Parameter Object
• [ ] No docstring? → Add Documentation
• [ ] No type hints? → Add Type Hints
• [ ] Cryptic names? → Rename
• [ ] Magic numbers? → Extract Constants

File-Level Checks

• [ ] No module docstring? → Add Documentation
• [ ] Mixed abstraction levels? → Separate Concerns
• [ ] Duplicate code? → Extract Common Logic
• [ ] Unused imports? → Remove
• [ ] Inconsistent naming? → Standardize

Architecture-Level Checks

• [ ] God classes (>500 lines)? → Split Responsibilities
• [ ] Mixed concerns? → Separate Layers
• [ ] Primitive obsession? → Domain Objects
• [ ] Unclear error handling? → Explicit Exceptions

Run this checklist systematically to identify refactoring opportunities across the codebase.

Cognitive Complexity: Complete Guide

Cognitive complexity measures how difficult code is to understand, not how many execution paths exist.

---

Calculation Rules

Rule 1: Base Increments (+1)

Every break in linear flow adds +1:

def example():
    if condition:        # +1
        pass
    for item in items:   # +1
        pass
    while running:       # +1
        pass
    try:                 # +0 (try does not increment)
        pass
    except Error:        # +1
        pass
    condition and do()   # +1 (logical operator as branch)

Structures that increment:

• if, elif, else
• for, while
• except, with
• and, or (when they change flow)
• Recursion (+1 per recursive call)
• break, continue with label

Structures that do NOT increment:

• try (only except increments)
• finally
• Simple lambdas
• Top-level ternary operator

---

Rule 2: Nesting Penalty (EXPONENTIAL)

Each nested structure adds +1 per nesting level:

def nested_example():
    if a:                    # +1 (nesting=0)
        if b:                # +2 (1 base + 1 nesting)
            if c:            # +3 (1 base + 2 nesting)
                if d:        # +4 (1 base + 3 nesting)
                    pass
# Total: 1+2+3+4 = 10 for just 4 ifs!

Impact of nesting:

Nesting is the PRIMARY ENEMY of readability.

---

Rule 3: Boolean Sequences

Same operator in sequence = FREE:

# Complexity +1 (counts as a single break)
if a and b and c and d:
    pass

Operator change = +1 per change:

# Complexity +3 (each and->or or or->and change = +1)
if a and b or c and d:
    #       ^       ^
    #      +1      +1 (plus the +1 base = 3)
    pass

Best practice: Extract complex conditions into named variables:

# BEFORE: Complexity +3
if user.active and user.verified or user.is_admin and not user.banned:
    ...

# AFTER: Complexity +1 (single condition)
is_regular_authorized = user.active and user.verified
is_admin_authorized = user.is_admin and not user.banned
if is_regular_authorized or is_admin_authorized:
    ...

---

Rule 4: Switch/Match Counts ONCE

if-elif chain = +1 per branch:

# Complexity = 4
def get_word(n):
    if n == 1:           # +1
        return "one"
    elif n == 2:         # +1
        return "couple"
    elif n == 3:         # +1
        return "few"
    else:                # +1
        return "lots"

match/switch = +1 TOTAL:

# Complexity = 1 (!)
def get_word(n):
    match n:             # +1 for the entire switch
        case 1: return "one"
        case 2: return "couple"
        case 3: return "few"
        case _: return "lots"

Use `match` (Python 3.10+) to drastically reduce complexity.

---

Rule 5: Extract Method RESETS Nesting

The most powerful pattern for reducing complexity:

# BEFORE: Complexity = 6
def process_items(items):
    for item in items:           # +1, nesting +1
        if item.valid:           # +2 (1 + nesting 1)
            if item.ready:       # +3 (1 + nesting 2)
                handle(item)
# Total: 1+2+3 = 6

# AFTER: Complexity = 3 (split across 2 functions)
def process_items(items):
    for item in items:           # +1
        process_single_item(item)
# Function 1 complexity: 1

def process_single_item(item):   # NESTING RESET TO 0!
    if not item.valid:           # +1 (nesting 0)
        return
    if not item.ready:           # +1 (nesting 0)
        return
    handle(item)
# Function 2 complexity: 2

# Total: 1 + 2 = 3 (50% reduction!)

---

Tools: Ruff + Complexipy

Recommended Stack

Ruff + Complexipy is the recommended stack: both written in Rust, actively maintained, modern ecosystem.

Setup

pip install ruff complexipy radon wily

Complexipy: Dedicated Cognitive Complexity Tool

Features:

• Written in Rust (very fast)
• Actively maintained (v5.1.0, December 2025)
• Configuration via pyproject.toml
• Snapshot for legacy code (gradual adoption)
• Pre-commit hook, GitHub Action, VSCode extension

Installation

pip install complexipy

CLI

# Basic analysis
complexipy src/

# Custom threshold (default: 15, same as SonarQube)
complexipy src/ --max-complexity-allowed 15

# JSON output for CI
complexipy src/ --output-json

# Show all functions (ignore threshold)
complexipy src/ --ignore-complexity

# Sort by complexity
complexipy src/ --sort desc

Configuration (pyproject.toml)

[tool.complexipy]
paths = ["src"]
max-complexity-allowed = 15    # SonarQube default
exclude = ["tests", "migrations", "vendor"]
quiet = false
output-json = false

Python API

from complexipy import file_complexity, code_complexity

# Analyze file
result = file_complexity("src/user_service.py")
print(f"File: {result.path}")
print(f"Total complexity: {result.complexity}")

for func in result.functions:
    status = "WARNING" if func.complexity > 15 else "OK"
    print(f"  {status} {func.name}: {func.complexity} (lines {func.line_start}-{func.line_end})")

# Analyze code string
code = """
def example(x):
    if x > 0:
        for i in range(x):
            if i % 2 == 0:
                print(i)
"""
result = code_complexity(code)
print(f"Complexity: {result.complexity}")

Snapshot for Legacy Code

Key feature for gradual adoption on existing codebases:

# 1. Create snapshot of current state
complexipy src/ --snapshot-create --max-complexity-allowed 15
# Creates: complexipy-snapshot.json

# 2. In CI: block only REGRESSIONS (new complex functions)
complexipy src/ --max-complexity-allowed 15
# Passes if no new functions exceed threshold
# Fails if NEW functions exceed threshold
# Existing functions in snapshot are "grandfathered"

# 3. When you fix a function, it is automatically removed from the snapshot

Pre-commit Hook

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/rohaquinlop/complexipy-pre-commit
    rev: v3.0.0
    hooks:
      - id: complexipy
        args: [--max-complexity-allowed, "15"]

GitHub Action

- uses: rohaquinlop/complexipy-action@v2
  with:
    paths: src/
    max_complexity_allowed: 15
    output_json: true

VSCode Extension

Install "Complexipy" from the marketplace for real-time analysis with visual indicators.

Ruff Configuration (for cyclomatic + linting)

# pyproject.toml
[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
select = [
    "E", "W",     # pycodestyle
    "F",          # Pyflakes
    "C90",        # McCabe cyclomatic complexity
    "B",          # flake8-bugbear
    "SIM",        # flake8-simplify
    "N",          # pep8-naming
    "UP",         # pyupgrade
    "I",          # isort
]

[tool.ruff.lint.mccabe]
max-complexity = 10  # Cyclomatic complexity

Complete Workflow

# 1. Fast linting
ruff check src/ --fix

# 2. Cognitive complexity
complexipy src/ --max-complexity-allowed 15

# 3. Maintainability Index
radon mi src/ -s

# 4. Historical trend (optional)
wily build src/ && wily report src/

---

Combining Metrics

Do not rely on a single metric.

Combined Setup

# Install all tools
pip install flake8 flake8-cognitive-complexity radon wily

# Combined analysis
flake8 src/ --max-cognitive-complexity=15 --max-complexity=10
radon cc src/ -a -s  # Cyclomatic + Average
radon mi src/ -s     # Maintainability Index

Recommended Targets

---

Progressive Thresholds for Legacy Code

Do not apply strict thresholds to legacy code immediately.

Ratcheting Strategy

# .github/workflows/quality.yml
- name: Quality Gate (Ratcheting)
  run: |
    # Save baseline if it doesn't exist
    if [ ! -f .quality-baseline.json ]; then
      python scripts/measure_all_metrics.py > .quality-baseline.json
    fi

    # Compare against baseline
    python scripts/compare_to_baseline.py .quality-baseline.json

    # Fail if WORSE, pass if equal or better

Changed Files Only Strategy

# Apply strict thresholds ONLY to files modified in the PR
CHANGED_FILES=$(git diff --name-only origin/main...HEAD -- '*.py')

for file in $CHANGED_FILES; do
    flake8 "$file" --max-cognitive-complexity=10  # Strict for new code
done

# Permissive threshold for everything else
flake8 src/ --max-cognitive-complexity=25  # Lenient for legacy

Adoption Phases

# Phase 1: Baseline (month 1-2)
max-cognitive-complexity = 30  # Permissive, block only extreme cases

# Phase 2: Reduction (month 3-6)
max-cognitive-complexity = 20  # Moderate

# Phase 3: Target (month 6+)
max-cognitive-complexity = 15  # SonarQube standard

# Phase 4: Strict (new code only)
max-cognitive-complexity = 10  # For greenfield code

---

Historical Tracking with Wily

Monitor trends over time, not just thresholds.

Setup

pip install wily

# Build cache (once)
wily build src/ -n 100  # Last 100 commits

# Report per file
wily report src/module.py

# Diff between commits
wily diff src/ -r HEAD~10..HEAD

# Trend graph
wily graph src/module.py complexity  # Opens browser

# Rank most complex files
wily rank src/ complexity

CI Integration

# .github/workflows/wily.yml
name: Complexity Trend

on:
  push:
    branches: [main]

jobs:
  wily:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
        with:
          fetch-depth: 50  # Wily needs history

      - name: Setup
        run: pip install wily

      - name: Build Wily Cache
        run: wily build src/ -n 50

      - name: Check for Regression
        run: |
          # Fail if complexity INCREASED compared to previous commit
          wily diff src/ -r HEAD~1..HEAD --exit-zero
          if wily diff src/ -r HEAD~1..HEAD | grep -q "increased"; then
            echo "Complexity increased!"
            exit 1
          fi

Dashboard Output

+--------------------------------------------------------------+
|                    COMPLEXITY TREND                            |
+--------------------------------------------------------------+
| File: src/services/user_service.py                            |
|                                                                |
| Commit    Date       CC    MI    CoC                          |
| -------------------------------------------                   |
| abc123    2024-01-01  12    75    18                           |
| def456    2024-01-15  10    78    15                           |
| ghi789    2024-02-01   8    82    12                           |
| jkl012    2024-02-15   6    85     9                           |
|                                                                |
| TREND: Improving (-50% complexity in 6 weeks)                 |
+--------------------------------------------------------------+

---

High-Impact Refactoring Patterns

Pattern 1: Dictionary Dispatch (eliminates if-elif chains)

# BEFORE: Cognitive Complexity = 8
def process_action(action, data):
    if action == "create":           # +1
        return create_item(data)
    elif action == "read":           # +1
        return read_item(data)
    elif action == "update":         # +1
        return update_item(data)
    elif action == "delete":         # +1
        return delete_item(data)
    elif action == "archive":        # +1
        return archive_item(data)
    elif action == "restore":        # +1
        return restore_item(data)
    elif action == "clone":          # +1
        return clone_item(data)
    else:                            # +1
        raise ValueError(f"Unknown action: {action}")

# AFTER: Cognitive Complexity = 1
ACTION_HANDLERS = {
    "create": create_item,
    "read": read_item,
    "update": update_item,
    "delete": delete_item,
    "archive": archive_item,
    "restore": restore_item,
    "clone": clone_item,
}

def process_action(action, data):
    handler = ACTION_HANDLERS.get(action)
    if handler is None:              # +1 (single branch)
        raise ValueError(f"Unknown action: {action}")
    return handler(data)

# Reduction: 87.5%!

Pattern 2: Guard Clauses (eliminates nesting)

# BEFORE: Cognitive Complexity = 10
def process_order(order):
    if order:                           # +1
        if order.is_valid():            # +2 (nesting)
            if order.has_items():       # +3 (nesting)
                if order.payment_ok():  # +4 (nesting)
                    return fulfill(order)
    return OrderResult.failed()

# AFTER: Cognitive Complexity = 4
def process_order(order):
    if not order:                   # +1
        return OrderResult.failed()
    if not order.is_valid():        # +1
        return OrderResult.failed()
    if not order.has_items():       # +1
        return OrderResult.failed()
    if not order.payment_ok():      # +1
        return OrderResult.failed()
    return fulfill(order)

# Reduction: 60%!

Pattern 3: Extract + Compose (breaks apart monster functions)

# BEFORE: Single function with Cognitive Complexity = 25
def process_user_registration(data):
    # 20 lines of validation
    # 15 lines of normalization
    # 10 lines of saving
    # 10 lines of notification
    # 15 lines of logging
    pass  # 70+ lines, CC=25

# AFTER: Composition of simple functions
def process_user_registration(data):
    validated = validate_registration(data)      # CC=4
    normalized = normalize_user_data(validated)  # CC=2
    user = save_user(normalized)                 # CC=3
    send_welcome_email(user)                     # CC=2
    log_registration(user)                       # CC=1
    return user
# Main function: CC=0 (no branches!)
# Total distributed: 4+2+3+2+1 = 12 (but never >4 in a single function)

---

Quick Reference

+-------------------------------------------------------------+
|              COGNITIVE COMPLEXITY CHEAT SHEET                 |
+-------------------------------------------------------------+
| INCREMENTS (+1):                                              |
|   if, elif, else, for, while, except, and, or, recursion    |
|                                                               |
| NESTING PENALTY (+1 per level):                              |
|   Each structure inside another adds a level                 |
|   4 nested ifs = 1+2+3+4 = 10 (not 4!)                      |
|                                                               |
| DOES NOT INCREMENT:                                           |
|   try (only except), finally, simple lambdas, switch/case    |
|                                                               |
| BOOLEAN SEQUENCES:                                            |
|   a and b and c = +1 (same operator)                         |
|   a and b or c  = +2 (operator change)                       |
+-------------------------------------------------------------+
| HIGH-IMPACT PATTERNS:                                         |
|   1. Guard clauses      - eliminates nesting penalty         |
|   2. Extract method     - resets nesting to 0                |
|   3. Dictionary dispatch - if-elif chain to O(1) lookup      |
|   4. match/switch       - n branches = +1 total              |
+-------------------------------------------------------------+
| RECOMMENDED THRESHOLDS:                                       |
|   Strict (new code):  <= 10                                  |
|   Standard (SonarQube): <= 15                                |
|   Legacy (initial):   <= 25                                  |
+-------------------------------------------------------------+

Python: Complexity Reduction Example

This example shows a complete refactoring of a complex Python function that processes user orders.

Before: Complex, Hard-to-Understand Code

def process_user_orders(user_id, start_date, end_date, status_filter=None):
    """Process user orders."""
    orders = []
    user = db.query(User).filter_by(id=user_id).first()
    if user:
        if user.is_active:
            order_list = db.query(Order).filter(
                Order.user_id == user_id,
                Order.created_at >= start_date,
                Order.created_at <= end_date
            ).all()
            if order_list:
                for order in order_list:
                    if status_filter:
                        if order.status == status_filter:
                            if order.items:
                                total = 0
                                for item in order.items:
                                    if item.quantity > 0:
                                        product = db.query(Product).filter_by(
                                            id=item.product_id
                                        ).first()
                                        if product:
                                            if product.is_available:
                                                item_total = item.quantity * product.price
                                                if product.discount > 0:
                                                    item_total *= (1 - product.discount)
                                                total += item_total
                                if total > 0:
                                    order.total = total
                                    orders.append({
                                        'order_id': order.id,
                                        'total': total,
                                        'items_count': len(order.items)
                                    })
                    else:
                        if order.items:
                            total = 0
                            for item in order.items:
                                if item.quantity > 0:
                                    product = db.query(Product).filter_by(
                                        id=item.product_id
                                    ).first()
                                    if product:
                                        if product.is_available:
                                            item_total = item.quantity * product.price
                                            if product.discount > 0:
                                                item_total *= (1 - product.discount)
                                            total += item_total
                            if total > 0:
                                order.total = total
                                orders.append({
                                    'order_id': order.id,
                                    'total': total,
                                    'items_count': len(order.items)
                                })
    return orders

Problems

1. Nesting depth: 8 levels deep (target: ≤3) 2. Function length: 55 lines (target: <30) 3. Cyclomatic complexity: 22 (target: <10) 4. Duplicate code: Order processing logic repeated for filtered/unfiltered cases 5. Mixed abstraction levels: Database queries, business logic, and calculations all mixed 6. Missing type hints: No type information for parameters or return value 7. Inadequate docstring: Doesn't explain parameters, returns, or behavior

After: Clear, Maintainable Code

from typing import List, Optional, Dict, Any
from datetime import datetime
from decimal import Decimal


def process_user_orders(
    user_id: int,
    start_date: datetime,
    end_date: datetime,
    status_filter: Optional[str] = None
) -> List[Dict[str, Any]]:
    """Process and calculate totals for user orders in date range.

    Args:
        user_id: ID of the user whose orders to process
        start_date: Start of date range (inclusive)
        end_date: End of date range (inclusive)
        status_filter: Optional status to filter orders by ('pending', 'completed', etc.)

    Returns:
        List of order summaries with calculated totals. Each summary contains:
        - order_id: Order identifier
        - total: Calculated order total after discounts
        - items_count: Number of items in the order

    Raises:
        UserNotFoundError: If user_id doesn't exist
        UserInactiveError: If user is not active

    Example:
        >>> orders = process_user_orders(
        ...     user_id=123,
        ...     start_date=datetime(2024, 1, 1),
        ...     end_date=datetime(2024, 1, 31),
        ...     status_filter='completed'
        ... )
    """
    user = get_active_user(user_id)
    orders = fetch_orders_in_date_range(user_id, start_date, end_date)

    if status_filter:
        orders = filter_orders_by_status(orders, status_filter)

    return [
        create_order_summary(order)
        for order in orders
        if has_valid_items(order)
    ]


def get_active_user(user_id: int) -> User:
    """Fetch user and verify they are active.

    Args:
        user_id: User identifier

    Returns:
        Active user object

    Raises:
        UserNotFoundError: If user doesn't exist
        UserInactiveError: If user is not active
    """
    user = db.query(User).filter_by(id=user_id).first()

    if not user:
        raise UserNotFoundError(f"User {user_id} not found")

    if not user.is_active:
        raise UserInactiveError(f"User {user_id} is not active")

    return user


def fetch_orders_in_date_range(
    user_id: int,
    start_date: datetime,
    end_date: datetime
) -> List[Order]:
    """Fetch all orders for user in date range.

    Args:
        user_id: User identifier
        start_date: Start of date range (inclusive)
        end_date: End of date range (inclusive)

    Returns:
        List of orders in date range
    """
    return db.query(Order).filter(
        Order.user_id == user_id,
        Order.created_at >= start_date,
        Order.created_at <= end_date
    ).all()


def filter_orders_by_status(orders: List[Order], status: str) -> List[Order]:
    """Filter orders by status.

    Args:
        orders: List of orders to filter
        status: Status to filter by

    Returns:
        Filtered list of orders
    """
    return [order for order in orders if order.status == status]


def has_valid_items(order: Order) -> bool:
    """Check if order has items with valid quantities.

    Args:
        order: Order to check

    Returns:
        True if order has at least one item with quantity > 0
    """
    if not order.items:
        return False

    return any(item.quantity > 0 for item in order.items)


def create_order_summary(order: Order) -> Dict[str, Any]:
    """Create order summary with calculated total.

    Args:
        order: Order to summarize

    Returns:
        Dictionary with order_id, total, and items_count
    """
    total = calculate_order_total(order)
    order.total = total  # Update order record

    return {
        'order_id': order.id,
        'total': float(total),
        'items_count': len(order.items)
    }


def calculate_order_total(order: Order) -> Decimal:
    """Calculate total for order including discounts.

    Args:
        order: Order to calculate total for

    Returns:
        Total amount after applying discounts
    """
    return sum(
        calculate_item_total(item)
        for item in order.items
        if item.quantity > 0
    )


def calculate_item_total(item: OrderItem) -> Decimal:
    """Calculate total for a single order item with discount.

    Args:
        item: Order item to calculate

    Returns:
        Item total after discount, or 0 if product unavailable
    """
    product = fetch_product(item.product_id)

    if not product or not product.is_available:
        return Decimal('0')

    base_total = Decimal(str(item.quantity)) * product.price

    return apply_discount(base_total, product.discount)


def fetch_product(product_id: int) -> Optional[Product]:
    """Fetch product by ID from database.

    Args:
        product_id: Product identifier

    Returns:
        Product object or None if not found
    """
    return db.query(Product).filter_by(id=product_id).first()


def apply_discount(amount: Decimal, discount_rate: float) -> Decimal:
    """Apply discount percentage to amount.

    Args:
        amount: Original amount
        discount_rate: Discount as decimal (0.15 for 15%)

    Returns:
        Amount after discount applied
    """
    if discount_rate <= 0:
        return amount

    return amount * (Decimal('1') - Decimal(str(discount_rate)))

Metrics Comparison

Improvements Made

1. Guard Clauses

• Converted deep nesting to early returns
• Moved validation to dedicated functions

2. Extract Method

• Split 55-line function into 10 focused functions
• Each function has single responsibility
• Average function length now 8 lines

3. Consistent Abstraction Levels

• Main function shows high-level flow
• Details delegated to helper functions
• Database queries isolated in repository functions

4. Removed Code Duplication

• Order processing logic was duplicated for filtered/unfiltered cases
• Now uses single code path with conditional filtering

5. Added Comprehensive Documentation

• Full docstrings with Args, Returns, Raises
• Example usage in main function docstring
• Type hints for all parameters and returns

6. Improved Naming

• Function names clearly describe purpose
• Boolean function has_valid_items uses has_ prefix
• Calculation functions use calculate_ prefix

7. Better Error Handling

• Explicit exceptions (UserNotFoundError, UserInactiveError)
• Clear error messages with context
• Separated validation from business logic

Testing Benefits

The refactored code is much easier to test:

# Can now test each piece independently

def test_calculate_item_total_with_discount():
    """Test item total calculation with discount applied."""
    item = OrderItem(quantity=2, product_id=1)
    product = Product(price=Decimal('100'), discount=0.10, is_available=True)

    with mock.patch('fetch_product', return_value=product):
        total = calculate_item_total(item)

    assert total == Decimal('180')  # 2 * 100 * 0.90


def test_has_valid_items_empty_order():
    """Test that order with no items is invalid."""
    order = Order(items=[])
    assert not has_valid_items(order)


def test_filter_orders_by_status():
    """Test filtering orders by status."""
    orders = [
        Order(id=1, status='pending'),
        Order(id=2, status='completed'),
        Order(id=3, status='pending'),
    ]

    filtered = filter_orders_by_status(orders, 'pending')

    assert len(filtered) == 2
    assert all(o.status == 'pending' for o in filtered)

Conclusion

This refactoring demonstrates:

• Complexity reduction through guard clauses and extraction
• Improved readability with clear function names and structure
• Better maintainability with single-responsibility functions
• Enhanced testability with isolated, focused units
• Professional documentation with comprehensive docstrings and type hints

The code is now much easier for developers to understand, modify, and maintain while preserving identical behavior.

Script-Like to OOP Transformation: Complete Example

This document demonstrates a complete transformation from script-like, procedural "spaghetti code" to clean, well-structured OOP architecture.

Overview

Before:

• Single file with scattered functions
• Global state
• No clear structure or boundaries
• Difficult to test, maintain, and extend

After:

• Organized modules with clear responsibilities
• Proper class hierarchy with encapsulation
• Dependency injection
• Testable, maintainable, extensible

---

BEFORE: Script-Like Code (Spaghetti Code)

File: user_processor.py (Single file, ~200 lines)

"""
Script to process user data from API and generate reports.
"""
import requests
import json
from datetime import datetime

# Global state - DANGER!
users_cache = {}
failed_users = []
CONFIG = None
db_connection = None
API_BASE_URL = "https://api.example.com"
RETRY_COUNT = 3
TIMEOUT = 30

def init():
    """Initialize global state."""
    global CONFIG, db_connection
    with open('config.json') as f:
        CONFIG = json.load(f)
    db_connection = create_db_connection()

def create_db_connection():
    """Create database connection."""
    # Direct database access
    import psycopg2
    return psycopg2.connect(
        host=CONFIG['db_host'],
        database=CONFIG['db_name'],
        user=CONFIG['db_user'],
        password=CONFIG['db_pass']
    )

def fetch_user(user_id):
    """Fetch user from API."""
    global users_cache

    # Check cache
    if user_id in users_cache:
        return users_cache[user_id]

    # Make API call
    url = f"{API_BASE_URL}/users/{user_id}"
    for attempt in range(RETRY_COUNT):
        try:
            response = requests.get(url, timeout=TIMEOUT)
            if response.status_code == 200:
                data = response.json()
                users_cache[user_id] = data
                return data
            elif response.status_code == 404:
                return None
        except:
            if attempt == RETRY_COUNT - 1:
                return None
            continue
    return None

def validate_user(user_data):
    """Validate user data."""
    if not user_data:
        return False
    if not user_data.get('email'):
        return False
    if '@' not in user_data['email']:
        return False
    if not user_data.get('status'):
        return False
    if user_data['status'] not in ['active', 'inactive', 'pending']:
        return False
    return True

def process_user(user_id):
    """Process a single user."""
    global failed_users, db_connection

    # Fetch user
    user = fetch_user(user_id)
    if not user:
        failed_users.append(user_id)
        return None

    # Validate
    if not validate_user(user):
        failed_users.append(user_id)
        return None

    # Transform data
    processed = {
        'user_id': user['id'],
        'email': user['email'],
        'name': user.get('first_name', '') + ' ' + user.get('last_name', ''),
        'status': user['status'],
        'created_at': user.get('created_at'),
        'last_login': user.get('last_login'),
        'is_premium': user.get('subscription_tier') == 'premium'
    }

    # Save to database
    cursor = db_connection.cursor()
    try:
        cursor.execute("""
            INSERT INTO processed_users
            (user_id, email, name, status, created_at, last_login, is_premium)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (user_id) DO UPDATE SET
                email = EXCLUDED.email,
                name = EXCLUDED.name,
                status = EXCLUDED.status,
                last_login = EXCLUDED.last_login,
                is_premium = EXCLUDED.is_premium
        """, (
            processed['user_id'],
            processed['email'],
            processed['name'],
            processed['status'],
            processed['created_at'],
            processed['last_login'],
            processed['is_premium']
        ))
        db_connection.commit()
    except Exception as e:
        db_connection.rollback()
        failed_users.append(user_id)
        return None

    return processed

def process_users_batch(user_ids):
    """Process multiple users."""
    results = []
    for user_id in user_ids:
        result = process_user(user_id)
        if result:
            results.append(result)
    return results

def generate_report():
    """Generate processing report."""
    global users_cache, failed_users

    total_cached = len(users_cache)
    total_failed = len(failed_users)
    total_processed = total_cached - total_failed

    report = {
        'timestamp': datetime.now().isoformat(),
        'total_processed': total_processed,
        'total_failed': total_failed,
        'success_rate': (total_processed / total_cached * 100) if total_cached > 0 else 0,
        'failed_user_ids': failed_users
    }

    # Save report
    with open(f'report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json', 'w') as f:
        json.dump(report, f, indent=2)

    return report

def cleanup():
    """Cleanup resources."""
    global db_connection, users_cache, failed_users
    if db_connection:
        db_connection.close()
    users_cache.clear()
    failed_users.clear()

# Script execution
if __name__ == "__main__":
    try:
        init()

        # Get user IDs from file
        with open('user_ids.txt') as f:
            user_ids = [int(line.strip()) for line in f if line.strip()]

        # Process users
        results = process_users_batch(user_ids)

        # Generate report
        report = generate_report()
        print(f"Processed {report['total_processed']} users")
        print(f"Failed: {report['total_failed']}")
        print(f"Success rate: {report['success_rate']:.1f}%")

    finally:
        cleanup()

Problems with the Script-Like Approach:

1. Global State Everywhere - users_cache, failed_users, CONFIG, db_connection 2. No Structure - Everything in one file 3. No Separation of Concerns - HTTP, validation, database, reporting all mixed 4. Impossible to Test - Functions depend on global state 5. No Dependency Injection - Hard-coded dependencies 6. Poor Error Handling - Generic exceptions, silent failures 7. No Type Safety - No type hints 8. Poor Reusability - Can't reuse components 9. Difficult to Extend - Adding features affects everything 10. No Clear Boundaries - Function responsibilities unclear

---

AFTER: OOP-Based Architecture

Project Structure

user_processor/
├── __init__.py
├── models/
│   ├── __init__.py
│   └── user.py
├── repositories/
│   ├── __init__.py
│   ├── user_repository.py
│   └── database_repository.py
├── services/
│   ├── __init__.py
│   ├── user_service.py
│   └── report_service.py
├── clients/
│   ├── __init__.py
│   └── api_client.py
├── config/
│   ├── __init__.py
│   └── settings.py
└── main.py

1. Models (Domain Objects)

File: `models/user.py`

"""User domain models."""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional


class UserStatus(Enum):
    """User status enumeration."""
    ACTIVE = 'active'
    INACTIVE = 'inactive'
    PENDING = 'pending'


@dataclass
class User:
    """User domain model."""
    id: int
    email: str
    first_name: str
    last_name: str
    status: UserStatus
    created_at: datetime
    last_login: Optional[datetime]
    subscription_tier: str

    @property
    def full_name(self) -> str:
        """Get full name."""
        return f"{self.first_name} {self.last_name}".strip()

    @property
    def is_premium(self) -> bool:
        """Check if user has premium subscription."""
        return self.subscription_tier == 'premium'

    def is_valid(self) -> bool:
        """Validate user data."""
        if not self.email or '@' not in self.email:
            return False
        if not self.status:
            return False
        return True


@dataclass
class ProcessedUser:
    """Processed user result."""
    user_id: int
    email: str
    name: str
    status: str
    created_at: datetime
    last_login: Optional[datetime]
    is_premium: bool

    @classmethod
    def from_user(cls, user: User) -> 'ProcessedUser':
        """Create from User domain model."""
        return cls(
            user_id=user.id,
            email=user.email,
            name=user.full_name,
            status=user.status.value,
            created_at=user.created_at,
            last_login=user.last_login,
            is_premium=user.is_premium
        )

2. Configuration

File: `config/settings.py`

"""Application settings."""
from dataclasses import dataclass
from typing import Optional
import json
from pathlib import Path


@dataclass
class DatabaseConfig:
    """Database configuration."""
    host: str
    database: str
    user: str
    password: str
    port: int = 5432


@dataclass
class APIConfig:
    """API configuration."""
    base_url: str
    timeout: int = 30
    retry_count: int = 3


@dataclass
class Settings:
    """Application settings."""
    database: DatabaseConfig
    api: APIConfig

    @classmethod
    def load_from_file(cls, config_path: str = 'config.json') -> 'Settings':
        """Load settings from JSON file.

        Args:
            config_path: Path to configuration file

        Returns:
            Settings instance

        Raises:
            FileNotFoundError: If config file doesn't exist
            ValueError: If config file is invalid
        """
        path = Path(config_path)
        if not path.exists():
            raise FileNotFoundError(f"Config file not found: {config_path}")

        try:
            with open(path) as f:
                data = json.load(f)

            return cls(
                database=DatabaseConfig(
                    host=data['db_host'],
                    database=data['db_name'],
                    user=data['db_user'],
                    password=data['db_pass'],
                    port=data.get('db_port', 5432)
                ),
                api=APIConfig(
                    base_url=data.get('api_base_url', 'https://api.example.com'),
                    timeout=data.get('api_timeout', 30),
                    retry_count=data.get('api_retry_count', 3)
                )
            )
        except (KeyError, json.JSONDecodeError) as e:
            raise ValueError(f"Invalid config file: {e}")

3. API Client (External Service Access)

File: `clients/api_client.py`

"""HTTP API client."""
import requests
from typing import Optional, Dict, Any
import logging
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


logger = logging.getLogger(__name__)


class APIClient:
    """HTTP API client with retry logic."""

    def __init__(self, base_url: str, timeout: int = 30, retry_count: int = 3):
        """Initialize API client.

        Args:
            base_url: Base URL for API
            timeout: Request timeout in seconds
            retry_count: Number of retry attempts
        """
        self._base_url = base_url.rstrip('/')
        self._timeout = timeout
        self._session = self._create_session(retry_count)

    def _create_session(self, retry_count: int) -> requests.Session:
        """Create requests session with retry logic."""
        session = requests.Session()
        retry_strategy = Retry(
            total=retry_count,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        return session

    def get(self, endpoint: str) -> Optional[Dict[str, Any]]:
        """Make GET request.

        Args:
            endpoint: API endpoint (e.g., '/users/123')

        Returns:
            Response data as dictionary, or None if not found

        Raises:
            APIError: If request fails
        """
        url = f"{self._base_url}{endpoint}"

        try:
            response = self._session.get(url, timeout=self._timeout)

            if response.status_code == 200:
                return response.json()
            elif response.status_code == 404:
                logger.info(f"Resource not found: {url}")
                return None
            else:
                logger.error(f"API request failed: {response.status_code} - {url}")
                raise APIError(f"API returned status {response.status_code}")

        except requests.exceptions.RequestException as e:
            logger.error(f"API request failed: {e}")
            raise APIError(f"Request failed: {e}")

    def close(self):
        """Close session and cleanup resources."""
        self._session.close()


class APIError(Exception):
    """API-related error."""
    pass

4. Repositories (Data Access)

File: `repositories/user_repository.py`

"""User data repository."""
from typing import Optional, Dict
import logging
from ..models.user import User, UserStatus
from ..clients.api_client import APIClient, APIError
from datetime import datetime


logger = logging.getLogger(__name__)


class UserRepository:
    """Repository for accessing user data from API."""

    def __init__(self, api_client: APIClient):
        """Initialize user repository.

        Args:
            api_client: API client for making HTTP requests
        """
        self._api_client = api_client
        self._cache: Dict[int, User] = {}

    def get_by_id(self, user_id: int) -> Optional[User]:
        """Get user by ID.

        Args:
            user_id: User identifier

        Returns:
            User instance if found, None otherwise

        Raises:
            UserRepositoryError: If data access fails
        """
        # Check cache first
        if user_id in self._cache:
            logger.debug(f"User {user_id} found in cache")
            return self._cache[user_id]

        # Fetch from API
        try:
            data = self._api_client.get(f"/users/{user_id}")
            if not data:
                logger.info(f"User {user_id} not found")
                return None

            # Convert to domain model
            user = self._data_to_user(data)

            # Cache result
            self._cache[user_id] = user

            return user

        except APIError as e:
            logger.error(f"Failed to fetch user {user_id}: {e}")
            raise UserRepositoryError(f"Failed to fetch user: {e}")

    def _data_to_user(self, data: dict) -> User:
        """Convert API data to User domain model."""
        return User(
            id=data['id'],
            email=data['email'],
            first_name=data.get('first_name', ''),
            last_name=data.get('last_name', ''),
            status=UserStatus(data['status']),
            created_at=datetime.fromisoformat(data['created_at']),
            last_login=datetime.fromisoformat(data['last_login']) if data.get('last_login') else None,
            subscription_tier=data.get('subscription_tier', 'free')
        )

    def clear_cache(self):
        """Clear user cache."""
        self._cache.clear()


class UserRepositoryError(Exception):
    """User repository error."""
    pass

File: `repositories/database_repository.py`

"""Database repository for storing processed users."""
import psycopg2
from typing import Optional, List
import logging
from ..models.user import ProcessedUser
from ..config.settings import DatabaseConfig


logger = logging.getLogger(__name__)


class DatabaseRepository:
    """Repository for database operations."""

    def __init__(self, config: DatabaseConfig):
        """Initialize database repository.

        Args:
            config: Database configuration
        """
        self._config = config
        self._connection = None

    def connect(self):
        """Establish database connection."""
        if self._connection is None or self._connection.closed:
            try:
                self._connection = psycopg2.connect(
                    host=self._config.host,
                    database=self._config.database,
                    user=self._config.user,
                    password=self._config.password,
                    port=self._config.port
                )
                logger.info("Database connection established")
            except psycopg2.Error as e:
                logger.error(f"Database connection failed: {e}")
                raise DatabaseError(f"Connection failed: {e}")

    def save_user(self, user: ProcessedUser) -> bool:
        """Save processed user to database.

        Args:
            user: Processed user data

        Returns:
            True if successful, False otherwise
        """
        if not self._connection or self._connection.closed:
            raise DatabaseError("No database connection")

        cursor = self._connection.cursor()
        try:
            cursor.execute("""
                INSERT INTO processed_users
                (user_id, email, name, status, created_at, last_login, is_premium)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (user_id) DO UPDATE SET
                    email = EXCLUDED.email,
                    name = EXCLUDED.name,
                    status = EXCLUDED.status,
                    last_login = EXCLUDED.last_login,
                    is_premium = EXCLUDED.is_premium
            """, (
                user.user_id,
                user.email,
                user.name,
                user.status,
                user.created_at,
                user.last_login,
                user.is_premium
            ))
            self._connection.commit()
            logger.debug(f"Saved user {user.user_id} to database")
            return True

        except psycopg2.Error as e:
            self._connection.rollback()
            logger.error(f"Failed to save user {user.user_id}: {e}")
            return False

    def close(self):
        """Close database connection."""
        if self._connection and not self._connection.closed:
            self._connection.close()
            logger.info("Database connection closed")


class DatabaseError(Exception):
    """Database operation error."""
    pass

5. Services (Business Logic)

File: `services/user_service.py`

"""User processing service."""
from typing import List, Optional
import logging
from ..models.user import User, ProcessedUser
from ..repositories.user_repository import UserRepository, UserRepositoryError
from ..repositories.database_repository import DatabaseRepository


logger = logging.getLogger(__name__)


class UserService:
    """Service for processing users."""

    def __init__(
        self,
        user_repository: UserRepository,
        database_repository: DatabaseRepository
    ):
        """Initialize user service.

        Args:
            user_repository: Repository for fetching user data
            database_repository: Repository for storing processed users
        """
        self._user_repo = user_repository
        self._db_repo = database_repository
        self._failed_user_ids: List[int] = []

    def process_user(self, user_id: int) -> Optional[ProcessedUser]:
        """Process a single user.

        Args:
            user_id: User identifier

        Returns:
            ProcessedUser if successful, None otherwise
        """
        try:
            # Fetch user
            user = self._user_repo.get_by_id(user_id)
            if not user:
                logger.warning(f"User {user_id} not found")
                self._failed_user_ids.append(user_id)
                return None

            # Validate
            if not user.is_valid():
                logger.warning(f"User {user_id} validation failed")
                self._failed_user_ids.append(user_id)
                return None

            # Transform
            processed = ProcessedUser.from_user(user)

            # Save
            if not self._db_repo.save_user(processed):
                logger.error(f"Failed to save user {user_id}")
                self._failed_user_ids.append(user_id)
                return None

            logger.info(f"Successfully processed user {user_id}")
            return processed

        except UserRepositoryError as e:
            logger.error(f"Error processing user {user_id}: {e}")
            self._failed_user_ids.append(user_id)
            return None

    def process_batch(self, user_ids: List[int]) -> List[ProcessedUser]:
        """Process multiple users.

        Args:
            user_ids: List of user identifiers

        Returns:
            List of successfully processed users
        """
        results = []

        for user_id in user_ids:
            result = self.process_user(user_id)
            if result:
                results.append(result)

        logger.info(f"Processed {len(results)} of {len(user_ids)} users")
        return results

    @property
    def failed_user_ids(self) -> List[int]:
        """Get list of failed user IDs."""
        return self._failed_user_ids.copy()

    def clear_failed(self):
        """Clear failed user IDs list."""
        self._failed_user_ids.clear()

File: `services/report_service.py`

"""Report generation service."""
from dataclasses import dataclass
from datetime import datetime
from typing import List
import json
import logging
from pathlib import Path


logger = logging.getLogger(__name__)


@dataclass
class ProcessingReport:
    """Report of user processing results."""
    timestamp: datetime
    total_processed: int
    total_failed: int
    success_rate: float
    failed_user_ids: List[int]

    def to_dict(self) -> dict:
        """Convert to dictionary."""
        return {
            'timestamp': self.timestamp.isoformat(),
            'total_processed': self.total_processed,
            'total_failed': self.total_failed,
            'success_rate': self.success_rate,
            'failed_user_ids': self.failed_user_ids
        }


class ReportService:
    """Service for generating processing reports."""

    def __init__(self, output_dir: str = '.'):
        """Initialize report service.

        Args:
            output_dir: Directory for saving reports
        """
        self._output_dir = Path(output_dir)
        self._output_dir.mkdir(exist_ok=True)

    def generate_report(
        self,
        total_processed: int,
        failed_user_ids: List[int]
    ) -> ProcessingReport:
        """Generate processing report.

        Args:
            total_processed: Number of successfully processed users
            failed_user_ids: List of failed user IDs

        Returns:
            Processing report
        """
        total_failed = len(failed_user_ids)
        total_users = total_processed + total_failed

        success_rate = (
            (total_processed / total_users * 100)
            if total_users > 0
            else 0.0
        )

        report = ProcessingReport(
            timestamp=datetime.now(),
            total_processed=total_processed,
            total_failed=total_failed,
            success_rate=success_rate,
            failed_user_ids=failed_user_ids
        )

        # Save to file
        self._save_report(report)

        return report

    def _save_report(self, report: ProcessingReport):
        """Save report to file."""
        filename = f"report_{report.timestamp.strftime('%Y%m%d_%H%M%S')}.json"
        filepath = self._output_dir / filename

        try:
            with open(filepath, 'w') as f:
                json.dump(report.to_dict(), f, indent=2)
            logger.info(f"Report saved to {filepath}")
        except IOError as e:
            logger.error(f"Failed to save report: {e}")

6. Main Application

File: `main.py`

"""Main application entry point."""
import logging
from pathlib import Path
from typing import List

from .config.settings import Settings
from .clients.api_client import APIClient
from .repositories.user_repository import UserRepository
from .repositories.database_repository import DatabaseRepository
from .services.user_service import UserService
from .services.report_service import ReportService


# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class Application:
    """Main application class."""

    def __init__(self, config_path: str = 'config.json'):
        """Initialize application.

        Args:
            config_path: Path to configuration file
        """
        # Load configuration
        self.settings = Settings.load_from_file(config_path)

        # Initialize components
        self.api_client = APIClient(
            base_url=self.settings.api.base_url,
            timeout=self.settings.api.timeout,
            retry_count=self.settings.api.retry_count
        )

        self.user_repository = UserRepository(self.api_client)

        self.db_repository = DatabaseRepository(self.settings.database)

        self.user_service = UserService(
            user_repository=self.user_repository,
            database_repository=self.db_repository
        )

        self.report_service = ReportService()

    def run(self, user_ids_file: str = 'user_ids.txt'):
        """Run user processing.

        Args:
            user_ids_file: Path to file containing user IDs
        """
        try:
            # Connect to database
            self.db_repository.connect()

            # Load user IDs
            user_ids = self._load_user_ids(user_ids_file)
            logger.info(f"Loaded {len(user_ids)} user IDs")

            # Process users
            results = self.user_service.process_batch(user_ids)
            logger.info(f"Processed {len(results)} users")

            # Generate report
            report = self.report_service.generate_report(
                total_processed=len(results),
                failed_user_ids=self.user_service.failed_user_ids
            )

            # Print summary
            print(f"\nProcessing Summary:")
            print(f"  Total processed: {report.total_processed}")
            print(f"  Total failed: {report.total_failed}")
            print(f"  Success rate: {report.success_rate:.1f}%")

            if report.failed_user_ids:
                print(f"\nFailed user IDs: {report.failed_user_ids}")

        finally:
            # Cleanup
            self.cleanup()

    def _load_user_ids(self, filepath: str) -> List[int]:
        """Load user IDs from file."""
        path = Path(filepath)
        if not path.exists():
            raise FileNotFoundError(f"User IDs file not found: {filepath}")

        with open(path) as f:
            user_ids = [
                int(line.strip())
                for line in f
                if line.strip() and line.strip().isdigit()
            ]

        return user_ids

    def cleanup(self):
        """Cleanup resources."""
        logger.info("Cleaning up resources...")
        self.api_client.close()
        self.db_repository.close()
        self.user_repository.clear_cache()
        logger.info("Cleanup complete")


def main():
    """Application entry point."""
    try:
        app = Application()
        app.run()
    except Exception as e:
        logger.exception(f"Application failed: {e}")
        return 1
    return 0


if __name__ == "__main__":
    exit(main())

---

Comparison: Benefits of OOP Approach

---

Key OOP Principles Applied

1. Single Responsibility Principle (SRP)

• Each class has one clear responsibility
• UserRepository only fetches user data
• DatabaseRepository only handles database operations
• UserService only contains business logic
• ReportService only generates reports

2. Dependency Injection

• Dependencies passed through constructors
• No hard-coded dependencies
• Easy to test with mocks
• Easy to swap implementations

3. Separation of Concerns

• Models: Domain objects and data structures
• Repositories: Data access layer
• Services: Business logic layer
• Clients: External service access
• Config: Configuration management

4. Encapsulation

• Private state (_cache, _connection)
• Public interfaces only
• Implementation details hidden

5. Domain-Driven Design

• Rich domain models (User, ProcessedUser)
• Value objects (UserStatus)
• Clear domain language

6. Layered Architecture

┌─────────────────────────┐
│   Application Layer     │  ← main.py
│   (Orchestration)       │
├─────────────────────────┤
│   Service Layer         │  ← Business Logic
│   (Business Logic)      │
├─────────────────────────┤
│   Repository Layer      │  ← Data Access
│   (Data Access)         │
├─────────────────────────┤
│   Infrastructure Layer  │  ← External Services
│   (API, Database)       │
└─────────────────────────┘

---

Testing Comparison

Script-Like Testing (Impossible)

# Can't test - depends on globals!
def test_process_user():
    global users_cache, failed_users, db_connection
    # How do we mock these?
    result = process_user(123)
    # What if it makes real API calls?

OOP Testing (Easy)

from unittest.mock import Mock
import pytest

def test_user_service_process_user_success():
    # Arrange
    mock_user_repo = Mock(spec=UserRepository)
    mock_db_repo = Mock(spec=DatabaseRepository)

    user = User(
        id=123,
        email='test@example.com',
        first_name='John',
        last_name='Doe',
        status=UserStatus.ACTIVE,
        # ...
    )
    mock_user_repo.get_by_id.return_value = user
    mock_db_repo.save_user.return_value = True

    service = UserService(mock_user_repo, mock_db_repo)

    # Act
    result = service.process_user(123)

    # Assert
    assert result is not None
    assert result.user_id == 123
    mock_user_repo.get_by_id.assert_called_once_with(123)
    mock_db_repo.save_user.assert_called_once()


def test_user_service_process_user_not_found():
    # Arrange
    mock_user_repo = Mock(spec=UserRepository)
    mock_db_repo = Mock(spec=DatabaseRepository)

    mock_user_repo.get_by_id.return_value = None

    service = UserService(mock_user_repo, mock_db_repo)

    # Act
    result = service.process_user(999)

    # Assert
    assert result is None
    assert 999 in service.failed_user_ids

---

Summary

Script-Like Code Problems:

• Global state everywhere
• No structure or organization
• Impossible to test
• Difficult to maintain and extend
• Poor error handling
• No reusability

OOP Architecture Benefits:

• Clear structure and organization
• Proper encapsulation and boundaries
• Easy to test with dependency injection
• Maintainable and extensible
• Proper error handling with typed exceptions
• Reusable components
• Type-safe with comprehensive type hints
• Follows SOLID principles
• Clean separation of concerns

The transformation demonstrates how OOP principles create code that is: 1. Readable - Clear structure and naming 2. Maintainable - Changes isolated to specific modules 3. Testable - Dependencies can be mocked 4. Extensible - Easy to add new features 5. Professional - Industry-standard architecture patterns

Flake8 Plugins Guide for Human-Readable Code

Curated 16-plugin selector list for readability-focused linting. Per-plugin rule walkthroughs live in each plugin's PyPI README; this file is the selection rationale + recommended-setup tiers + the small set of rules worth memorizing.

When to use

Setting up flake8 for a new project, or auditing an existing flake8 / ruff config to see which readability rules are missing. For ruff equivalents, see notes below.

The curated 16 plugins (the local IP)

Organized in three priority tiers. The tier you adopt depends on team appetite for lint-driven feedback.

Essential (5) -- highest ROI, install first

Recommended (7) -- catches subtler issues

Optional (4) -- strict / opinionated

Recommended setup tiers

Minimal (5)

pip install flake8 flake8-bugbear flake8-simplify \
            flake8-cognitive-complexity pep8-naming flake8-docstrings

Recommended (12)

pip install flake8 flake8-bugbear flake8-simplify \
            flake8-cognitive-complexity pep8-naming flake8-docstrings \
            flake8-comprehensions flake8-expression-complexity \
            flake8-functions flake8-variables-names tryceratops \
            flake8-builtins flake8-eradicate

Full (16) -- adds the strict/opinionated tier

# Plus:
pip install flake8-unused-arguments flake8-annotations pydoclint flake8-spellcheck

Gotchas

• Cognitive complexity > Cyclomatic complexity for readability. Cyclomatic counts branches; cognitive penalizes nesting (a triple-nested if is exponentially worse than three sequential ones). Use flake8-cognitive-complexity (or ruff's C901).
• `flake8-docstrings` checks PEP 257 compliance but not coverage -- pair with interrogate for "% functions with docstrings."
• `flake8-bugbear B006` is the highest-value single rule -- mutable default arg is one of the top 3 Python footguns.
• `tryceratops TRY003` ("avoid specifying long messages outside exception class") is opinionated -- toggle if your team prefers inline messages.
• `pep8-naming N803` flags non-snake_case argument names. Disable for compatibility shims that mirror C library APIs.
• Spell-check (`flake8-spellcheck`) needs a whitelist file (whitelist.txt) for project-specific terms -- otherwise it flags every brand name and acronym.

Rules you'll be tempted to disable but shouldn't

• B006 (mutable default args)
• B008 (function-call default args, e.g. datetime.now() in a default)
• B904 ("raise without from inside except" -- always preserve traceback chain)
• SIM102 (collapse nested ifs)
• SIM105 (contextlib.suppress over try/except/pass)
• N803 / N806 (snake_case discipline)
• E800 (no commented-out code -- VCS exists)

ruff: the 2025+ alternative

ruff re-implements most of these in Rust at 100× the speed. Equivalent rule sets:

[tool.ruff.lint]
select = [
  "E", "W", "F",         # pycodestyle + pyflakes
  "B",                   # flake8-bugbear
  "SIM",                 # flake8-simplify
  "C4",                  # flake8-comprehensions
  "C901",                # cognitive complexity (mccabe-based)
  "N",                   # pep8-naming
  "D",                   # pydocstyle
  "TRY",                 # tryceratops
  "A",                   # flake8-builtins
  "ERA",                 # flake8-eradicate
  "ARG",                 # flake8-unused-arguments
  "ANN",                 # flake8-annotations
  "I",                   # isort
  "UP",                  # pyupgrade
]

ruff covers ~95% of the curated 16 plus more. New projects in 2025-2026 should start with ruff; flake8 stays for legacy projects with custom plugin sets.

Per-plugin documentation

Cognitive complexity reference

Threshold guidance from Sonar (which flake8-cognitive-complexity adapts):

Default flake8-cognitive-complexity threshold is 15. Lower to 10 for high-discipline projects.

Related

• cognitive_complexity_guide.md -- the full Sonar-derived metric explanation with worked examples
• anti-patterns.md -- catalog with detection criteria + impact analysis
• patterns.md -- positive refactoring patterns (the alternatives these plugins suggest)
• python-refactor/SKILL.md -- workflow that runs these plugins as gates