Skip to content

ROP Conformity Metric Tool Planning

Overview

A Flake8 extension to analyze Python code and provide a deterministic metric for Railway-Oriented Programming (ROP) conformity. The tool will leverage Flake8's AST processing capabilities to identify patterns and anti-patterns, assigning penalty points for violations.

Technical Approach

1. Analysis Engine Selection

We'll implement this as a Flake8 extension because: - Integrates with existing Python tooling ecosystem - Provides robust AST handling infrastructure - Can be used with existing IDE integrations - Supports parallel file processing out of the box - Can be combined with other Flake8 plugins

2. Detailed ROP Rules and Scoring

Critical Violations (ROP1XX - 5 points each)

  1. Control Flow Violations
  2. ROP101: Try/Except blocks detected (use Result instead)
  3. ROP102: For/While loops detected (use map/filter/functional operations)
  4. ROP103: If/Else statements outside pattern matching
  5. ROP104: Direct exception raising (use Error return)

  6. ROP105: Return None or Optional usage (use Option type)

  7. State Management Violations

  8. ROP110: Mutable class attributes (missing @dataclass(frozen=True))
  9. ROP111: Global variable usage
  10. ROP112: Class or instance variable modification
  11. ROP113: List/Dict/Set mutation methods (append, update, etc.)

  12. ROP114: Assignment to function parameters

  13. Railway Pattern Violations

  14. ROP120: Missing Result type for functions that can fail
  15. ROP121: Direct access to Result.value/error (use pattern matching)
  16. ROP122: Missing error handling in pipeline
  17. ROP123: Mixing Result with exception handling

  18. ROP124: Using async/await instead of @effect.result

  19. Pydantic Model Violations

  20. ROP130: Non-immutable Pydantic model (missing ConfigDict(frozen=True))
  21. ROP131: Direct model instantiation without Result (not using create factory)
  22. ROP132: Exception raising in model validators
  23. ROP133: Mutable model defaults (lists, dicts)
  24. ROP134: Direct model attribute modification

Major Violations (ROP2XX - 3 points each)

  1. Type System Violations
  2. ROP201: Missing type hints
  3. ROP202: Using Any type
  4. ROP203: Missing Result/Option type annotations
  5. ROP204: Incorrect Result/Option generic types

  6. ROP205: Using Union instead of tagged_union

  7. Functional Pattern Violations

  8. ROP210: Using .bind() method (use pipeline or @effect.result)
  9. ROP211: Direct attribute access without pattern matching
  10. ROP212: Missing pipeline for sequential Result operations
  11. ROP213: Non-pure functions (side effects)

  12. ROP214: Missing static factory methods for tagged_union

  13. Data Model Violations

  14. ROP220: Non-frozen dataclasses
  15. ROP221: Missing immutable collections (using list instead of tuple)
  16. ROP222: Mutable default arguments
  17. ROP223: Missing validation in data models

  18. ROP224: Direct attribute modification

  19. Pydantic Pattern Violations

  20. ROP230: Missing model_validator decorators
  21. ROP231: Missing ImmutableModel base class
  22. ROP232: Incorrect validator mode (not using 'after')
  23. ROP233: Missing create factory method
  24. ROP234: Incorrect error handling in validators

Minor Violations (ROP3XX - 1 point each)

  1. Documentation Violations
  2. ROP301: Missing docstrings in functions
  3. ROP302: Non-descriptive error messages
  4. ROP303: Missing type documentation
  5. ROP304: Missing error case documentation
  6. ROP305: Missing pipeline step documentation
  7. ROP306: Multi-line docstring detected (must be single line)
  8. ROP307: Empty docstring

  9. ROP308: Docstring with redundant type information

  10. Import Violations

  11. ROP310: Missing required expression imports
  12. ROP311: Unused expression imports
  13. ROP312: Star imports
  14. ROP313: Relative imports

  15. ROP314: Missing future annotations import

  16. Style Violations

  17. ROP320: Non-descriptive Result error messages
  18. ROP321: Inconsistent pattern matching style
  19. ROP322: Complex pattern matching (too many cases)
  20. ROP323: Missing line breaks in long pipelines
  21. ROP324: Inconsistent Result/Option naming

3. Pattern Detection Rules

Required Imports Detection

