feat: add black8 linter

Author: leandrodamascena

flake8-durable-execution/README.md

@@ -0,0 +1,64 @@
+# flake8-durable-execution
+
+Flake8 plugin for AWS Durable Execution SDK best practices.
+
+## Installation
+
+```bash
+pip install flake8-durable-execution
+```
+
+## Usage
+
+```bash
+flake8 your_code.py
+```
+
+The plugin is automatically detected by Flake8.
+
+## Rules
+
+| Code | Description |
+|------|-------------|
+| DAR001 | Avoid random calls in durable functions - results differ on replay |
+| DAR002 | Avoid datetime.now()/today() in durable functions - time differs on replay |
+| DAR003 | Avoid uuid1()/uuid4() in durable functions - generates different values on replay |
+| DAR004 | Avoid time.time()/monotonic() in durable functions - time differs on replay |
+| DAR005 | Avoid os.environ access in durable functions - environment may differ on replay |
+| DAR006 | Wrap network calls in ctx.step() - responses may differ on replay |
+| DAR007 | Durable operations cannot be called inside @durable_step |
+| DAR008 | Avoid mutating closure variables in steps - mutations are lost on replay |
+
+## Why?
+
+Durable functions may be replayed multiple times. Non-deterministic operations like `random.random()`, `datetime.now()`, or `uuid.uuid4()` will produce different values on each replay, causing unexpected behavior.
+
+Instead, use durable steps to checkpoint these values so they remain consistent across replays.
+
+## Examples
+
+### Bad
+
+```python
+@durable_execution
+def handler(event, ctx):
+    # DAR001: random value changes on replay
+    value = random.randint(1, 100)
+    
+    # DAR002: time changes on replay
+    now = datetime.now()
+    
+    # DAR006: network response may differ
+    response = requests.get("https://api.example.com")
+```
+
+### Good
+
+```python
+@durable_execution
+def handler(event, ctx):
+    # Checkpoint non-deterministic values
+    value = ctx.step(lambda: random.randint(1, 100))
+    now = ctx.step(lambda: datetime.now().isoformat())
+    response = ctx.step(lambda: requests.get("https://api.example.com").json())
+```

flake8-durable-execution/examples/bad_patterns.py

@@ -0,0 +1,71 @@
+"""Example file with bad patterns inside @durable_execution for testing."""
+
+import os
+import random
+import time
+import uuid
+from datetime import datetime
+
+import requests
+
+from aws_durable_execution_sdk_python import (
+    DurableContext,
+    StepContext,
+    durable_execution,
+    durable_step,
+)
+
+
+# This function is NOT decorated - should NOT trigger any warnings
+def normal_function():
+    value = random.random()  # OK - not in durable context
+    now = datetime.now()  # OK
+    return value
+
+
+@durable_execution
+def handler(event: dict, context: DurableContext) -> dict:
+    # GOOD - inside ctx.step(), result is checkpointed
+    safe_random = context.step(lambda: random.random())
+    safe_time = context.step(lambda: datetime.now().isoformat())
+    safe_uuid = context.step(lambda: str(uuid.uuid4()))
+
+    # DAR001: random calls OUTSIDE step
+    value = random.random()
+    num = random.randint(1, 100)
+
+    # DAR002: datetime calls OUTSIDE step
+    now = datetime.now()
+    today = datetime.today()
+
+    # DAR003: uuid generation OUTSIDE step
+    id1 = uuid.uuid4()
+    id2 = uuid.uuid1()
+
+    # DAR004: time calls OUTSIDE step
+    ts = time.time()
+    mono = time.monotonic()
+
+    # DAR005: os.environ access OUTSIDE step
+    api_key = os.environ["API_KEY"]
+
+    # DAR006: network calls OUTSIDE step
+    response = requests.get("https://api.example.com")
+    data = requests.post("https://api.example.com", json={})
+
+    # DAR008: closure mutation
+    count = 0
+    count += 1
+    items = {}
+    items["key"] = "value"
+
+    return {"value": value}
+
+
+# DAR007: durable ops inside @durable_step
+@durable_step
+def bad_step(step_ctx: StepContext) -> dict:
+    # This is wrong - can't call durable ops inside a step
+    result = step_ctx.step(lambda: 1)
+    step_ctx.wait(duration=5)
+    return {"result": result}

