Implementing a Metaclass for Class-Level Contract Validation

Good to see you again. In the previous lesson, you traced the machinery behind a class statement: Python prepares a namespace, executes the class body, and calls the selected metaclass to produce the class object. You also saw that __init_subclass__ runs only after that object has been constructed.

Now you will use the earlier point in that lifecycle: a metaclass’s __new__. The goal is to enforce a contract when a handler class is defined, rather than allowing a malformed handler to remain dormant until a worker tries to use it. We will build a small contract for job handlers that requires each concrete handler to declare its routing configuration and its asynchronous handle implementation directly in its own class body.

---

Why validate a class at definition time?

A job-processing service needs handler classes whose configuration is reliable. Consider a worker that chooses a handler based on a job type:

class ResizeImageHandler(JobHandler):
    job_type = "resize_image"
    max_attempts = 3

    async def handle(self, payload: dict[str, object]) -> None:
        ...

A missing job_type is not a normal runtime input error. It is a programming error in the class declaration. Discovering it when the application imports its handlers is far better than discovering it after a production job has already been claimed.

Watch the central example in this short segment before implementing the pattern.

A useful distinction is between two possible policies:

Policy	Question being asked	Example
Available contract	“Does this class have this attribute, perhaps through inheritance?”	getattr(cls, "max_attempts")
Declaration contract	“Did this class body explicitly declare this attribute?”	"max_attempts" in namespace

For handler configuration, the second policy is often safer. Each concrete handler should state its own job_type and retry limit, rather than silently inheriting a sibling or parent handler’s values.

A metaclass receives the namespace produced by the current class body. That makes it a natural place to enforce this declaration-level contract.

---

The precise validation boundary

The validation happens after the class body has executed but before the metaclass delegates to type.__new__ to create the class object. This ordering is why the metaclass can inspect declarations such as job_type, max_attempts, and handle.

There are two consequences worth keeping clear:

1. A class-body error happens before metaclass validation. If evaluating max_attempts = unknown_name raises NameError, the metaclass is never reached.
2. A contract error happens during class creation. If the class body executes successfully but omits job_type, the metaclass raises TypeError, and the class name is never bound.

This is definition-time validation, usually during module import or test collection. It is not compile-time checking in the static-language sense.

---

Build a handler-contract metaclass

Create experiments/handler_contract.py and start with this implementation:

from __future__ import annotations

import inspect


class HandlerContractMeta(type):
    """Validate declarations made by concrete job-handler classes."""

    def __new__(
        mcls,
        name: str,
        bases: tuple[type, ...],
        namespace: dict[str, object],
        **class_options: object,
    ) -> type:
        # This is a local marker: inherited values do not affect it.
        is_template = namespace.get("__contract_template__", False)

        if type(is_template) is not bool:
            raise TypeError(
                f"{name}.__contract_template__ must be a bool"
            )

        if not is_template:
            required = ("job_type", "max_attempts", "handle")
            missing = [
                attribute
                for attribute in required
                if attribute not in namespace
            ]

            if missing:
                names = ", ".join(missing)
                raise TypeError(
                    f"{name} must declare {names} in its class body"
                )

            job_type = namespace["job_type"]
            if not isinstance(job_type, str) or not job_type.strip():
                raise TypeError(
                    f"{name}.job_type must be a non-empty string"
                )

            max_attempts = namespace["max_attempts"]
            if (
                isinstance(max_attempts, bool)
                or not isinstance(max_attempts, int)
                or max_attempts < 1
            ):
                raise TypeError(
                    f"{name}.max_attempts must be a positive integer"
                )

            raw_handle = namespace["handle"]
            if isinstance(raw_handle, (classmethod, staticmethod)):
                raise TypeError(
                    f"{name}.handle must be an instance method"
                )

            if not inspect.iscoroutinefunction(raw_handle):
                raise TypeError(
                    f"{name}.handle must be declared with async def"
                )

        return super().__new__(
            mcls,
            name,
            bases,
            namespace,
            **class_options,
        )

Then define the framework base and one valid concrete implementation:

class JobHandler(metaclass=HandlerContractMeta):
    __contract_template__ = True

    async def handle(self, payload: dict[str, object]) -> None:
        raise NotImplementedError


class SendDigestHandler(JobHandler):
    job_type = "send_digest"
    max_attempts = 3

    async def handle(self, payload: dict[str, object]) -> None:
        recipient = payload["recipient"]
        print(f"Sending digest to {recipient}")

These assertions should succeed:

assert type(JobHandler) is HandlerContractMeta
assert type(SendDigestHandler) is HandlerContractMeta
assert SendDigestHandler.job_type == "send_digest"
assert SendDigestHandler.max_attempts == 3

Read the code in construction order

When Python evaluates SendDigestHandler, the metaclass receives a namespace conceptually resembling this:

{
    "__module__": "__main__",
    "__qualname__": "SendDigestHandler",
    "job_type": "send_digest",
    "max_attempts": 3,
    "handle": <function SendDigestHandler.handle>,
}

The validation deliberately uses namespace, rather than asking questions of the finished class with getattr.

For example, this class is rejected:

class IncompleteHandler(JobHandler):
    async def handle(self, payload: dict[str, object]) -> None:
        pass

It inherits no useful concrete configuration because it must define job_type and max_attempts itself. This is exactly what the membership check enforces:

"job_type" in namespace

By contrast, the following check would mean something weaker:

hasattr(SomeFinishedHandler, "job_type")

It could return True merely because a parent class supplied the attribute.

Why the template marker is read from namespace

JobHandler itself is a framework template, not a usable handler, so it sets:

__contract_template__ = True

The metaclass looks only in the namespace being constructed:

is_template = namespace.get("__contract_template__", False)

