User service

FastAPI service acting as both a Pact consumer and a Pact provider.

This module is the centrepiece of the example. user-service sits in the middle of a two-hop request path:

frontend-web  →  user-service  →  auth-service

This means it plays two Pact roles simultaneously:

• Provider of the POST /accounts endpoint consumed by frontend-web. The provider verification test lives in test_provider.

• Consumer of auth-service's POST /auth/validate endpoint. The consumer contract test lives in test_consumer_auth.

Testability design

Provider verification requires the service to be started as a real HTTP server. To avoid needing a real auth-service during those tests, this module uses the CredentialsVerifier protocol as a seam. In production the seam is filled by AuthClient; in tests it is replaced with a StubAuthVerifier via set_auth_verifier.

This avoids mocking at the HTTP level. The real FastAPI application runs, and only the collaborator that calls auth-service is swapped out.

Module-level state

SERVICE_STATE and ACCOUNT_STORE are intentional module-level globals. Because provider state handlers in Pact run in the same process as the application, these globals are the simplest way for the test harness to reconfigure the service between interactions without an additional HTTP endpoint.

Attributes

ACCOUNT_STORE = InMemoryAccountStore() module-attribute

SERVICE_STATE = _ServiceState() module-attribute

app = FastAPI() module-attribute

Classes

CreateAccountRequest pydantic-model

Bases: BaseModel

Request payload used by frontend-web.

Fields:

• username (str)
• password (str)

Attributes

password: str pydantic-field

username: str pydantic-field

CreateAccountResponse pydantic-model

Bases: BaseModel

Response payload returned to frontend-web.

Fields:

• id (int)
• username (str)
• status (str)

Attributes

id: int pydantic-field

status: str = 'created' pydantic-field

username: str pydantic-field

CredentialsVerifier

Bases: Protocol

              flowchart TD
              examples.http.service_consumer_provider.user_service.CredentialsVerifier[CredentialsVerifier]

              

              click examples.http.service_consumer_provider.user_service.CredentialsVerifier href "" "examples.http.service_consumer_provider.user_service.CredentialsVerifier"
            

Behaviour required for credential verification.

Functions

validate_credentials(username: str, password: str) -> bool

Validate credentials.

Source code in examples/http/service_consumer_provider/user_service.py

def validate_credentials(self, username: str, password: str) -> bool:
    """
    Validate credentials.
    """

InMemoryAccountStore()

Small in-memory store for example purposes.

Source code in examples/http/service_consumer_provider/user_service.py

def __init__(self) -> None:
    """
    Initialise the in-memory store.
    """
    self._next_id = 1
    self._accounts: dict[int, UserAccount] = {}

Functions

create(username: str) -> UserAccount

Create and store a new account.

PARAMETER	DESCRIPTION
username	Username for the new account. TYPE: str

RETURNS	DESCRIPTION
UserAccount	The created account.

Source code in examples/http/service_consumer_provider/user_service.py

def create(self, username: str) -> UserAccount:
    """
    Create and store a new account.

    Args:
        username:
            Username for the new account.

    Returns:
        The created account.
    """
    account = UserAccount(id=self._next_id, username=username)
    self._accounts[account.id] = account
    self._next_id += 1
    return account

reset() -> None

Reset all stored accounts.

Source code in examples/http/service_consumer_provider/user_service.py

def reset(self) -> None:
    """
    Reset all stored accounts.
    """
    self._next_id = 1
    self._accounts.clear()

UserAccount(id: int, username: str) dataclass

Stored account record.

Attributes

id: int instance-attribute

username: str instance-attribute

Functions

create_account(payload: CreateAccountRequest) -> CreateAccountResponse async

Create an account after validating credentials with auth-service.

PARAMETER	DESCRIPTION
payload	Account request payload. TYPE: CreateAccountRequest

RETURNS	DESCRIPTION
CreateAccountResponse	Created account response.

RAISES	DESCRIPTION
HTTPException	If credentials are invalid.

Source code in examples/http/service_consumer_provider/user_service.py

@app.post("/accounts", status_code=status.HTTP_201_CREATED)
async def create_account(payload: CreateAccountRequest) -> CreateAccountResponse:
    """
    Create an account after validating credentials with auth-service.

    Args:
        payload:
            Account request payload.

    Returns:
        Created account response.

    Raises:
        HTTPException:
            If credentials are invalid.
    """
    if not SERVICE_STATE.auth_verifier.validate_credentials(
        payload.username,
        payload.password,
    ):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    account = ACCOUNT_STORE.create(payload.username)
    return CreateAccountResponse(id=account.id, username=account.username)

reset_state() -> None

Reset internal provider state.

Source code in examples/http/service_consumer_provider/user_service.py

def reset_state() -> None:
    """
    Reset internal provider state.
    """
    ACCOUNT_STORE.reset()

set_auth_verifier(verifier: CredentialsVerifier) -> None

Replace the auth verifier implementation.

PARAMETER	DESCRIPTION
verifier	New verifier implementation. TYPE: CredentialsVerifier

Source code in examples/http/service_consumer_provider/user_service.py

def set_auth_verifier(verifier: CredentialsVerifier) -> None:
    """
    Replace the auth verifier implementation.

    Args:
        verifier:
            New verifier implementation.
    """
    SERVICE_STATE.auth_verifier = verifier