Branded Types & Result Monads: Functional Safety in Python

Two bugs the type checker won’t catch

Let’s start writing down the code of the two examples from the introduction and build up from there.

On one hand, we have a function that accepts two arguments, both typed int. The checker won’t see an issue there, the failure is silent and the carruption flows free through other services.

def add_product(cart_id: int, product_id: int) -> None:
    print(f"Adding product {product_id} to cart {cart_id}")


add_product(12345, 999)  # swapped: passes type check

On the other hand, we have a function that raises exceptions. If the caller forgets the try/except, the exception propagates to the top of the stack, potentially triggering a panic. The static checker cannot see the raise, because exceptions are not part of the annotated contract.

class ProductNotFoundError(Exception): ...


def get_product(product_id: int) -> dict[str, int]:
    if product_id != 999:
        raise ProductNotFoundError("Product not found")
    return {"id": product_id}

Branded types: one primitive, different types

NewType creates a distinct static type from a primitive. In runtime, the value’s type is the primitive, but a static checker is able to see it as a unique type:

from typing import NewType

CartId = NewType("CartId", int)
ProductId = NewType("ProductId", int)


def add_product(cart_id: CartId, product_id: ProductId) -> None:
    print(f"Adding product {product_id} to cart {cart_id}")


cid = CartId(999)
pid = ProductId(12345)
add_product(cid, pid)

A swapped call, add_product(pid, cid), is now a static error: the checker rejects ProductId where CartId is expected.

As we saw in a previous post, the runtime counterpart for boundary parsing is Annotated with a validator. For example, a CartId that must be a positive integer arriving in an HTTP request body:

from typing import Annotated
from pydantic import AfterValidator


def _positive_int(value: int) -> int:
    if value <= 0:
        raise ValueError(f"Invalid cart id: {value}")
    return value


CartId = Annotated[int, AfterValidator(_positive_int)]

Pydantic runs the validator once at construction, and downstream code receives a CartId it does not re-validate. Annotated combines both the static and runtime checks in one line.

We have seen now two patterns to create a branded type:

1. CartId = NewType("CartId", int) for the internal type.
2. CartId = Annotated[int, AfterValidator(...)] for the boundary schema.

But, if you check the previous code block, you’ll see the get_product function returns the product object, even though it can raise an exception. How can we model this fallible function? What if it instead returned a Result[CartId, ValidationError]?.

Errors as values

I’m familiar with the Result abstraction because of my experience with Rust, but it’s not a feature unique to this language: it just picked it from languages that have been using it for decades (see Scala, Hashell, or OCaml).

A function that can fail returns Result<T, E> (the value or the error, never both, never neither) and the compiler refuses to let you touch the value without acknowledging both cases. In Rust, this looks like this:

let product: Result<Product, ProductError> = db_get_product(product_id);

match product {
    Ok(p) => ship(p),
    Err(err) => log(err),
}

Most panic-free APIs in the standard library (file reads, socket writes, HashMap::get) report their failures through Result. The signature carries the failure path: what happens if this fails is readable off the function type instead of buried in the docstring.

This approach offers several advantages:

• The failure path becomes part of the signature: a function typed -> dict can raise anything, while a function typed -> Result[dict, ProductError] advertises exactly what can go wrong, and a checker or IDE can see it.
• The compiler forces acknowledgment: callers either handle the error case or the build fails.
• Errors compose: chaining fallible functions is a value flowing through a pipeline, not nested try/except pyramids. Failures short-circuit, errors propagate as data, and the happy path stays straight.

The most popular languages that are used to build the most challening critical systems have this abstraction. Now you can use it in Python too.

Introducing dry-python/returns

Whether your code raises an expected or unexpected failure, nothing in the type system records which is which. dry-python/returns brings a Result monad to Python, as well as other useful primitives. Check the docs to learn more about the library; here we explore the core features of the Result abstraction.

A Result[T, E] is a container with exactly two possible states: Success, wrapping the value of type T, or Failure, wrapping the error of type E. Construction performs no logic; it is just a value:

from returns.result import Failure, Success