Therefore SendDigestHandler does not inherit the exemption. Its namespace has no __contract_template__, so is_template becomes False and the full contract is checked.

An intermediate template may opt in explicitly:

class NotificationHandler(JobHandler):
    __contract_template__ = True

    async def handle(self, payload: dict[str, object]) -> None:
        raise NotImplementedError

A real implementation below it must still make all required declarations:

class SendWeeklyReportHandler(NotificationHandler):
    job_type = "send_weekly_report"
    max_attempts = 5

    async def handle(self, payload: dict[str, object]) -> None:
        print("Sending weekly report")

The marker only controls this metaclass’s validation. It does not make a class non-instantiable in the way abc.ABC and @abstractmethod do. Instantiation policy is a separate design concern.

---

Validate semantics, not merely names

Checking for attribute names alone is insufficient. A declaration such as this should not satisfy the contract:

class BrokenRetryHandler(JobHandler):
    job_type = "retry_broken"
    max_attempts = True

    async def handle(self, payload: dict[str, object]) -> None:
        pass

Although bool is technically a subclass of int in Python, True is not a meaningful retry count. This is why the validation explicitly rejects booleans before accepting positive integers:

if (
    isinstance(max_attempts, bool)
    or not isinstance(max_attempts, int)
    or max_attempts < 1
):
    ...

Similarly, an empty or whitespace-only route key should fail fast:

class UnroutableHandler(JobHandler):
    job_type = "   "
    max_attempts = 2

    async def handle(self, payload: dict[str, object]) -> None:
        pass

Finally, the handle checks establish a deliberately narrow protocol:

• handle must be declared in the concrete class body.
• It must be an ordinary instance method, not a @classmethod or @staticmethod.
• It must be declared with async def.

The metaclass does not inspect the full function signature. Signature validation is possible with inspect.signature, but becomes fragile quickly: optional dependencies, decorators, positional-only parameters, and evolving payload types can make an overly rigid check expensive to maintain. Validate only properties that the framework genuinely requires.

---

Test the contract as class-creation behavior

Because errors occur when a class is defined, tests should define invalid classes inside a pytest.raises context. Put the following in tests/test_handler_contract.py:

import pytest

from experiments.handler_contract import (
    HandlerContractMeta,
    JobHandler,
)


def test_valid_handler_is_created_with_the_contract_metaclass() -> None:
    class PurgeExpiredHandler(JobHandler):
        job_type = "purge_expired"
        max_attempts = 2

        async def handle(self, payload: dict[str, object]) -> None:
            pass

    assert type(PurgeExpiredHandler) is HandlerContractMeta
    assert PurgeExpiredHandler.job_type == "purge_expired"
    assert PurgeExpiredHandler.max_attempts == 2


def test_missing_configuration_fails_when_the_class_is_defined() -> None:
    with pytest.raises(
        TypeError,
        match="must declare job_type, max_attempts",
    ):

        class MissingConfigurationHandler(JobHandler):
            async def handle(
                self,
                payload: dict[str, object],
            ) -> None:
                pass


def test_inherited_configuration_does_not_satisfy_the_contract() -> None:
    class ParentHandler(JobHandler):
        job_type = "parent"
        max_attempts = 1

        async def handle(self, payload: dict[str, object]) -> None:
            pass

    with pytest.raises(
        TypeError,
        match="must declare job_type, max_attempts",
    ):

        class ChildHandler(ParentHandler):
            async def handle(
                self,
                payload: dict[str, object],
            ) -> None:
                pass


def test_boolean_retry_limit_is_rejected() -> None:
    with pytest.raises(
        TypeError,
        match="max_attempts must be a positive integer",
    ):

        class InvalidRetryHandler(JobHandler):
            job_type = "invalid_retry"
            max_attempts = True

            async def handle(
                self,
                payload: dict[str, object],
            ) -> None:
                pass

Run:

pytest -q tests/test_handler_contract.py

The second and third tests are especially important. They prove the metaclass enforces local declarations, not merely inherited availability.

For a short implementation extension, add one more valid handler with a distinct job type and retry policy. Then temporarily make its handle method synchronous by replacing async def with def. The resulting failing test should identify the contract violation at the class-definition boundary.

---

When this is—and is not—the right tool

The previous lesson’s __init_subclass__ hook can also validate subclasses, often with less machinery. In fact, for ordinary registration and configuration checks, it is normally the better default.

This metaclass is justified when you specifically need access to the original class-body namespace before normal type construction finishes, or when a framework already owns class creation through a metaclass. In this lesson, that access lets us distinguish these two cases reliably:

class DeclaredHere:
    job_type = "local"

class InheritedFromParent(DeclaredHere):
    pass

The first namespace contains job_type; the second does not.

Keep the metaclass narrow:

• Raise clear, class-specific errors.
• Validate stable framework requirements rather than incidental style preferences.
• Delegate to super().__new__ with the original namespace and class options.
• Avoid silently rewriting class declarations unless transformation is a genuine part of the framework’s API.

The final point preserves normal class-creation behavior, including the handling required for zero-argument super().

---

Takeaways

• A metaclass can reject malformed classes at definition time by validating the namespace passed to __new__.
• Checking namespace enforces what the current class body declares; checking a completed class with getattr or hasattr can accidentally accept inherited values.
• A useful handler contract can require a non-empty routing key, a positive retry count, and a locally declared asynchronous instance method.
• Contract failures should raise TypeError during class creation and be tested with classes defined inside pytest.raises.
• __init_subclass__ remains the simpler choice for many subclass hooks; use a metaclass when pre-creation namespace control is genuinely needed.

Next, the course moves from how Python creates classes to how it resolves attributes on them. You will begin with descriptors and learn why some class attributes take precedence over instance attributes.