mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-19 17:03:02 +00:00
66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
import xml.etree.ElementTree as ET
|
|
|
|
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 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:
|
|
body = serialise_relay_request(
|
|
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),
|
|
],
|
|
username=self._config.username,
|
|
password=self._config.password,
|
|
)
|
|
try:
|
|
response = self._session.post(self._config.endpoint_url, data=body)
|
|
response.raise_for_status()
|
|
except requests.RequestException as error:
|
|
raise AbriTransportError(str(error)) from error
|
|
root = ET.fromstring(response.content)
|
|
if root.tag == "response":
|
|
return self._parse_rejection(root)
|
|
job_logged = root.find("Jobs/Job_logged")
|
|
if job_logged is None:
|
|
raise ValueError("Job_logged element missing from relay response")
|
|
return JobLogged(
|
|
job_no=job_logged.attrib["job_no"],
|
|
logged_info=job_logged.attrib["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 ValueError("malformed relay failure response")
|
|
return AbriRequestRejected(code=code, message=message)
|