caselawclient.types

  1from dataclasses import dataclass, field
  2from datetime import UTC, datetime
  3
  4from lxml import etree
  5
  6
  7@dataclass
  8class DocumentCategory:
  9    name: str
 10    subcategories: list["DocumentCategory"] = field(default_factory=list)
 11
 12
 13class InvalidDocumentURIException(Exception):
 14    """The document URI is not valid."""
 15
 16
 17class InvalidMarkLogicDocumentURIException(Exception):
 18    """The MarkLogic document URI is not valid."""
 19
 20
 21class MarkLogicDocumentURIString(str):
 22    def __new__(cls, content: str) -> "MarkLogicDocumentURIString":
 23        # Check that the URI begins with a slash
 24        if content[0] != "/":
 25            raise InvalidMarkLogicDocumentURIException(
 26                f'"{content}" is not a valid MarkLogic document URI; URIs must begin with a slash.'
 27            )
 28
 29        # Check that the URI ends with ".xml"
 30        if not content.endswith(".xml"):
 31            raise InvalidMarkLogicDocumentURIException(
 32                f'"{content}" is not a valid MarkLogic document URI; URIs must end with ".xml". '
 33            )
 34
 35        # If everything is good, return as usual
 36        return str.__new__(cls, content)
 37
 38    def as_document_uri(self) -> "DocumentURIString":
 39        return DocumentURIString(self.lstrip("/").rstrip(".xml"))
 40
 41
 42class DocumentURIString(str):
 43    """
 44    This class checks that the string is actually a valid Document URI on creation. It does _not_ manipulate the string.
 45    """
 46
 47    def __new__(cls, content: str) -> "DocumentURIString":
 48        # Check that the URI doesn't begin or end with a slash
 49        if content[0] == "/" or content[-1] == "/":
 50            raise InvalidDocumentURIException(
 51                f'"{content}" is not a valid document URI; URIs cannot begin or end with slashes.'
 52            )
 53
 54        # Check that the URI doesn't contain a full stop
 55        if "." in content:
 56            raise InvalidDocumentURIException(
 57                f'"{content}" is not a valid document URI; URIs cannot contain full stops.'
 58            )
 59
 60        # If everything is good, return as usual
 61        return str.__new__(cls, content)
 62
 63    def as_marklogic(self) -> MarkLogicDocumentURIString:
 64        return MarkLogicDocumentURIString(f"/{self}.xml")
 65
 66
 67class DocumentIdentifierSlug(str):
 68    pass
 69
 70
 71class DocumentIdentifierValue(str):
 72    pass
 73
 74
 75class SuccessFailureMessageTuple(tuple[bool, list[str]]):
 76    """
 77    Return whether an operation has succeeded or failed
 78    (and optionally a list of messages associated with that operation).
 79    Typically the messages will be exposed to the end-user.
 80    Use only where a failure is a routine event (such as during validation).
 81    """
 82
 83    def __new__(cls, success: bool, messages: list[str]) -> "SuccessFailureMessageTuple":
 84        return super().__new__(cls, [success, messages])
 85
 86    @property
 87    def success(self) -> bool:
 88        return self[0]
 89
 90    @property
 91    def messages(self) -> list[str]:
 92        return self[1]
 93
 94    def __repr__(self) -> str:
 95        return f"SuccessFailureMessageTuple({self.success!r}, {self.messages!r})"
 96
 97    def __bool__(self) -> bool:
 98        return self.success
 99
100    def __or__(self, other: "SuccessFailureMessageTuple") -> "SuccessFailureMessageTuple":
101        """Allows us to write combined_tuple = first_tuple | second_tuple"""
102        return SuccessFailureMessageTuple(self.success and other.success, self.messages + other.messages)
103
104
105def SuccessTuple() -> SuccessFailureMessageTuple:
106    return SuccessFailureMessageTuple(True, [])
107
108
109def FailureTuple(message: str | list[str]) -> SuccessFailureMessageTuple:
110    messages = message if isinstance(message, list) else [message]
111    return SuccessFailureMessageTuple(success=False, messages=messages)
112
113
114@dataclass
115class DocumentLock:
116    document_uri: DocumentURIString
117    owner: str
118    timestamp: datetime
119    timeout: int
120
121    NAMESPACES = {
122        "lock": "http://marklogic.com/xdmp/lock",
123    }
124
125    @classmethod
126    def from_string(cls, xml_string: str) -> "DocumentLock":
127        root = etree.fromstring(xml_string.encode("utf-8"))
128
129        return cls(
130            document_uri=MarkLogicDocumentURIString(root.xpath("/lock/document/text()")[0]).as_document_uri(),
131            owner=root.xpath(
132                "/lock/details/lock:lock/lock:active-locks/lock:active-lock/lock:owner/text()",
133                namespaces=cls.NAMESPACES,
134            )[0],
135            timestamp=datetime.fromtimestamp(
136                int(
137                    root.xpath(
138                        "/lock/details/lock:lock/lock:active-locks/lock:active-lock/lock:timestamp/text()",
139                        namespaces=cls.NAMESPACES,
140                    )[0]
141                ),
142                tz=UTC,
143            ),
144            timeout=int(
145                root.xpath(
146                    "/lock/details/lock:lock/lock:active-locks/lock:active-lock/lock:timeout/text()",
147                    namespaces=cls.NAMESPACES,
148                )[0]
149            ),
150        )