Add Http Post and Delete examples

.pre-commit-config.yaml

@@ -81,5 +81,5 @@ repos:
         entry: hatch run mypy
         language: system
         types: [python]
-        exclude: ^(src/pact|tests)/(?!v3/).*\.py$
+        exclude: ^(src/pact|tests|examples/tests)/(?!v3/).*\.py$
         stages: [pre-push]

examples/.ruff.toml

@@ -2,9 +2,10 @@ extend = "../pyproject.toml"
 
 [lint]
 ignore = [
-  "S101", # Forbid assert statements
-  "D103", # Require docstring in public function
-  "D104", # Require docstring in public package
+  "S101",    # Forbid assert statements
+  "D103",    # Require docstring in public function
+  "D104",    # Require docstring in public package
+  "PLR2004", # Forbid Magic Numbers
 ]
 
 [lint.per-file-ignores]

examples/src/consumer.py

@@ -12,6 +12,13 @@
 [`User`][examples.src.consumer.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. The `pact-python`
 dependency only appears in the tests. This is because the consumer is not
 concerned with Pact, only the tests are.
@@ -102,3 +109,47 @@ def get_user(self, user_id: int) -> User:
             name=data["name"],
             created_on=datetime.fromisoformat(data["created_on"]),
         )
+
+    def create_user(
+        self,
+        *,
+        name: str,
+    ) -> User:
+        """
+        Create a new user on the server.
+
+        Args:
+            name: The name of the user to create.
+
+        Returns:
+            The user, if successfully created.
+
+        Raises:
+            requests.HTTPError: If the server returns a non-200 response.
+        """
+        uri = f"{self.base_uri}/users/"
+        response = requests.post(uri, json={"name": name}, 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"]),
+        )
+
+    def delete_user(self, uid: int | User) -> None:
+        """
+        Delete a user by ID from the server.
+
+        Args:
+            uid: The user ID or user object to delete.
+
+        Raises:
+            requests.HTTPError: If the server returns a non-200 response.
+        """
+        if isinstance(uid, User):
+            uid = uid.id
+
+        uri = f"{self.base_uri}/users/{uid}"
+        response = requests.delete(uri, timeout=5)
+        response.raise_for_status()

examples/src/fastapi.py

@@ -12,19 +12,68 @@
 (the consumer) and returns a response. In this example, we have a simple
 endpoint which returns a user's information from a (fake) database.
 
+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. The User class
+defined here has additional fields which are not used by the consumer. Should
+the provider later decide to add or remove fields, Pact's consumer-driven
+testing will provide feedback on whether the consumer is compatible with the
+provider's changes.
+
 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.
 """
 
 from __future__ import annotations
 
+import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime
 from typing import Any, Dict
 
-from fastapi import FastAPI
-from fastapi.responses import JSONResponse
+from fastapi import FastAPI, HTTPException
 
 app = FastAPI()
+logger = logging.getLogger(__name__)
+
+
+@dataclass()
+class User:
+    """User data class."""
+
+    id: int
+    name: str
+    created_on: datetime
+    email: str | None
+    ip_address: str | None
+    hobbies: list[str]
+    admin: bool
+
+    def __post_init__(self) -> None:
+        """
+        Validate the User data.
+
+        This performs the following checks:
+
+        - The name cannot be empty
+        - The id must be a positive integer
+
+        Raises:
+            ValueError: If any of the above checks fail.
+        """
+        if not self.name:
+            msg = "User must have a name"
+            raise ValueError(msg)
+
+        if self.id < 0:
+            msg = "User ID must be a positive integer"
+            raise ValueError(msg)
+
+    def __repr__(self) -> str:
+        """Return the user's name."""
+        return f"User({self.id}:{self.name})"
+
 
 """
 As this is a simple example, we'll use a simple dict to represent a database.
@@ -34,11 +83,11 @@
 be mocked out to avoid the need for a real database. An example of this can be
 found in the [test suite][examples.tests.test_01_provider_fastapi].
 """
-FAKE_DB: Dict[int, Dict[str, Any]] = {}
+FAKE_DB: Dict[int, User] = {}
 
 
 @app.get("/users/{uid}")
-async def get_user_by_id(uid: int) -> JSONResponse:
+async def get_user_by_id(uid: int) -> User:
     """
     Fetch a user by their ID.
 
@@ -50,5 +99,48 @@ async def get_user_by_id(uid: int) -> JSONResponse:
     """
     user = FAKE_DB.get(uid)
     if not user:
