uRadical blog - The Policy Pattern in Python: Replace Conditional Sprawl with a Pipeline

Split validation rules into individual functions, compose them into a pipeline, drive the whole thing from configuration. stdlib Python, zero dependencies.

You've written this function. Five lines of validation, then someone adds a role check, then a feature flag, then an audit log inside a nested conditional because the ticket was due Friday. Now it's 90 lines, nobody wants to touch it, and every new rule makes it worse. The function works. It just can't be tested, reused, or extended without risk.

The policy pattern is a straightforward fix: each rule becomes its own function, a runner threads data through them in sequence, and a registry lets you control which rules are active from configuration. No framework, no abstract base classes, nothing outside stdlib. It's a cleaner way to organise code that's already there, for when the conditional version stops scaling.

The Monolith

Here's a deployment gate function. It decides whether a build is allowed to ship.

					gate.py — the one nobody wants to touch
					python
				

import logging
from datetime import datetime

logger = logging.getLogger(__name__)


def check_deployment(build: dict) -> dict:
    if not build.get("tests_passed"):
        raise RuntimeError(f"Build {build['id']}: tests have not passed")

    if build.get("secrets_detected"):
        raise RuntimeError(f"Build {build['id']}: secrets detected in source")

    if build.get("requires_changelog") and not build.get("changelog_updated"):
        raise RuntimeError(f"Build {build['id']}: changelog not updated")

    now = datetime.now()
    if build.get("freeze_start") and build.get("freeze_end"):
        if build["freeze_start"] <= now <= build["freeze_end"]:
            raise RuntimeError(f"Build {build['id']}: deployment freeze active")

    if build.get("requires_approval") and not build.get("approved_by"):
        raise RuntimeError(f"Build {build['id']}: missing approval")

    logger.info(
        "Deployment approved: build=%s, deployer=%s",
        build["id"],
        build.get("deployer", "unknown"),
    )
    build["audit_log"] = build.get("audit_log", [])
    build["audit_log"].append({
        "event": "deployment_approved",
        "timestamp": now.isoformat(),
        "build": build["id"],
    })

    build["approved"] = True
    return build

Forty-five lines. Six responsibilities. Zero isolation.

Now imagine adding rule seven — a container vulnerability scan. You scroll past five checks you don't care about, find the right insertion point, hope your new variable names don't collide with now or build["audit_log"], and write a test that has to satisfy rules one through six before it can even reach your code. Repeat quarterly.

Can you test the freeze window logic without also exercising the secrets check? Can you reuse the approval gate in a different pipeline? Can you disable the changelog requirement for a hotfix without adding yet another flag? No, because every rule is coupled to every other rule, and the only interface is the entire function.

Policies as Functions

A policy is one rule, one function, one responsibility. It takes a build, does its job, and returns the build. If something is wrong, it raises. That's the entire contract.

					policies.py
					python
				

import logging
from datetime import datetime
from dataclasses import dataclass, replace

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class Build:
    id: str
    deployer: str
    tests_passed: bool = False
    secrets_detected: bool = False
    requires_changelog: bool = False
    changelog_updated: bool = False
    freeze_start: datetime | None = None
    freeze_end: datetime | None = None
    requires_approval: bool = False
    approved_by: str | None = None
    approved: bool = False
    audit_log: tuple[dict, ...] = ()


def tests_passed(build: Build) -> Build:
    """Builds must have passing tests."""
    if not build.tests_passed:
        raise RuntimeError(f"Build {build.id}: tests have not passed")
    return build


def no_secrets(build: Build) -> Build:
    """No secrets may be present in source."""
    if build.secrets_detected:
        raise RuntimeError(f"Build {build.id}: secrets detected in source")
    return build


def changelog_current(build: Build) -> Build:
    """Changelog must be updated when required."""
    if build.requires_changelog and not build.changelog_updated:
        raise RuntimeError(f"Build {build.id}: changelog not updated")
    return build


def not_in_freeze(build: Build) -> Build:
    """Deployment must not occur during a freeze window."""
    if build.freeze_start and build.freeze_end:
        now = datetime.now()
        if build.freeze_start <= now <= build.freeze_end:
            raise RuntimeError(f"Build {build.id}: deployment freeze active")
    return build


def approval_granted(build: Build) -> Build:
    """Builds requiring approval must have an approver on record."""
    if build.requires_approval and not build.approved_by:
        raise RuntimeError(f"Build {build.id}: missing approval")
    return build


