Soap Request and Response#

SOAPRequest and SOAPResponse are the foundation any application built around azerothcore soap services. The request object allow you to construct a request and send it out, while the response object allow you to parse the response.

request.py
  1# -*- coding: utf-8 -*-
  2
  3"""
  4SOAP request and response.
  5"""
  6
  7import json
  8import dataclasses
  9from pathlib import Path
 10import xml.etree.ElementTree as ET
 11
 12import requests
 13
 14from .exc import SOAPResponseParseError, SOAPCommandFailedError
 15
 16
 17# ------------------------------------------------------------------------------
 18# Soap Request and Response
 19# ------------------------------------------------------------------------------
 20path_xml = Path(__file__).absolute().parent / "execute-command.xml"
 21
 22# default soap request headers
 23_SOAP_REQUEST_HEADERS = {"Content-Type": "application/xml"}
 24_SOAP_REQUEST_XML_TEMPLATE = path_xml.read_text(encoding="utf-8")
 25DEFAULT_USERNAME = "admin"
 26DEFAULT_PASSWORD = "admin"
 27DEFAULT_HOST = "localhost"
 28DEFAULT_PORT = 7878
 29
 30
 31@dataclasses.dataclass
 32class Base:
 33    """
 34    Base class for :class:`SOAPRequest` and :class:`SOAPResponse`.
 35    """
 36
 37    @classmethod
 38    def from_dict(cls, dct: dict):
 39        """
 40        Construct an object from a dict.
 41        """
 42        return cls(**dct)
 43
 44    def to_dict(self) -> dict:
 45        """
 46        Convert the object to a dict.
 47        """
 48        return {k: v for k, v in dataclasses.asdict(self).items() if v is not None}
 49
 50    @classmethod
 51    def from_json(cls, json_str: str):
 52        """
 53        Construct an object from a JSON string.
 54        """
 55        return cls.from_dict(json.loads(json_str))
 56
 57    def to_json(self) -> str:  # pragma: no cover
 58        """
 59        Convert the object to a JSON string.
 60        """
 61        return json.dumps(self.to_dict())
 62
 63
 64@dataclasses.dataclass
 65class SOAPRequest(Base):
 66    """
 67    :class:`~acore_soap_app.agent.impl.SOAPRequest` is a dataclass to represent
 68    the SOAP XML request.
 69
 70    Usage example
 71
 72    .. code-block:: python
 73
 74        # this code only works in where the worldserver is running
 75        >>> request = SOAPRequest(command=".server info")
 76        >>> response = request.send()
 77        >>> response.to_json()
 78        {
 79            "body": "<?xml version=\"1.0\" encoding=\"UTF-8\"?><SOAP-ENV:Envelope xmlns:SOAP-ENV=\"http://schemas.xmlsoap.org/soap/envelope/\" xmlns:SOAP-ENC=\"http://schemas.xmlsoap.org/soap/encoding/\" xmlns:xsi=\"http://www.w3.org/1999/XMLSchema-instance\" xmlns:xsd=\"http://www.w3.org/1999/XMLSchema\" xmlns:ns1=\"urn:AC\"><SOAP-ENV:Body><ns1:executeCommandResponse><result>AzerothCore rev. 85311fa55983 2023-03-25 22:36:05 +0000 (master branch) (Unix, RelWithDebInfo, Static)&#xD;Connected players: 0. Characters in world: 0.&#xD;Connection peak: 0.&#xD;Server uptime: 54 minute(s) 3 second(s)&#xD;Update time diff: 10ms, average: 10ms.&#xD;</result></ns1:executeCommandResponse></SOAP-ENV:Body></SOAP-ENV:Envelope>",
 80            "message": "AzerothCore rev. 85311fa55983 2023-03-25 22:36:05 +0000 (master branch) (Unix, RelWithDebInfo, Static)Connected players: 0. Characters in world: 0.Connection peak: 0.Server uptime: 54 minute(s) 3 second(s)Update time diff: 10ms, average: 10ms.",
 81            "succeeded": true
 82        }
 83
 84    :param command: the command to execute.
 85    :param username: the in game GM account username, default "admin".
 86    :param password: the in game GM account password, default "admin".
 87    :param host: wow world server host, default "localhost".
 88    :param port: wow world server SOAP port, default 7878.
 89
 90    More methods from base class:
 91
 92    - :meth:`~Base.from_dict`
 93    - :meth:`~Base.to_dict`
 94    - :meth:`~Base.from_json`
 95    - :meth:`~Base.to_json`
 96    """
 97
 98    command: str = dataclasses.field()
 99    username: str = dataclasses.field(default=DEFAULT_USERNAME)