order = Success({"id": 1, "total": 42})
failure = Failure(ProductNotFoundError("Product 111 not found"))

The library provides some interesting features on top of these two containers.

First, @safe is a convenient decorator that turns a function that raises into one that returns Result (I’d rather be explicit than using this decorator though):

from returns.result import safe


@safe
def get_product(product_id: int) -> dict[str, int]:
    # raises -> Failure[Exception]; returns -> Success[dict]
    ...

Second, flow composes fallible functions top-to-bottom, the way Rust’s ? operator composes them left-to-right. A Failure short-circuits the remaining steps and hands the error to the end of the chain, so the happy path reads straight down.

The pipeline works on typed domain errors, each a frozen dataclass with a message field, combined into a union error type:

from dataclasses import dataclass
from typing import Any

from returns.result import Result


@dataclass(frozen=True)
class ProductNotFoundError:
    message: str


@dataclass(frozen=True)
class OutOfStockError:
    message: str


type ProductResult = Result[dict[str, Any], ProductNotFoundError | OutOfStockError]

Each pipeline stage advertises the same error union and returns a Failure for its own case:

from returns.result import Failure, Success


def db_get_product(product_id: ProductId) -> ProductResult:
    if product_id == ProductId(111):
        return Failure(ProductNotFoundError(f"Product {product_id} not found"))
    return Success({"id": int(product_id), "stock": 0})


def check_stock(product: dict[str, Any]) -> ProductResult:
    if product["stock"] == 0:
        return Failure(OutOfStockError("Product is out of stock"))
    return Success(product)

flow threads the input through both steps:

from returns.pipeline import flow
from returns.pointfree import bind


result = flow(
    product_id,
    db_get_product,
    bind(check_stock),
)

Third, the error case is a tagged union, so Python’s match gives you the same forced-exhaustiveness Rust gets from its compiler: this fails type checking the day someone adds an error case and forgets to handle it.

match result.failure():
    case ProductNotFoundError(msg):
        print(f"Lookup failed: {msg}")
    case OutOfStockError(msg):
        print(f"Out of stock: {msg}")

The returns library also provides an IO container to model functions with side effects, while Result is intended for pure functions.

Exceptions still have a job

There are still cases where raising exceptions is the right call. As a rule of thumb, I like to think about errors as recoverable (return an error within a Result) or not (raise an exception). Some examples:

• The user is not authorized, or the order is already paid. The error is expected; the caller has to handle it; the type system forces the call site to acknowledge it -> It should use Result.
• The database connection is lost mid-transaction, or the JSON deserializer received malformed input. The error is unexpected; the caller has no good recovery in-domain; it should propagate to the framework’s handler -> It should raise an exception.

import sqlite3

from returns.result import Failure, Success


class ProductStorageError(Exception):
    """The product storage failed in a way the domain cannot recover from."""


def get_product_from_db(product_id: ProductId) -> ProductResult:
    try:
        row = db.execute(
            "SELECT id, stock FROM product WHERE id = ?", (product_id,)
        ).fetchone()
    except sqlite3.Error as err:
        raise ProductStorageError(f"Product storage failure: {err}") from err
    if row is None:
        return Failure(ProductNotFoundError(f"Product {product_id} not found"))
    return Success({"id": row[0], "stock": row[1]})

Takeaways

• Brand primitives with NewType to stop parameter swaps; use Annotated + AfterValidator (Pydantic) or Meta (msgspec) when the brand must also validate at runtime.
• Use Result for expected domain errors and exceptions for unexpected panics.
• Translate third-party and engine exceptions to domain failures at the adapter boundary.
• Rely on Result to carry the failure path; dry-python/returns is the Python implementation of that abstraction.

References and additional resources

• King, Alexis. “Parse, don’t validate” (2019).
• Wlaschin, Scott. Against Railway-Oriented Programming (2019).
• Bernhardt, Gary. “Boundaries” (2014). The functional core, imperative shell framing.
• dry-python/returns and the documentation
• PEP 484 – Type Hints (NewType section)
• PEP 593 – Flexible Type Hints (Annotated)
• Pydantic v2 validators
• msgspec constraints