def audit_trail(build: Build) -> Build:
    """Append an audit entry and mark the build as approved."""
    entry = {
        "event": "deployment_approved",
        "timestamp": datetime.now().isoformat(),
        "build": build.id,
    }
    return replace(
        build,
        approved=True,
        audit_log=build.audit_log + (entry,),
    )

Two things to notice.

Build is a frozen dataclass. Policies don't mutate — they return new instances via dataclasses.replace. This isn't functional programming for the aesthetic; it means you can run the same build through two different policy pipelines and compare results without worrying about shared state. That matters the moment you need a dry-run mode alongside the real one.

Each function's docstring is the rule in plain English. When your auditor asks "what checks happen before a deployment?", the answer is the list of functions. Not a paragraph of prose someone wrote in Confluence that drifted from the code eight months ago.

Composing the Pipeline

Threading a build through a list of functions is a five-line problem:

					pipeline.py
					python
				

from functools import reduce
from collections.abc import Callable

from policies import Build

PolicyFn = Callable[[Build], Build]


def apply_policies(build: Build, policies: list[PolicyFn]) -> Build:
    return reduce(lambda b, policy: policy(b), policies, build)

reduce threads the build through each policy in order. First policy gets the original build, second gets the output of the first, and so on. This is the entire orchestration layer. There is no framework.

If this reminds you of the strategy pattern — it should. Structurally they're similar: behaviour encoded in a callable, selected at runtime. Wikipedia treats them as synonyms. But the intent is different. Strategy is about choosing one algorithm from several (pick a sorting method, pick a pricing model). Policy is about composing multiple rules that all apply. You don't pick one policy instead of another; you run all the active ones in sequence. That distinction shapes how you think about the code: strategy asks "which one?", policy asks "which ones, and in what order?"

Fail-Fast vs Fail-and-Collect

The pipeline above is fail-fast: the first broken policy raises, and you never see the rest. That's fine for deployment gates where you want to stop immediately. But plenty of real systems need fail-and-collect — run every check, gather all violations, report them together. A form validation endpoint that only tells you about the first bad field is a miserable user experience.

Both modes use the same policies. Only the runner changes:

					pipeline.py (extended)
					python
				

from dataclasses import dataclass, field


@dataclass
class PolicyResult:
    build: Build
    violations: list[str] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        return len(self.violations) == 0


def apply_policies_collect(
    build: Build, policies: list[PolicyFn]
) -> PolicyResult:
    result = PolicyResult(build=build)
    for policy in policies:
        try:
            result.build = policy(result.build)
        except RuntimeError as exc:
            result.violations.append(str(exc))
    return result

Same policy functions. Same list. Different runner. The policies don't know or care which mode they're being called in, because they shouldn't.

Testing Individual Rules

Tutorials love to say "and now each piece is independently testable" and then move on. Here's what that actually looks like:

					test_policies.py
					python
				

import pytest
from policies import Build, tests_passed, no_secrets, changelog_current


def _build(**overrides) -> Build:
    """Factory with sane defaults for testing."""
    defaults = {"id": "test-1", "deployer": "ci"}
    return Build(**{**defaults, **overrides})


class TestTestsPassed:
    def test_passes_when_tests_pass(self):
        build = _build(tests_passed=True)
        result = tests_passed(build)
        assert result is build  # unchanged, just passed through

    def test_raises_when_tests_fail(self):
        build = _build(tests_passed=False)
        with pytest.raises(RuntimeError, match="tests have not passed"):
            tests_passed(build)


class TestNoSecrets:
    def test_passes_when_clean(self):
        build = _build(secrets_detected=False)
        assert no_secrets(build) is build

    def test_raises_when_secrets_found(self):
        build = _build(secrets_detected=True)
        with pytest.raises(RuntimeError, match="secrets detected"):
            no_secrets(build)


class TestChangelogCurrent:
    def test_skipped_when_not_required(self):
        build = _build(requires_changelog=False, changelog_updated=False)
        assert changelog_current(build) is build

    def test_passes_when_updated(self):
        build = _build(requires_changelog=True, changelog_updated=True)
        assert changelog_current(build) is build

    def test_raises_when_required_but_missing(self):
        build = _build(requires_changelog=True, changelog_updated=False)
        with pytest.raises(RuntimeError, match="changelog not updated"):
            changelog_current(build)