100    password: str = dataclasses.field(default=DEFAULT_PASSWORD)
101    host: str = dataclasses.field(default=DEFAULT_HOST)
102    port: int = dataclasses.field(default=DEFAULT_PORT)
103
104    @property
105    def endpoint(self) -> str:
106        """
107        Construct the Soap service endpoint URL.
108        """
109        return f"http://{self.username}:{self.password}@{self.host}:{self.port}/"
110
111    def send(self) -> "SOAPResponse":  # pragma: no cover
112        """
113        Run soap command via HTTP request. This function "has to" be run on the
114        game server and talk to the localhost. You should NEVER open SOAP port
115        to public!
116        """
117        http_response = requests.post(
118            self.endpoint,
119            headers=_SOAP_REQUEST_HEADERS,
120            data=_SOAP_REQUEST_XML_TEMPLATE.format(command=self.command),
121        )
122        return SOAPResponse.parse(http_response.text)
123
124
125@dataclasses.dataclass
126class SOAPResponse(Base):
127    """
128    :class:`~acore_soap_app.agent.impl.SOAPResponse` is a dataclass to represent
129    the SOAP XML response.
130
131    Usage:
132
133    .. code-block:: python
134
135        >>> res = SOAPResponse.parse(
136        ...  '''
137        ...      <?xml version="1.0" encoding="UTF-8"?><SOAP-ENV:Envelope
138        ...      ...<result>Account created: test&#xD;</result>...</SOAP-ENV:Envelope>
139        ...  '''
140        ... )
141        >>> res.message
142        Account created: test
143        >>> res.succeeded
144        True
145
146    :param body: the raw SOAP XML response
147    :param message: if succeeded, it is the ``<result>...</result>`` part.
148        if failed, it is the ``<faultstring>...</faultstring>`` part
149    :param succeeded: a boolean flag to indicate whether the command is succeeded
150
151    More methods from base class:
152
153    - :meth:`~Base.from_dict`
154    - :meth:`~Base.to_dict`
155    - :meth:`~Base.from_json`
156    - :meth:`~Base.to_json`
157    """
158
159    body: str = dataclasses.field()
160    message: str = dataclasses.field()
161    succeeded: bool = dataclasses.field()
162
163    @classmethod
164    def parse(cls, body: str) -> "SOAPResponse":
165        """
166        Parse the SOAP XML response.
167        """
168        root = ET.fromstring(body)
169
170        results = list(root.iter("result"))
171        if len(results):
172            result = results[0]
173            if result.text:
174                message = result.text.strip()
175            else:
176                message = "No result"
177            return cls(
178                body=body.strip(),
179                message=message,
180                succeeded=True,
181            )
182
183        faultstrings = list(root.iter("faultstring"))
184        if len(faultstrings):
185            faultstring = faultstrings[0]
186            if faultstring.text:
187                message = faultstring.text.strip()
188            else:
189                message = "No fault string"
190            return cls(
191                body=body.strip(),
192                message=message,
193                succeeded=False,
194            )
195
196        # todo: add logic to handle SOAPCommandFailedError situation
197        raise SOAPResponseParseError(f"Cannot parse the response: {body!r}")
198
199    def print(self):  # pragma: no cover
200        """
201        Print the dataclass, ignore the raw response body.
202        """
203        print({"succeeded": self.succeeded, "message": self.message})
204
205
206def ensure_response_succeeded(
207    request: SOAPRequest,
208    response: SOAPResponse,
209    raises: bool,
210):
211    """
212    Ensure the response succeeded, otherwise raise an exception.
213    """
214    if response.succeeded:
215        return response
216    else:
217        if raises:
218            raise SOAPCommandFailedError(
219                f"request failed: {request.command!r}, "
220                f"response: {response.message!r}"
221            )