REQUIRED_IMPORTS = {
    'expression': [
        'Result', 'Ok', 'Error',
        'Option', 'Some', 'Nothing',
        'effect', 'pipeline', 'pipe',
        'tagged_union', 'case', 'tag'
    ],
    'typing': ['TypeVar', 'TypeAlias', 'Self'],
    'collections.abc': ['Callable', 'Awaitable', 'Generator'],
    'pydantic': ['BaseModel', 'ConfigDict', 'model_validator', 'TypeAdapter']
}

Pattern Matching Detection

VALID_PATTERN_MATCH = '''
match result:
    case Ok(value) if isinstance(value, dict):
        # handle dict success
    case Ok(value):
        # handle other success
    case Error(msg) if "database" in str(msg):
        # handle database errors
    case Error():
        # handle other errors
'''

INVALID_PATTERN_MATCH = '''
if isinstance(result, Ok):
    value = result.value
    if isinstance(value, dict):
        # handle dict
else:
    error = result.error
'''

Type Annotation Examples

type Point = tuple[float, float]
type JsonDict = dict[str, "JsonValue"]
type JsonValue = str | int | float | bool | None | JsonDict | list[JsonValue]

def process_data(data: JsonDict) -> Result[JsonValue, str]:
    match data:
        case {"type": "point", "coords": [x, y]}:
            return Ok((float(x), float(y)))
        case _:
            return Error("Invalid data format")

Pipeline Pattern Detection

VALID_PIPELINE = '''
result = pipeline(
    validate_data,
    transform_data,
    save_data
)(data)
'''

INVALID_PIPELINE = '''
result1 = validate_data(data)
if is_ok(result1):
    result2 = transform_data(result1.value)
    if is_ok(result2):
        result3 = save_data(result2.value)
'''

Pydantic Model Pattern Detection

VALID_PYDANTIC_MODEL = '''
from typing import Self

class ImmutableModel(BaseModel):
    """Base model for all Pydantic models in ROP style"""
    model_config = ConfigDict(frozen=True, strict=True)

    @classmethod
    def create(cls: type[Self], **data) -> Result[Self, str]:
        try:
            adapter = TypeAdapter(cls)
            instance = adapter.validate_python(data)
            return Ok(instance)
        except Exception as e:
            return Error(str(e))

class User(ImmutableModel):
    name: str
    email: str

    @model_validator(mode='after')
    def validate_user(self) -> Self:
        # validation logic
        return self
'''

INVALID_PYDANTIC_MODEL = '''
class User(BaseModel):
    name: str
    email: str

    def validate(self) -> "User":
        if not self.email:
            raise ValueError("Email required")
        return self
'''

Effect Pattern Detection

VALID_EFFECT = '''
@effect.result[str, str]()
def process_user(user_id: str) -> Result[str, str]:
    user = yield from fetch_user(user_id)
    match user:
        case {"status": "active", **data}:
            return Ok(f"User {data['name']} is active")
        case _:
            return Error("Invalid user data")
'''

INVALID_EFFECT = '''
async def process_user(user_id: str):
    try:
        user = await fetch_user(user_id)
        if user["status"] == "active":
            return user["name"]
    except Exception as e:
        return None
'''

Pydantic Validation Pattern Detection

VALID_VALIDATION = '''
@model_validator(mode='after')
def validate_order(self) -> 'Order':
    if len(self.items) == 0 and self.total > 0:
        raise ValueError("Cannot have total > 0 with no items")
    return self
'''

INVALID_VALIDATION = '''
def validate(self):
    if len(self.items) == 0 and self.total > 0:
        return False
'''

Docstring Pattern Detection

VALID_DOCSTRING = '''
def process_user(name: str) -> Result[User, str]:
    """Creates a new user with validation and returns Result."""
    # function implementation
'''

INVALID_DOCSTRING_MULTILINE = '''
def process_user(name: str) -> Result[User, str]:
    """Creates a new user with validation and returns Result.

    Args:
        name: The user's name
    Returns:
        Result containing the user or error
    """
    # function implementation
'''

INVALID_DOCSTRING_TYPE_INFO = '''
def process_user(name: str) -> Result[User, str]:
    """Takes a string name and returns Result[User, str]."""
    # function implementation
'''

INVALID_DOCSTRING_EMPTY = '''
def process_user(name: str) -> Result[User, str]:
    """"""
    # function implementation
'''

