pa_api.xmlapi.types.operations.job

 1import enum
 2import logging
 3from dataclasses import dataclass
 4from datetime import datetime, time
 5from typing import Optional
 6
 7from pa_api.utils import (
 8    first,
 9)
10from pa_api.xmlapi.types.utils import (
11    mksx,
12    parse_datetime,
13    parse_time,
14    pd,
15)
16
17
18def parse_tdeq(d):
19    if "null" in d:
20        return None
21    try:
22        return parse_time(d)
23    except Exception as e:
24        logging.debug(e)
25    return parse_datetime(d)
26
27
28def parse_progress(progress):
29    try:
30        return float(progress)
31    except Exception as e:
32        logging.debug(f"{e} => Fallback to datetime parsing")
33
34    # When finished, progress becomes the date of the end
35    if parse_datetime(progress):
36        return 100.0
37    return None
38
39
40class JobResult(enum.Enum):
41    OK = "OK"
42    FAIL = "FAIL"
43
44
45@dataclass
46class Job:
47    # TODO: Use pydantic
48    tenq: datetime
49    tdeq: time
50    id: str
51    user: str
52    type: str
53    status: str
54    queued: bool
55    stoppable: bool
56    result: str
57    tfin: datetime
58    description: str
59    position_in_queue: int
60    progress: float
61    details: str
62    warnings: str
63
64    @staticmethod
65    def from_xml(xml) -> Optional["Job"]:
66        # TODO: Use correct pydantic functionalities
67        if isinstance(xml, (list, tuple)):
68            xml = first(xml)
69        if xml is None:
70            return None
71        p = mksx(xml)
72        return Job(
73            p("./tenq/text()", parser=pd),
74            p("./tdeq/text()", parser=parse_tdeq),
75            p("./id/text()"),
76            p("./user/text()"),
77            p("./type/text()"),
78            p("./status/text()"),
79            p("./queued/text()") != "NO",
80            p("./stoppable/text()") != "NO",
81            p("./result/text()"),
82            p("./tfin/text()", parser=pd),
83            p("./description/text()"),
84            p("./positionInQ/text()", parser=int),
85            p("./progress/text()", parser=parse_progress),
86            "\n".join(xml.xpath("./details/line/text()")),
87            p("./warnings/text()"),
88        )