Skip to content

Consumer

Simple Consumer Implementation.

This modules defines a simple consumer 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.

Note that the code in this module is agnostic of Pact. The pact-python dependency only appears in the tests. This is because the consumer is not concerned with Pact, only the tests are.

Classes

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

User data class.

Attributes

created_on: datetime instance-attribute
id: int instance-attribute
name: str instance-attribute

UserConsumer(base_uri: str)

Example consumer.

This class defines a simple consumer which will interact with a provider over HTTP to fetch a user's information, and then return an instance of the User class.

PARAMETER DESCRIPTION
base_uri

The uri of the provider

TYPE: str

Source code in examples/src/consumer.py 
71
72
73
74
75
76
77
78
def __init__(self, base_uri: str) -> None:
    """
    Initialise the consumer.

    Args:
        base_uri: The uri of the provider
    """
    self.base_uri = base_uri

Attributes

base_uri = base_uri instance-attribute

Functions

get_user(user_id: int) -> User

Fetch a user by ID from the server.

PARAMETER DESCRIPTION
user_id

The ID of the user to fetch.

TYPE: int

RETURNS DESCRIPTION
User

The user if found.
User

In all other cases, an error dictionary is returned with the key
User

error and the value as the error message.
RAISES DESCRIPTION
HTTPError

If the server returns a non-200 response.
Source code in examples/src/consumer.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_user(self, user_id: int) -> User:
    """
    Fetch a user by ID from the server.

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

    Returns:
        The user if found.

        In all other cases, an error dictionary is returned with the key
        `error` and the value as the error message.

    Raises:
        requests.HTTPError: If the server returns a non-200 response.
    """
    uri = f"{self.base_uri}/users/{user_id}"
    response = requests.get(uri, timeout=5)
    response.raise_for_status()
    data: Dict[str, Any] = response.json()
    return User(
        id=data["id"],
        name=data["name"],
        created_on=datetime.fromisoformat(data["created_on"]),
    )