4. Auto-fix Suggestions

  1. Control Flow Fixes 

    # Before
try:
    result = risky_operation()
except Exception as e:
    handle_error(e)

# After
def risky_operation() -> Result[str, str]:
    match condition:
        case True: return Ok("success")
        case False: return Error("operation failed")

  2. Loop Fixes 

    # Before
results = []
for x in items:
    results.append(process(x))

# After
results = list(map(process, items))

  3. Pattern Matching Fixes 

    # Before
if is_ok(result):
    value = result.value
else:
    error = result.error

# After
match result:
    case Ok(value):
        handle_success(value)
    case Error(error):
        handle_error(error)

  4. Pydantic Model Fixes 

    # Before (Old style)
class User(BaseModel):
    name: str
    email: str

    class Config:
        validate_assignment = True

    def __init__(self, **data):
        try:
            super().__init__(**data)
        except Exception as e:
            raise ValueError(str(e))

# After (Python 3.12 + Pydantic 2.0)
from typing import Self

class User(ImmutableModel):
    name: str
    email: str

    model_config = ConfigDict(
        frozen=True,
        strict=True,
        validate_assignment=True
    )

    @classmethod
    def create(cls: type[Self], **data) -> Result[Self, str]:
        try:
            adapter = TypeAdapter(cls)
            instance = adapter.validate_python(data)
            return Ok(instance)
        except Exception as e:
            return Error(str(e))

  5. Pydantic Validation Fixes 

    # Before
def validate_total(self):
    if self.total < 0:
        raise ValueError("Total cannot be negative")

# After
@model_validator(mode='after')
def validate_total(self) -> 'Order':
    if self.total < 0:
        raise ValueError("Total cannot be negative")
    return self

  6. Docstring Fixes 

    # Before (Multi-line with type info)
def validate_user(user: User) -> Result[User, str]:
    """Validates user data and returns Result.

    Args:
        user: User instance to validate
    Returns:
        Result[User, str]: Validated user or error
    """
    # implementation

# After (Single line, no type info)
def validate_user(user: User) -> Result[User, str]:
    """Validates user data and returns success or validation errors."""
    # implementation

# Before (Empty or meaningless)
def process_data(data: dict) -> Result[dict, str]:
    """Process data"""  # Too vague
    # implementation

# After (Descriptive single line)
def process_data(data: dict) -> Result[dict, str]:
    """Transforms raw data into normalized format with validation."""
    # implementation

5. Flake8 Extension Structure

from flake8.options.manager import OptionManager
from flake8_rop.visitor import ROPVisitor

class ROPChecker:
    name = 'flake8-rop'
    version = '0.1.0'

    # Error codes prefix with ROP
    # ROP1XX: Critical violations
    # ROP2XX: Major violations
    # ROP3XX: Minor violations

    def __init__(self, tree, filename):
        self.tree = tree
        self.filename = filename

    @classmethod
    def add_options(cls, parser: OptionManager):
        parser.add_option(
            '--rop-score-threshold',
            type=int,
            default=10,
            help='Maximum allowed ROP violation score'
        )

    def run(self):
        visitor = ROPVisitor(self.filename)
        visitor.visit(self.tree)
        for violation in visitor.violations:
            yield (
                violation['line'],
                violation['col'],
                f"ROP{violation['code']}: {violation['message']}",
                type(self)
            )

6. Configuration Options

# setup.cfg or .flake8
[flake8]
rop-score-threshold = 10
rop-ignore = ROP301,ROP302
rop-select = ROP1XX,ROP2XX
max-complexity = 10

Implementation Strategy

Phase 1: Core Extension (MVP)

  1. Basic Flake8 plugin setup
  2. Critical violation detection
  3. Error code system
  4. Basic configuration

Phase 2: Enhanced Analysis

  1. Type hint verification
  2. Pattern matching detection
  3. Custom options handling
  4. Detailed error messages

Phase 3: Advanced Features

  1. Auto-fix suggestions via flake8-fix
  2. Score calculation and reporting
  3. Integration with popular IDEs
  4. Performance optimization

Code Structure