flake8-durable-execution/pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "flake8-durable-execution"
+version = "0.1.0"
+description = "Flake8 plugin for AWS Durable Execution SDK best practices"
+readme = "README.md"
+requires-python = ">=3.11"
+license = "Apache-2.0"
+keywords = ["flake8", "linter", "durable-execution", "aws", "lambda"]
+authors = [{ name = "AWS" }]
+classifiers = [
+  "Development Status :: 3 - Alpha",
+  "Framework :: Flake8",
+  "Intended Audience :: Developers",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+]
+dependencies = ["flake8>=6.0.0"]
+
+[project.entry-points."flake8.extension"]
+DAR = "flake8_durable_execution:DurableExecutionChecker"
+
+[project.urls]
+Source = "https://github.com/aws/aws-durable-execution-sdk-python"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/flake8_durable_execution"]

flake8-durable-execution/src/flake8_durable_execution/__init__.py

@@ -0,0 +1,4 @@
+from flake8_durable_execution.checker import DurableExecutionChecker
+
+__version__ = "0.1.0"
+__all__ = ["DurableExecutionChecker"]

flake8-durable-execution/src/flake8_durable_execution/checker.py

@@ -0,0 +1,67 @@
+"""Flake8 checker for AWS Durable Execution SDK best practices."""
+
+import ast
+from typing import Iterator
+
+from flake8_durable_execution.rules import ALL_RULES
+from flake8_durable_execution.rules.nested_durable_op import NestedDurableOpRule
+
+DURABLE_EXECUTION_DECORATORS = {"durable_execution"}
+DURABLE_STEP_DECORATORS = {"durable_step"}
+DURABLE_OPS = {"step", "invoke", "parallel", "map"}
+
+
+class DurableExecutionChecker:
+    name = "flake8-durable-execution"
+    version = "0.1.0"
+
+    def __init__(self, tree: ast.AST) -> None:
+        self.tree = tree
+        self.rules = [rule() for rule in ALL_RULES if rule != NestedDurableOpRule]
+        self.nested_rule = NestedDurableOpRule()
+
+    def run(self) -> Iterator[tuple[int, int, str, type]]:
+        for node in ast.walk(self.tree):
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                # Check @durable_execution functions for non-determinism
+                if self._has_decorator(node, DURABLE_EXECUTION_DECORATORS):
+                    yield from self._check_durable_function(node)
+
+                # Check @durable_step functions for nested durable ops
+                if self._has_decorator(node, DURABLE_STEP_DECORATORS):
+                    yield from self.nested_rule.check_function(node)
+
+    def _has_decorator(self, node: ast.FunctionDef | ast.AsyncFunctionDef, decorators: set[str]) -> bool:
+        for decorator in node.decorator_list:
+            if isinstance(decorator, ast.Name) and decorator.id in decorators:
+                return True
+            if isinstance(decorator, ast.Attribute) and decorator.attr in decorators:
+                return True
+        return False
+
+    def _check_durable_function(self, func: ast.FunctionDef | ast.AsyncFunctionDef) -> Iterator[tuple[int, int, str, type]]:
+        # Collect all nodes that are inside ctx.step() calls - these are safe
+        safe_nodes: set[int] = set()
+        self._collect_safe_nodes(func, safe_nodes)
+
+        for node in ast.walk(func):
+            if id(node) in safe_nodes:
+                continue
+            for rule in self.rules:
+                yield from rule.check(node)
+
+    def _collect_safe_nodes(self, node: ast.AST, safe_nodes: set[int]) -> None:
+        """Collect all nodes inside ctx.step(lambda: ...) calls."""
+        for child in ast.walk(node):
+            if not isinstance(child, ast.Call):
+                continue
+
+            # Check if this is a ctx.step(...) call
+            if isinstance(child.func, ast.Attribute) and child.func.attr in DURABLE_OPS:
+                # Mark all nodes inside the step arguments as safe
+                for arg in child.args:
+                    for inner in ast.walk(arg):
+                        safe_nodes.add(id(inner))
+                for kw in child.keywords:
+                    for inner in ast.walk(kw.value):
+                        safe_nodes.add(id(inner))

flake8-durable-execution/src/flake8_durable_execution/rules/__init__.py

@@ -0,0 +1,31 @@
+from flake8_durable_execution.rules.random_call import RandomCallRule
+from flake8_durable_execution.rules.datetime_now import DatetimeNowRule
+from flake8_durable_execution.rules.uuid_generation import UuidGenerationRule
+from flake8_durable_execution.rules.time_time import TimeTimeRule
+from flake8_durable_execution.rules.os_environ import OsEnvironRule
+from flake8_durable_execution.rules.network_call import NetworkCallRule
+from flake8_durable_execution.rules.nested_durable_op import NestedDurableOpRule
+from flake8_durable_execution.rules.closure_mutation import ClosureMutationRule
+
+ALL_RULES = [
+    RandomCallRule,
+    DatetimeNowRule,
+    UuidGenerationRule,
+    TimeTimeRule,
+    OsEnvironRule,
+    NetworkCallRule,
+    NestedDurableOpRule,
+    ClosureMutationRule,
+]
+
+__all__ = [
+    "RandomCallRule",
+    "DatetimeNowRule",
+    "UuidGenerationRule",
+    "TimeTimeRule",
+    "OsEnvironRule",
+    "NetworkCallRule",
+    "NestedDurableOpRule",
+    "ClosureMutationRule",
+    "ALL_RULES",
+]

