mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
import xml.etree.ElementTree as ET
|
|
from typing import List, Tuple, Union
|
|
|
|
import requests
|
|
|
|
from domain.abri.models import (
|
|
AbriRequestRejected,
|
|
JobLogged,
|
|
LogJobRequest,
|
|
LogJobResult,
|
|
)
|
|
from infrastructure.abri.config import AbriConfig
|
|
from infrastructure.abri.envelope import serialise_relay_request
|
|
from infrastructure.abri.errors import AbriResponseParseError, AbriTransportError
|
|
|
|
STD_JOB_CODE = "SCSEXT"
|
|
CLIENT_CODE = "HSG"
|
|
RESOURCE_GROUP = "Surveyors"
|
|
|
|
|
|
class AbriClient:
|
|
def __init__(self, config: AbriConfig) -> None:
|
|
self._config = config
|
|
self._session = requests.Session()
|
|
|
|
def log_job(self, request: LogJobRequest) -> LogJobResult:
|
|
outcome = self._exchange(
|
|
request_type="logjob",
|
|
parameters=[
|
|
("place_ref", request.place_ref),
|
|
("std_job_code", STD_JOB_CODE),
|
|
("client", CLIENT_CODE),
|
|
("short_description", request.short_description),
|
|
("long_description", request.long_description),
|
|
("client_ref", request.client_ref),
|
|
("appointment_date", request.appointment_date.strftime("%d/%m/%Y")),
|
|
("appointment_time", request.appointment_time),
|
|
("resource", self._config.default_resource),
|
|
("resource_group", RESOURCE_GROUP),
|
|
],
|
|
)
|
|
|
|
if isinstance(outcome, AbriRequestRejected):
|
|
return outcome
|
|
|
|
return self._parse_job_logged(outcome)
|
|
|
|
def _exchange(
|
|
self, request_type: str, parameters: List[Tuple[str, str]]
|
|
) -> Union[ET.Element, AbriRequestRejected]:
|
|
envelope = serialise_relay_request(
|
|
request_type=request_type,
|
|
parameters=parameters,
|
|
username=self._config.username,
|
|
password=self._config.password,
|
|
)
|
|
|
|
reply = self._parse_reply(self._post(envelope))
|
|
|
|
if self._is_failure_document(reply):
|
|
return self._parse_rejection(reply)
|
|
|
|
return reply
|
|
|
|
def _post(self, envelope: bytes) -> bytes:
|
|
try:
|
|
response = self._session.post(self._config.endpoint_url, data=envelope)
|
|
response.raise_for_status()
|
|
except requests.RequestException as error:
|
|
raise AbriTransportError(str(error)) from error
|
|
|
|
return response.content
|
|
|
|
@staticmethod
|
|
def _parse_reply(body: bytes) -> ET.Element:
|
|
try:
|
|
return ET.fromstring(body)
|
|
except ET.ParseError as error:
|
|
raise AbriResponseParseError(str(error)) from error
|
|
|
|
@staticmethod
|
|
def _is_failure_document(reply: ET.Element) -> bool:
|
|
return reply.tag == "response"
|
|
|
|
@staticmethod
|
|
def _parse_job_logged(root: ET.Element) -> JobLogged:
|
|
job_logged = root.find("Jobs/Job_logged")
|
|
|
|
if job_logged is None:
|
|
raise AbriResponseParseError(
|
|
"Job_logged element missing from relay response"
|
|
)
|
|
|
|
job_no = job_logged.get("job_no")
|
|
logged_info = job_logged.get("logged_info")
|
|
|
|
if job_no is None or logged_info is None:
|
|
raise AbriResponseParseError(
|
|
"Job_logged element missing job_no or logged_info"
|
|
)
|
|
|
|
return JobLogged(job_no=job_no, logged_info=logged_info)
|
|
|
|
@staticmethod
|
|
def _parse_rejection(root: ET.Element) -> AbriRequestRejected:
|
|
success = root.findtext("success")
|
|
code = root.findtext("code")
|
|
message = root.findtext("message")
|
|
|
|
if success != "false" or code is None or message is None:
|
|
raise AbriResponseParseError("malformed relay failure response")
|
|
|
|
return AbriRequestRejected(code=code, message=message)
|