flake8_rop/
├── __init__.py
├── checker.py        # Main Flake8 plugin class
├── core/
│   ├── visitor.py    # AST visitor implementation
│   ├── rules.py      # Violation rules
│   └── scoring.py    # Scoring logic
├── fixes/           # Auto-fix suggestions
│   ├── loops.py
│   ├── exceptions.py
│   └── patterns.py
└── utils/
    ├── ast_helpers.py
    └── type_analysis.py

Installation and Usage

# Installation
pip install flake8-rop

# Usage
flake8 path/to/code/

# With specific options
flake8 --rop-score-threshold=15 path/to/code/

# Generate detailed report
flake8 --format=rop-report path/to/code/

Integration Examples

VS Code settings.json

{
    "python.linting.flake8Enabled": true,
    "python.linting.flake8Args": [
        "--rop-score-threshold=10",
        "--rop-select=ROP1XX,ROP2XX"
    ]
}

Pre-commit Configuration

repos:
-   repo: https://github.com/pycqa/flake8
    rev: '6.1.0'
    hooks:
    -   id: flake8
        additional_dependencies: [flake8-rop]
        args: [--rop-score-threshold=10]

Testing Strategy

  1. Unit Tests: 

    def test_try_except_violation():
    code = '''
    try:
        do_something()
    except Exception:
        handle_error()
    '''
    tree = ast.parse(code)
    checker = ROPChecker(tree, 'test.py')
    errors = list(checker.run())
    assert len(errors) == 1
    assert errors[0][2].startswith('ROP101')

  2. Integration Tests:

  3. Full Flake8 pipeline testing
  4. Configuration handling
  5. Multiple file analysis

  6. Plugin interactions

  7. Performance Tests:

  8. Large codebase analysis
  9. Memory usage monitoring
  10. Processing time benchmarks

Benefits of Flake8 Integration

  1. Ecosystem Benefits:
  2. Works with existing CI/CD pipelines
  3. IDE integration out of the box
  4. Compatible with other Flake8 extensions

  5. Familiar configuration format

  6. Technical Benefits:

  7. Robust AST handling
  8. Parallel file processing
  9. Standardized error reporting

  10. Plugin system for extensions

  11. User Experience:

  12. Familiar interface
  13. Standard installation process
  14. Configurable severity levels
  15. IDE integration

Development Roadmap

  1. Week 1: Basic Extension
  2. Flake8 plugin structure
  3. Basic AST visitor

  4. Error code system

  5. Week 2-3: Core Rules

  6. Critical violation detection
  7. Configuration handling

  8. Basic reporting

  9. Week 4: Enhanced Features

  10. Advanced pattern detection
  11. Scoring system

  12. Documentation

  13. Week 5-6: Polish & Release

  14. Testing
  15. Performance optimization
  16. PyPI release
  17. Documentation

Docstring Guidelines

  1. Single Line Rule
  2. Must be exactly one line
  3. No line breaks
  4. No empty lines

  5. Maximum length of 100 characters

  6. Content Rules

  7. Focus on WHAT the function does, not HOW
  8. No type information (types go in annotations)
  9. No parameter descriptions (clear parameter names instead)

  10. No return value descriptions (clear return type instead)

  11. Style Rules

  12. Start with a capital letter
  13. End with a period
  14. Use active voice

  15. Be descriptive but concise

  16. Examples of Good Docstrings 

    def validate_email(email: str) -> Result[str, str]:
    """Ensures email format is valid and domain exists."""

def process_order(order: Order) -> Result[Order, str]:
    """Validates and processes order with inventory and payment checks."""

def transform_data(data: JsonDict) -> Result[NormalizedData, str]:
    """Normalizes raw JSON data into standard internal format."""

@effect.result[User, str]()
def create_user(data: dict) -> Result[User, str]:
    """Creates new user with validation and permission checks."""

  17. Examples of Bad Docstrings 

    def validate_email(email: str) -> Result[str, str]:
    """This function takes an email string and returns a Result[str, str]."""  # No type info

def process_order(order: Order) -> Result[Order, str]:
    """Processes the order.
    Returns success or error."""  # No multi-line

def transform_data(data: JsonDict) -> Result[NormalizedData, str]:
    """Takes data and transforms it."""  # Too vague

@effect.result[User, str]()
def create_user(data: dict) -> Result[User, str]:
    """Creates user by taking the data dict and validating each field
    then saving to database if valid."""  # Too detailed/multi-line