Three policies, seven tests, every edge case covered. Each test creates exactly the state it needs and exercises exactly one rule. Try writing tests this clean against the monolith — you can't, because you can't reach the changelog check without first satisfying tests and secrets, so every test for rule N is implicitly testing rules 1 through N-1. That implicit coupling is where bugs hide.

Auto-Registration

A manual registry dict is fine for six policies. It gets tedious at twenty. A decorator solves this permanently:

					registry.py
					python
				

from pipeline import PolicyFn

POLICY_REGISTRY: dict[str, PolicyFn] = {}


def policy(name: str):
    """Register a function as a named policy."""
    def decorator(fn: PolicyFn) -> PolicyFn:
        if name in POLICY_REGISTRY:
            raise ValueError(f"Duplicate policy name: {name}")
        POLICY_REGISTRY[name] = fn
        return fn
    return decorator

Now each policy registers itself at definition:

					policies.py
					python
				

from registry import policy

@policy("tests_passed")
def tests_passed(build: Build) -> Build:
    ...

@policy("no_secrets")
def no_secrets(build: Build) -> Build:
    ...

No separate file mapping names to functions. No risk of the registry drifting from the actual implementations. The decorator is nine lines and you'll never think about registration again.

Configuration

The active policy set comes from the environment. No Pydantic. No YAML. No config framework.

					settings.py
					python
				

import os

DEFAULT_POLICIES = [
    "tests_passed",
    "no_secrets",
    "changelog_current",
    "not_in_freeze",
    "approval_granted",
    "audit_trail",
]


def get_enabled_policies() -> list[str]:
    raw = os.environ.get("DEPLOY_POLICIES", "")
    if not raw:
        return DEFAULT_POLICIES
    return [p.strip() for p in raw.split(",") if p.strip()]

And the main module:

					main.py
					python
				

from policies import Build
from pipeline import apply_policies
from registry import POLICY_REGISTRY
from settings import get_enabled_policies


def build_pipeline():
    enabled = get_enabled_policies()
    policies = []
    for name in enabled:
        fn = POLICY_REGISTRY.get(name)
        if fn is None:
            raise ValueError(f"Unknown policy: {name}")
        policies.append(fn)
    return policies


def main():
    build = Build(
        id="build-4821",
        deployer="alan",
        tests_passed=True,
        secrets_detected=False,
        requires_changelog=True,
        changelog_updated=True,
    )

    pipeline = build_pipeline()
    result = apply_policies(build, pipeline)
    print(result)


if __name__ == "__main__":
    main()

Control behaviour without touching code:

					terminal
					bash
				

# All checks
python main.py

# Hotfix: skip changelog and freeze
DEPLOY_POLICIES="tests_passed,no_secrets,approval_granted,audit_trail" python main.py

# Emergency: tests and audit only
DEPLOY_POLICIES="tests_passed,audit_trail" python main.py

No feature flag library. No conditional explosion. A list and an environment variable.

Adding a New Policy

Write the function. The decorator registers it. Add the name to config. Done.

					policies.py
					python
				

@policy("container_scan_clean")
def container_scan_clean(build: Build) -> Build:
    """Container image must pass vulnerability scanning."""
    if getattr(build, "vulnerabilities_found", False):
        raise RuntimeError(f"Build {build.id}: container vulnerabilities detected")
    return build

					terminal
					bash
				

DEPLOY_POLICIES="tests_passed,no_secrets,container_scan_clean,audit_trail"

No existing function was modified. No conditional was added. The new rule slots into the pipeline at whatever position you specify.

Trade-Offs

More code. The monolith was 45 lines; the policy version across all modules is roughly double. You pay upfront for structure you may not need yet. Use the monolith for scripts, prototypes, and things you'll delete next quarter.

Use policies when rules will grow (they always do in production), when different environments need different rule sets, when you need to test rules in isolation, or when multiple pipelines share the same checks. The pattern also makes order explicit — in the monolith, reordering checks means cutting and pasting code blocks. In the pipeline, you move a name in a list.

The real payoff is that each rule becomes individually testable, reviewable, and replaceable. And because the rules are decoupled from how they're composed, you can run the same policies fail-fast for deployment gates and fail-and-collect for form validation without changing a single policy function.

The Stack

No dependencies were added. The entire mechanism is functools.reduce, dataclasses, os.environ, and a nine-line decorator. This is stdlib Python doing exactly what it's good at.

The flags don't leak into the logic. The policies stay focused. Configuration controls composition. Adding a rule is: write a function, decorate it, add a name.