-        return JSONResponse(status_code=404, content={"error": "User not found"})
-    return JSONResponse(status_code=200, content=user)
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
+
+
+@app.post("/users/")
+async def create_new_user(user: dict[str, Any]) -> User:
+    """
+    Create a new user .
+
+    Args:
+        user: The user data to create
+
+    Returns:
+        The status code 200 and user data if successfully created, HTTP 404 if not
+    """
+    if "id" in user:
+        raise HTTPException(status_code=400, detail="ID should not be provided.")
+    uid = len(FAKE_DB)
+    FAKE_DB[uid] = User(
+        id=uid,
+        name=user["name"],
+        created_on=datetime.now(tz=UTC),
+        email=user.get("email"),
+        ip_address=user.get("ip_address"),
+        hobbies=user.get("hobbies", []),
+        admin=user.get("admin", False),
+    )
+    return FAKE_DB[uid]
+
+
+@app.delete("/users/{uid}", status_code=204)
+async def delete_user(uid: int):  # noqa: ANN201
+    """
+     Delete an existing user .
+
+    Args:
+        uid: The ID of the user to delete
+
+    Returns:
+        The status code 204, HTTP 404 if not
+    """
+    if uid not in FAKE_DB:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    del FAKE_DB[uid]

examples/src/flask.py

@@ -19,12 +19,68 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Union
+import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Dict, Tuple
 
-from flask import Flask
+from flask import Flask, Response, abort, jsonify, request
 
+logger = logging.getLogger(__name__)
 app = Flask(__name__)
 
+
+@dataclass()
+class User:
+    """User data class."""
+
+    id: int
+    name: str
+    created_on: datetime
+    email: str | None
+    ip_address: str | None
+    hobbies: list[str]
+    admin: bool
+
+    def __post_init__(self) -> None:
+        """
+        Validate the User data.
+
+        This performs the following checks:
+
+        - The name cannot be empty
+        - The id must be a positive integer
+
+        Raises:
+            ValueError: If any of the above checks fail.
+        """
+        if not self.name:
+            msg = "User must have a name"
+            raise ValueError(msg)
+
+        if self.id < 0:
+            msg = "User ID must be a positive integer"
+            raise ValueError(msg)
+
+    def __repr__(self) -> str:
+        """Return the user's name."""
+        return f"User({self.id}:{self.name})"
+
+    def dict(self) -> dict[str, Any]:
+        """
+        Return the user's data as a dict.
+        """
+        return {
+            "id": self.id,
+            "name": self.name,
+            "created_on": self.created_on.isoformat(),
+            "email": self.email,
+            "ip_address": self.ip_address,
+            "hobbies": self.hobbies,
+            "admin": self.admin,
+        }
+
+
 """
 As this is a simple example, we'll use a simple dict to represent a database.
 This would be replaced with a real database in a real application.
@@ -33,11 +89,11 @@
 be mocked out to avoid the need for a real database. An example of this can be
 found in the [test suite][examples.tests.test_01_provider_flask].
 """
-FAKE_DB: Dict[int, Dict[str, Any]] = {}
+FAKE_DB: Dict[int, User] = {}
 
 
-@app.route("/users/<uid>")
-def get_user_by_id(uid: int) -> Union[Dict[str, Any], tuple[Dict[str, Any], int]]:
+@app.route("/users/<int:uid>")
+def get_user_by_id(uid: int) -> Response | Tuple[Response, int]:
     """
     Fetch a user by their ID.
 
@@ -49,5 +105,32 @@ def get_user_by_id(uid: int) -> Union[Dict[str, Any], tuple[Dict[str, Any], int]
     """
     user = FAKE_DB.get(uid)
     if not user:
-        return {"error": "User not found"}, 404
-    return user
+        return jsonify({"detail": "User not found"}), 404
+    return jsonify(user.dict())
+
+
+@app.route("/users/", methods=["POST"])
+def create_user() -> Response:
+    if request.json is None:
+        abort(400, description="Invalid JSON data")
+
+    user: Dict[str, Any] = request.json
+    uid = len(FAKE_DB)
+    FAKE_DB[uid] = User(
+        id=uid,
+        name=user["name"],
+        created_on=datetime.now(tz=UTC),
+        email=user.get("email"),
+        ip_address=user.get("ip_address"),
+        hobbies=user.get("hobbies", []),
+        admin=user.get("admin", False),
+    )
+    return jsonify(FAKE_DB[uid].dict())
+
+
+@app.route("/users/<int:uid>", methods=["DELETE"])
+def delete_user(uid: int) -> Tuple[str | Response, int]:
+    if uid not in FAKE_DB:
+        return jsonify({"detail": "User not found"}), 404
+    del FAKE_DB[uid]
+    return "", 204