# Part 3: Building SOAP Services with Python and spyne

## Why Build a SOAP Server?

Most of the time, developers interact with SOAP as a *client* — consuming an existing enterprise service. But there are situations where you need to expose your own service using SOAP:

* You are building an API that must integrate with an enterprise platform that only speaks SOAP (SAP, Oracle, legacy CRM)
* You are standing up a mock service to test your client code independently
* You are migrating an old Java or .NET SOAP service to Python and need to maintain the same contract

I built my first spyne service when I needed to provide a data feed to an insurance platform that required a specific WSDL contract. The platform could not consume REST. This section uses a similar inventory management service as the example — the kind of service you would genuinely build for a B2B integration.

## spyne Overview

[spyne](https://spyne.io/) is a Python library for building SOAP (and other protocol) services. It handles:

* WSDL generation based on your Python classes
* XML serialisation and deserialisation
* SOAP Fault generation
* Multiple transports (HTTP via WSGI, TCP, etc.)

The core concepts in spyne:

| Concept           | Purpose                                                 |
| ----------------- | ------------------------------------------------------- |
| `ServiceBase`     | Base class for your service; define `@rpc` methods here |
| `@rpc`            | Decorator that marks a method as a SOAP operation       |
| `ComplexModel`    | Base class for defining XSD complex types               |
| `Application`     | Wires the service, protocol, and transport together     |
| `WsgiApplication` | WSGI adapter for hosting over HTTP                      |

## Project Structure

```
soap_inventory/
├── models.py           # ComplexModel type definitions
├── service.py          # ServiceBase with @rpc operations
├── app.py              # WSGI application setup
├── server.py           # Flask server entrypoint
└── requirements.txt
```

## Step 1: Define Data Models

Define your XSD types as Python classes that extend `ComplexModel`.

```python
# soap_inventory/models.py
from spyne import ComplexModel, String, Decimal, Boolean, Integer, Array
from spyne.model import Unicode


class ProductInput(ComplexModel):
    """Input model for creating or updating a product."""
    class Attributes(ComplexModel.Attributes):
        # The XSD namespace for these types
        target_namespace = "http://example.com/inventory"

    ProductId = String(min_occurs=1, max_occurs=1)
    Name = String(min_occurs=1, max_occurs=1)
    Price = Decimal(min_occurs=1, max_occurs=1)
    InStock = Boolean(min_occurs=0, max_occurs=1)


class Product(ComplexModel):
    """Full product representation returned in responses."""
    class Attributes(ComplexModel.Attributes):
        target_namespace = "http://example.com/inventory"

    ProductId = String
    Name = String
    Price = Decimal
    InStock = Boolean
    StockCount = Integer


class ProductList(ComplexModel):
    """A paginated list of products."""
    class Attributes(ComplexModel.Attributes):
        target_namespace = "http://example.com/inventory"

    Products = Array(Product)
    TotalCount = Integer
    Page = Integer
    PageSize = Integer


class StockUpdateResult(ComplexModel):
    """Result of a stock count update operation."""
    class Attributes(ComplexModel.Attributes):
        target_namespace = "http://example.com/inventory"

    ProductId = String
    OldStockCount = Integer
    NewStockCount = Integer
    UpdatedAt = String  # ISO 8601 string; use DateTime for native datetime handling
```

`ComplexModel` classes map directly to XSD `complexType` definitions in the generated WSDL. When spyne generates the WSDL, it reads these class attributes and produces the corresponding XSD automatically.

## Step 2: Define the Service

The service class inherits from `ServiceBase` and defines operations using the `@rpc` decorator.

```python
# soap_inventory/service.py
from datetime import datetime, timezone
from decimal import Decimal

from spyne import ServiceBase, rpc, String, Integer, Unicode
from spyne.error import ResourceNotFoundError, ArgumentError

from .models import Product, ProductInput, ProductList, StockUpdateResult

# In-memory store for this example.
# In a real project, replace this with a database query.
_PRODUCTS: dict[str, Product] = {
    "SKU-001": Product(
        ProductId="SKU-001",
        Name="Widget Pro",
        Price=Decimal("29.99"),
        InStock=True,
        StockCount=150,
    ),
    "SKU-002": Product(
        ProductId="SKU-002",
        Name="Gadget Lite",
        Price=Decimal("14.50"),
        InStock=True,
        StockCount=42,
    ),
}


class InventoryService(ServiceBase):
    """
    SOAP service for product inventory management.
    Intended for B2B integration with partner systems.
    """

    @rpc(String, _returns=Product)
    def GetProduct(ctx, product_id: str) -> Product:
        """
        Retrieve a single product by its ID.

        Args:
            product_id: The unique product SKU identifier.

        Returns:
            Product: The full product record.

        Raises:
            ResourceNotFoundError: When no product exists with the given ID.
        """
        product = _PRODUCTS.get(product_id)
        if product is None:
            raise ResourceNotFoundError(product_id)
        return product

    @rpc(Integer, Integer, _returns=ProductList)
    def ListProducts(ctx, page: int, page_size: int) -> ProductList:
        """
        Return a paginated list of all products.

        Args:
            page: Page number (1-based).
            page_size: Number of products per page.

        Returns:
            ProductList: Paginated product list with total count.
        """
        if page_size is None or page_size <= 0:
            page_size = 10
        if page is None or page <= 0:
            page = 1

        all_products = list(_PRODUCTS.values())
        start = (page - 1) * page_size
        end = start + page_size
        page_items = all_products[start:end]

        return ProductList(
            Products=page_items,
            TotalCount=len(all_products),
            Page=page,
            PageSize=page_size,
        )

    @rpc(ProductInput, _returns=Product)
    def CreateProduct(ctx, product_input: ProductInput) -> Product:
        """
        Create a new product.

        Args:
            product_input: Product data including ID, name, and price.

        Returns:
            Product: The newly created product record.

        Raises:
            ArgumentError: When a product with the given ID already exists.
        """
        if product_input.ProductId in _PRODUCTS:
            raise ArgumentError(
                f"Product already exists: {product_input.ProductId}"
            )

        new_product = Product(
            ProductId=product_input.ProductId,
            Name=product_input.Name,
            Price=product_input.Price,
            InStock=product_input.InStock if product_input.InStock is not None else True,
            StockCount=0,
        )
        _PRODUCTS[new_product.ProductId] = new_product
        return new_product

    @rpc(String, Integer, _returns=StockUpdateResult)
    def UpdateStock(ctx, product_id: str, new_stock_count: int) -> StockUpdateResult:
        """
        Update the stock count for a product.

        Args:
            product_id: The SKU of the product to update.
            new_stock_count: The new stock quantity (must be >= 0).

        Returns:
            StockUpdateResult: Before/after stock levels and timestamp.

        Raises:
            ResourceNotFoundError: When product does not exist.
            ArgumentError: When new_stock_count is negative.
        """
        if new_stock_count < 0:
            raise ArgumentError("Stock count cannot be negative")

        product = _PRODUCTS.get(product_id)
        if product is None:
            raise ResourceNotFoundError(product_id)

        old_count = product.StockCount
        product.StockCount = new_stock_count
        product.InStock = new_stock_count > 0

        return StockUpdateResult(
            ProductId=product_id,
            OldStockCount=old_count,
            NewStockCount=new_stock_count,
            UpdatedAt=datetime.now(timezone.utc).isoformat(),
        )
```

### Understanding `@rpc`

The `@rpc` decorator is the core of spyne's operation definition.

```python
@rpc(String, Integer, _returns=StockUpdateResult)
def UpdateStock(ctx, product_id: str, new_stock_count: int) -> StockUpdateResult:
```

* Positional arguments to `@rpc` are the **input parameter types** in order — `String`, `Integer`
* `_returns` is the **return type**
* `ctx` is the spyne context (request metadata, headers, etc.) — always first, not an input parameter
* Parameter names in the function signature become the XSD element names in the WSDL

## Step 3: Wire the Application

```python
# soap_inventory/app.py
from spyne import Application
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication

from .service import InventoryService

# Build the spyne Application.
# - services: list of ServiceBase subclasses
# - tns: target namespace — becomes the WSDL targetNamespace
# - in_protocol / out_protocol: SOAP 1.1 or 1.2
application = Application(
    services=[InventoryService],
    tns="http://example.com/inventory",
    name="InventoryService",
    in_protocol=Soap11(validator="lxml"),   # Use lxml for XSD validation
    out_protocol=Soap11(),
)

wsgi_app = WsgiApplication(application)
```

Setting `validator="lxml"` on `in_protocol` enables XSD validation of incoming requests. Requests that do not match the schema return a SOAP Fault automatically without reaching your service code.

## Step 4: Serve Over HTTP with Flask

```python
# soap_inventory/server.py
from wsgiref.simple_server import make_server

from flask import Flask, request, Response

from .app import wsgi_app

flask_app = Flask(__name__)


@flask_app.route("/inventory", methods=["GET", "POST"])
def inventory_endpoint():
    # Flask does not natively speak WSGI the way spyne expects,
    # so we delegate to the spyne WSGI app for all requests.
    environ = request.environ
    response_started = []

    def start_response(status, headers, exc_info=None):
        response_started.append((status, headers))

    result = wsgi_app(environ, start_response)

    if response_started:
        status, headers = response_started[0]
        code = int(status.split(" ", 1)[0])
        resp = Response(
            response=b"".join(result),
            status=code,
            headers=dict(headers),
        )
        return resp

    return Response(b"".join(result), status=500)


if __name__ == "__main__":
    flask_app.run(host="0.0.0.0", port=8000, debug=True)
```

### Alternative: Serve with wsgiref Directly

For local development and testing, Python's built-in `wsgiref` is simpler:

```python
# serve_direct.py
from wsgiref.simple_server import make_server
from soap_inventory.app import wsgi_app

if __name__ == "__main__":
    server = make_server("0.0.0.0", 8000, wsgi_app)
    print("Serving SOAP on http://localhost:8000/")
    print("WSDL at: http://localhost:8000/?wsdl")
    server.serve_forever()
```

Start the server:

```bash
python serve_direct.py
# Serving SOAP on http://localhost:8000/
# WSDL at: http://localhost:8000/?wsdl
```

## Viewing the Generated WSDL

Once the server is running, spyne automatically exposes the WSDL at `?wsdl`:

```bash
curl http://localhost:8000/?wsdl
```

You can also load it in a browser or import it into SoapUI for visual inspection.

The WSDL generated by spyne maps directly to what you defined:

* Each `@rpc` method becomes an `operation` in the `portType`
* Each `ComplexModel` class becomes an `xsd:complexType`
* The `tns` argument to `Application` becomes the `targetNamespace`

## Calling the Service

With the server running, here is a quick smoke test using zeep:

```python
# quick_test.py
from decimal import Decimal
from zeep import Client

client = Client("http://localhost:8000/?wsdl")

# GetProduct
product = client.service.GetProduct(product_id="SKU-001")
print(f"Got: {product.Name} @ {product.Price}")

# CreateProduct
from zeep.helpers import serialize_object
new_product = client.service.CreateProduct(
    product_input={
        "ProductId": "SKU-003",
        "Name": "Super Widget",
        "Price": Decimal("49.99"),
        "InStock": True,
    }
)
print(f"Created: {new_product.ProductId}")

# UpdateStock
result = client.service.UpdateStock(
    product_id="SKU-001",
    new_stock_count=200,
)
print(f"Stock updated: {result.OldStockCount} -> {result.NewStockCount}")
```

## Logging Incoming SOAP Messages

During development, I always add logging to inspect raw requests. In spyne, you handle this with an event handler on the application.

```python
# soap_inventory/app.py (with logging)
import logging
from lxml import etree
from spyne import Application, event
from spyne.protocol.soap import Soap11
from spyne.server.wsgi import WsgiApplication

from .service import InventoryService

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("soap.inventory")


def _on_method_call(ctx):
    logger.debug(
        "Incoming SOAP call: service=%s operation=%s",
        ctx.service_class.__name__,
        ctx.function.__name__,
    )


def _on_method_return_object(ctx):
    logger.debug(
        "SOAP response: %s returned successfully",
        ctx.function.__name__,
    )


application = Application(
    services=[InventoryService],
    tns="http://example.com/inventory",
    name="InventoryService",
    in_protocol=Soap11(validator="lxml"),
    out_protocol=Soap11(),
)

application.event_manager.add_listener("method_call", _on_method_call)
application.event_manager.add_listener("method_return_object", _on_method_return_object)

wsgi_app = WsgiApplication(application)
```

## Common spyne Pitfalls

Things that caught me when I first built a spyne service:

**1. Parameter names in `@rpc` become XSD element names.** If your method has `product_id: str`, the WSDL will have `<product_id>` in the input element. This matters when you need the WSDL to match an existing contract — use the exact capitalization expected.

**2. `ctx` is not a typed parameter.** spyne automatically passes the context as the first argument. Do not include it in `@rpc(...)` type list.

**3. `Array(SomeComplexModel)` generates the correct unbounded list type.** Using a plain Python `list` will not be serialised correctly — always use `Array(T)`.

**4. The `_returns` must be set explicitly.** If missing, spyne generates a void return, and your response data will not appear in the WSDL or be serialised.

**5. Use `Decimal` not `float` for monetary values.** `Float` loses precision in XML serialisation.

## What's Next

You have built a working SOAP service with spyne:

* Data models with `ComplexModel`
* Operations with `@rpc`
* XSD validation on incoming requests
* HTTP hosting with wsgiref and Flask
* Logging with application event handlers

In [Part 4](https://blog.htunnthuthu.com/getting-started/programming/soap-101/part-4-soap-client-zeep), we switch to the client side and use zeep to consume SOAP services robustly — handling complex types, managing sessions, caching WSDLs, and debugging transport issues.