flake8-durable-execution/src/flake8_durable_execution/rules/base.py

@@ -0,0 +1,23 @@
+"""Base class for all rules."""
+
+import ast
+from abc import ABC, abstractmethod
+from typing import Iterator
+
+Error = tuple[int, int, str, type]
+
+
+class BaseRule(ABC):
+    """Base class for lint rules."""
+
+    code: str
+    message: str
+
+    @abstractmethod
+    def check(self, node: ast.AST) -> Iterator[Error]:
+        """Check a node and yield errors."""
+        pass
+
+    def _error(self, node: ast.AST) -> Error:
+        """Create an error tuple for the given node."""
+        return (node.lineno, node.col_offset, f"{self.code} {self.message}", type(self))

flake8-durable-execution/src/flake8_durable_execution/rules/closure_mutation.py

@@ -0,0 +1,30 @@
+"""DAR008: Detect closure mutations that get lost on replay."""
+
+import ast
+from typing import Iterator
+
+from flake8_durable_execution.rules.base import BaseRule, Error
+
+
+class ClosureMutationRule(BaseRule):
+    code = "DAR008"
+    message = "Avoid mutating closure variables in steps - mutations are lost on replay"
+
+    def __init__(self) -> None:
+        self._outer_vars: set[str] = set()
+        self._in_step_function = False
+
+    def check(self, node: ast.AST) -> Iterator[Error]:
+        # Look for augmented assignments (+=, -=, etc.) to outer scope vars
+        if isinstance(node, ast.AugAssign):
+            if isinstance(node.target, ast.Name):
+                # This is a heuristic - we flag any augmented assignment
+                # that looks like it could be mutating an outer variable
+                yield self._error(node)
+
+        # Look for attribute mutations on outer variables
+        if isinstance(node, ast.Assign):
+            for target in node.targets:
+                if isinstance(target, ast.Subscript):
+                    # list[0] = x or dict["key"] = x
+                    yield self._error(node)

flake8-durable-execution/src/flake8_durable_execution/rules/datetime_now.py

@@ -0,0 +1,21 @@
+"""DEX002: Detect non-deterministic datetime calls."""
+
+import ast
+from typing import Iterator
+
+from flake8_durable_execution.rules.base import BaseRule, Error
+
+DATETIME_METHODS = {"now", "utcnow", "today"}
+
+
+class DatetimeNowRule(BaseRule):
+    code = "DAR002"
+    message = "Avoid datetime.now()/today() in durable functions - time differs on replay"
+
+    def check(self, node: ast.AST) -> Iterator[Error]:
+        if not isinstance(node, ast.Call):
+            return
+
+        if isinstance(node.func, ast.Attribute):
+            if node.func.attr in DATETIME_METHODS:
+                yield self._error(node)

flake8-durable-execution/src/flake8_durable_execution/rules/nested_durable_op.py

@@ -0,0 +1,28 @@
+"""DAR007: Detect durable operations inside @durable_step functions."""
+
+import ast
+from typing import Iterator
+
+from flake8_durable_execution.rules.base import BaseRule, Error
+
+DURABLE_OPS = {"step", "wait", "invoke", "parallel", "map", "wait_for_callback", "wait_for_condition"}
+
+
+class NestedDurableOpRule(BaseRule):
+    code = "DAR007"
+    message = "Durable operations cannot be called inside @durable_step"
+
+    def __init__(self) -> None:
+        self.in_durable_step = False
+
+    def check_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> Iterator[Error]:
+        """Check a @durable_step function for nested durable ops."""
+        for child in ast.walk(node):
+            if isinstance(child, ast.Call) and isinstance(child.func, ast.Attribute):
+                if child.func.attr in DURABLE_OPS:
+                    yield (child.lineno, child.col_offset, f"{self.code} {self.message}", type(self))
+
+    def check(self, node: ast.AST) -> Iterator[Error]:
+        # This rule is handled specially by the checker
+        return
+        yield  # Make this a generator