Update to latest Specmatic, and Order API v3

Updated to the latest API specifications, added the missing API routes, implemented error handling, and included stub data, among other improvements.

.env

@@ -0,0 +1,2 @@
+ORDER_API_HOST = 127.0.0.1
+ORDER_API_PORT = 8080

.github/workflows/main.yml

@@ -23,7 +23,7 @@ jobs:
     - name: Setup python
       uses: actions/setup-python@v4
       with:
-          python-version: '3.8'
+          python-version: '3.12'
           cache: 'pip'
     - name: Run pip install
       working-directory: main

.gitignore

@@ -6,4 +6,8 @@ specmatic.jar
 specmatic_python.egg-info/*
 dist/*
 venv
-.specmatic
+.venv
+.specmatic
+__pycache__
+.vscode
+build

README.md

@@ -1,54 +1,77 @@
-This is a Python implementation of the [Specmatic Order BFF Service](https://github.com/znsio/specmatic-order-ui)
-project.  
-The implementation is based on the [Sanic](https://sanic.dev/en/) framework.
 
-The open api contract for the services is defined in
-the [Specmatic Central Contract Repository](https://github.com/znsio/specmatic-order-contracts/blob/main/in/specmatic/examples/store/api_order_v1.yaml)
+# Specmatic Sample: Python-Sanic BFF
 
-The order bff service internally calls the order api service (on port 9090).
+* [Specmatic Website](https://specmatic.in)
+* [Specmatic Documentation](https://specmatic.in/documentation.html)
 
-The purpose of this project is to demonstrate how specmatic can be used to validate the contract of the bff service
-while stubbing out the order api service at the same time.
+This example project illustrates the practice of contract-driven development and contract testing within a Sanic (Python) application that relies on an external domain service. In this context, Specmatic is utilized to stub calls to domain API services according to its OpenAPI specification.
 
-```Dev Setup```
+Here is the domain api [contract/open api spec](https://github.com/znsio/specmatic-order-contracts/blob/main/in/specmatic/examples/store/api_order_v3.yaml)
 
-- Install Python 3.11 ( use homebrew if you are on mac os)
+## Definitions
 
-- Install JRE 17 or later
+* BFF: Backend for Frontend
+* Domain API: API managing the domain model
+* Specmatic Stub/Mock Server: Generate a server that simulates a real service using its OpenAPI or AsyncAPI specification
 
-- Install pip
+## Background
 
-- Install virtualenv by running:  
-  ```pip install virtualenv```
+A standard web application setup may resemble the following structure. By leveraging Specmatic, we can engage in contract-driven development and perform comprehensive testing on all the components listed below. In this illustrative project, we demonstrate the process for a Sanic BFF, which relies on the Domain API Service, showcasing OpenAPI support within **Specmatic**.
 
+![HTML client talks to client API which talks to backend API](assets/specmatic-order-bff-architecture.gif)
 
-- Clone the git repository
+## Tech
 
+1. Sanic
+2. Specmatic
+3. PyTest
+4. Coverage
 
-- **Virtual Environment Setup**
-    - Create a "virtual environment" named 'venv' by running:  
-      ```virtualenv venv ```
+## Setup
 
-      This will create a virtual environment using the default python installation.  
-      If you wish to provide a specific python installation, run:  
-      ```virtualenv venv --python="/opt/homebrew/opt/[email protected]/libexec/bin/python"```
+1. Install [Python 3.12](https://www.python.org/)
+2. Install JRE 17 or later.
 
-    - To activate your virtual environment, execute this from a terminal window in your root folder:  
-      ```source venv/bin/activate```
+## Setup Virtual Environment
 
+1. ### Create a virtual environment named ".venv" by executing the following command in the terminal from the project's root directory
 
-- **Install project requirements**  
-  From a terminal window in your root folder, run:  
-  ``` pip install -r requirements.txt```
+   ```shell
+    python -m venv .venv
+    ```
 
+2. ### Activate virtual environment by executing
 
-- **Run Tests**  
-  Download the Specmatic standalone jar from the [specmatic website](https://specmatic.in/getting_started.html)  
-  Open a terminal window in the root folder and run:  
-  ```pytest test -v -s```
+* **on MacOS and Linux**
 
-  If you see any errors regarding missing pcakages, try reloading the virtual env:  
-  ```deactivate```   
-  ```source venv/bin/activate```  
-
-
+   ```shell
+   source .venv/bin/activate
+   ```
+
+* **on Windows CMD**
+
+  ```cmd
+  .venv\Scripts\activate.bat
+  ```
+
+* **on Windows Powershell (you may need to adjust the ExecutionPolicy)**
+
+  ```powershell
+  .\.venv\Scripts\Activate.ps1
+  ```
+
+## Install Dependencies
+
+To install all necessary dependencies for this project, navigate to the project's root directory in your terminal and execute
+
+```shell
+pip install -r requirements.txt
+```
+
+## Execute Tests and Validate Contracts with Specmatic
+
+Executing this command will initiate Specmatic and execute the tests on the Sanic application.
+
+```shell
+pytest test -v -s
+```

api/__init__.py

@@ -1,14 +1,50 @@
-import configparser
+import os
+from datetime import UTC, datetime
 
-from sanic import Sanic
-from definitions import ROOT_DIR
-
-config = configparser.ConfigParser()
-config.read(ROOT_DIR + '/cfg.ini')
+from dotenv import load_dotenv
+from marshmallow import ValidationError
+from sanic import Sanic, json
+from sanic.exceptions import HTTPException
 
+load_dotenv()
 app = Sanic("OrderBFF")
-app.config["ORDER_API_HOST"] = config.get('dev', 'ORDER_API_HOST')
-app.config["ORDER_API_PORT"] = config.get('dev', 'ORDER_API_PORT')
+app.config["ORDER_API_HOST"] = os.getenv("ORDER_API_HOST")
+app.config["ORDER_API_PORT"] = os.getenv("ORDER_API_PORT")
+app.config["API_URL"] = f"http://{app.config['ORDER_API_HOST']}:{app.config['ORDER_API_PORT']}"
+app.config["AUTH_TOKEN"] = os.getenv("AUTH_TOKEN") or "API-TOKEN-SPEC"
+app.config["REQ_TIMEOUT"] = os.getenv("REQ_TIMEOUT") or 3000
+
+
+@app.exception(ValidationError)
+async def handle_marshmallow_validation_error(_, exc: "ValidationError"):
+    # NOTE:API SPEC V4 specifies that message should be a string not an object / array
+    errors = "\n".join([f"{field}: {", ".join(err)}" for field, err in exc.normalized_messages().items()])
+    return json(
+        {
+            "timestamp": datetime.now(tz=UTC).isoformat(),
+            "status": 400,
+            "error": "Bad Request",
+            "message": errors,
+        },
+        status=400,
+    )
+
+
+@app.exception(HTTPException)
+async def http_error_handler(_, exception: "HTTPException"):
+    return json(
+        {
+            "timestamp": datetime.now(tz=UTC).isoformat(),
+            "status": exception.status_code,
+            "error": exception.__class__.__name__,
+            "message": str(exception),
+        },
+        status=exception.status_code,
+    )
+
 
+from api.orders.routes import orders  # noqa: E402
+from api.products.routes import products  # noqa: E402
 
-from api.routes import *
+app.blueprint(products)
+app.blueprint(orders)

api/orders/models.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+from typing import Any
+
+from api.schemas import OrderSchema, OrderStatus
+
+order_schema = OrderSchema()
+
+
+@dataclass
+class Order:
+    status: OrderStatus
+    productid: int
+    count: int
+
+    @staticmethod
+    def load(data: Any):
+        data = order_schema.load(data)  # type: ignore[reportAssignmentType]
+        return Order(**data)
+
+    def asdict(self):
+        return {"status": self.status.value, "productid": self.productid, "count": self.count}

api/orders/routes.py

@@ -0,0 +1,18 @@
+from typing import TYPE_CHECKING
+
+from sanic import Blueprint, json
+
+from api.orders.models import Order
+from api.services import OrdersService
+
+if TYPE_CHECKING:
+    from sanic import Request
+
+orders = Blueprint("orders")
+
+
+@orders.route("/orders", methods=["POST"])
+async def create_order(request: "Request"):
+    data: Order = Order.load(request.json)
+    order = OrdersService.create_order(data)
+    return json({"id": order["id"]}, status=201)

api/products/models.py

@@ -0,0 +1,45 @@
+from dataclasses import dataclass
+from typing import Any
+
+from sanic import json
+
+from api.schemas import AvailableProductSchema, ProductSchema, ProductType
+
+avail_prod_schema = AvailableProductSchema()
+product_schema = ProductSchema()
+
+
+@dataclass
+class Product:
+    name: str
+    type: ProductType
+    inventory: int
+    id: int | None = None
+    description: str | None = ""
+
+    @staticmethod
+    def validate_args(page_size: str | None, p_type: str | None) -> tuple[int, ProductType | None]:
+        args = {"page_size": page_size, "type": p_type}
+        data: dict = avail_prod_schema.load(args)  # type: ignore[reportAssignmentType]
+        return data.get("page_size"), data.get("type")  # type: ignore[reportReturnType]
+
+    @staticmethod
+    def load(data: Any):
+        data = product_schema.load(data)  # type: ignore[reportAssignmentType]
+        return Product(**data)
+
+    @staticmethod
+    def load_many(data: Any):
+        data = product_schema.load(data, many=True)  # type: ignore[reportAssignmentType]
+        return [Product(**d) for d in data]
+
+    @staticmethod
+    def dump(products: "list[Product] | Product"):
+        return product_schema.dump(products, many=isinstance(products, list))
+
+    def asdict(self):
+        return {
+            "name": self.name,
+            "type": self.type.value,
+            "inventory": self.inventory,
+        }

api/products/routes.py

@@ -0,0 +1,31 @@
+from typing import TYPE_CHECKING
+
+from sanic import Blueprint, json
+from sanic.exceptions import ServiceUnavailable
+
+from api.products.models import Product
+from api.services import ProductService, ProductType
+
+if TYPE_CHECKING:
+    from sanic import Request
+
+products = Blueprint("products")
+
+
+@products.route("/findAvailableProducts", methods=["GET"])
+async def find_available_products(request: "Request"):
+    page_size, p_type = Product.validate_args(request.headers.get("pageSize"), request.args.get("type"))
+
+    # NOTE: API_SPEC v4 requires expects TIMEOUT when type="other" or pageSize=20
+    if p_type == ProductType.OTHER or page_size == 20:
+        raise ServiceUnavailable("Timeout")
+
+    products = ProductService.find_products(p_type)
+    return json(Product.dump(products), status=200)
+
+
+@products.route("/products", methods=["POST"])
+async def add_product(request: "Request"):
+    data: Product = Product.load(request.json)
+    product = ProductService.create_product(data)
+    return json({"id": product["id"]}, status=201)

api/schemas.py

@@ -0,0 +1,35 @@
+import enum
+
+from marshmallow import Schema, fields, validate
+
+
+class OrderStatus(str, enum.Enum):
+    FULFILLED = "fulfilled"
+    PENDING = "pending"
+    CANCELLED = "cancelled"
+
+
+class ProductType(str, enum.Enum):
+    GADGET = "gadget"
+    FOOD = "food"
+    BOOK = "book"
+    OTHER = "other"
+
+
+class AvailableProductSchema(Schema):
+    type = fields.Enum(ProductType, required=False, by_value=True, load_default=None, allow_none=True)
+    page_size = fields.Integer(required=True, validate=validate.Range(min=0, error="pageSize must be positive"))
+
+
+class ProductSchema(Schema):
+    id = fields.Integer(required=False, strict=True)
+    name = fields.String(required=True)
+    type = fields.Enum(ProductType, required=True, by_value=True)
+    inventory = fields.Integer(required=True, strict=True)
+    description = fields.String(required=False, dump_default="")
+
+
+class OrderSchema(Schema):
+    productid = fields.Integer(required=True, strict=True)
+    count = fields.Integer(required=True, strict=True)
+    status = fields.Enum(OrderStatus, by_value=True, load_default=OrderStatus.PENDING)