Consumer

Requests consumer example.

This module defines a simple consumer using the synchronous requests library which will be tested with Pact in the consumer test. As Pact is a consumer-driven framework, the consumer defines the interactions which the provider must then satisfy.

The consumer is the application which makes requests to another service (the provider) and receives a response to process. In this example, we have a simple User class and the consumer fetches a user's information from a HTTP endpoint.

This also showcases how Pact tests differ from merely testing adherence to an OpenAPI specification. The Pact tests are more concerned with the practical use of the API, rather than the formally defined specification. So you will see below that as far as this consumer is concerned, the only information needed from the provider is the user's ID, name, and creation date. This is despite the provider having additional fields in the response.

Note that the code in this module is agnostic of Pact (i.e., this would be your production code). The pact-python dependency only appears in the tests. This is because the consumer is not concerned with Pact, only the tests are.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

User(id: int, name: str, created_on: datetime) dataclass

Represents a user as seen by the consumer.

This class is intentionally minimal, including only the fields the consumer actually uses. It may differ from the provider's user model, which could have additional fields. This demonstrates the consumer-driven nature of contract testing: the consumer defines what it needs, not what the provider exposes.

Attributes

created_on: datetime instance-attribute

id: int instance-attribute

name: str instance-attribute

UserClient(hostname: str, base_path: str | None = None)

HTTP client for interacting with a user provider service.

This class is a simple consumer that fetches user data from a provider over HTTP. It demonstrates how to structure consumer code for use in contract testing, keeping it independent of Pact or any contract testing framework.

PARAMETER	DESCRIPTION
hostname	The base URL of the provider (must include scheme, e.g., http://). TYPE: str
base_path	The base path for the provider's API endpoints. Defaults to /. TYPE: str | None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the hostname does not start with 'http://' or https://.

Source code in examples/http/requests_and_fastapi/consumer.py

def __init__(self, hostname: str, base_path: str | None = None) -> None:
    """
    Initialise the user client.

    Args:
        hostname:
            The base URL of the provider (must include scheme, e.g.,
            `http://`).

        base_path:
            The base path for the provider's API endpoints. Defaults to `/`.

    Raises:
        ValueError:
            If the hostname does not start with 'http://' or `https://`.
    """
    if not hostname.startswith(("http://", "https://")):
        msg = "Invalid base URI"
        raise ValueError(msg)
    self._hostname = hostname
    self._base_path = base_path or "/"
    if not self._base_path.endswith("/"):
        self._base_path += "/"

    self._session = requests.Session()
    logger.debug(
        "Initialised UserClient with base URL: %s%s",
        self.base_url,
        self._base_path,
    )

Attributes

base_path: str property

The base path as a string.

base_url: str property

The base URL as a string.

hostname: str property

The hostname as a string.

This includes the scheme.

Functions

create_user(*, name: str) -> User

Create a new user on the provider.

PARAMETER	DESCRIPTION
name	The name of the user to create. TYPE: str

RETURNS	DESCRIPTION
User	A User instance representing the newly created user.

RAISES	DESCRIPTION
HTTPError	If the server returns a non-2xx response or the request fails.

Source code in examples/http/requests_and_fastapi/consumer.py

def create_user(
    self,
    *,
    name: str,
) -> User:
    """
    Create a new user on the provider.

    Args:
        name:
            The name of the user to create.

    Returns:
        A `User` instance representing the newly created user.

    Raises:
        requests.HTTPError:
            If the server returns a non-2xx response or the request fails.
    """
    logger.debug("Creating user %s", name)
    response = self._session.post(
        f"{self.hostname}{self.base_path}users",
        json={"name": name},
    )
    response.raise_for_status()
    data = response.json()

    # Python < 3.11 don't support ISO 8601 offsets without a colon
    if sys.version_info < (3, 11) and data["created_on"][-4:].isdigit():
        data["created_on"] = data["created_on"][:-2] + ":" + data["created_on"][-2:]
    logger.debug("Created user %s", data["id"])
    return User(
        id=data["id"],
        name=data["name"],
        created_on=datetime.fromisoformat(data["created_on"]),
    )

delete_user(uid: int | User) -> None

Delete a user by ID from the provider.

PARAMETER	DESCRIPTION
uid	The user ID (int) or a User instance to delete. TYPE: int | User

RAISES	DESCRIPTION
HTTPError	If the server returns a non-2xx response or the request fails.

Source code in examples/http/requests_and_fastapi/consumer.py

def delete_user(self, uid: int | User) -> None:
    """
    Delete a user by ID from the provider.

    Args:
        uid:
            The user ID (int) or a `User` instance to delete.

    Raises:
        requests.HTTPError:
            If the server returns a non-2xx response or the request fails.
    """
    if isinstance(uid, User):
        uid = uid.id
    logger.debug("Deleting user %s", uid)
    response = self._session.delete(f"{self.hostname}{self.base_path}users/{uid}")
    response.raise_for_status()

get_user(user_id: int) -> User

Fetch a user by ID from the provider.

This method demonstrates how a consumer fetches only the data it needs from a provider, regardless of what else the provider may return.

PARAMETER	DESCRIPTION
user_id	The ID of the user to fetch. TYPE: int

RETURNS	DESCRIPTION
User	A User instance representing the fetched user.

RAISES	DESCRIPTION
HTTPError	If the server returns a non-2xx response or the request fails.

Source code in examples/http/requests_and_fastapi/consumer.py

def get_user(self, user_id: int) -> User:
    """
    Fetch a user by ID from the provider.

    This method demonstrates how a consumer fetches only the data it needs
    from a provider, regardless of what else the provider may return.

    Args:
        user_id:
            The ID of the user to fetch.

    Returns:
        A `User` instance representing the fetched user.

    Raises:
        requests.HTTPError:
            If the server returns a non-2xx response or the request fails.
    """
    logger.debug("Fetching user %s", user_id)
    response = self._session.get(f"{self.hostname}{self.base_path}users/{user_id}")
    response.raise_for_status()
    data: dict[str, Any] = response.json()

    # Python < 3.11 don't support ISO 8601 offsets without a colon
    if sys.version_info < (3, 11) and data["created_on"][-4:].isdigit():
        data["created_on"] = data["created_on"][:-2] + ":" + data["created_on"][-2:]
    return User(
        id=data["id"],
        name=data["name"],
        created_on=datetime.fromisoformat(data["created_on"]),
    )