caselawclient.Client

   1import importlib.metadata
   2import json
   3import logging
   4import os
   5import re
   6import warnings
   7from datetime import datetime, time, timedelta
   8from pathlib import Path
   9from typing import Any, Optional, Type, Union
  10from xml.etree.ElementTree import Element
  11
  12import environ
  13import requests
  14from dateutil.parser import isoparse
  15from defusedxml import ElementTree
  16from defusedxml.ElementTree import ParseError, fromstring
  17from ds_caselaw_utils.types import NeutralCitationString
  18from lxml import etree
  19from requests.auth import HTTPBasicAuth
  20from requests.structures import CaseInsensitiveDict
  21from requests_toolbelt.multipart import decoder
  22
  23from caselawclient import xquery_type_dicts as query_dicts
  24from caselawclient.identifier_resolution import IdentifierResolutions
  25from caselawclient.models.documents import (
  26    DOCUMENT_COLLECTION_URI_JUDGMENT,
  27    DOCUMENT_COLLECTION_URI_PRESS_SUMMARY,
  28    Document,
  29)
  30from caselawclient.models.documents.versions import VersionAnnotation
  31from caselawclient.models.judgments import Judgment
  32from caselawclient.models.press_summaries import PressSummary
  33from caselawclient.models.utilities import move
  34from caselawclient.search_parameters import SearchParameters
  35from caselawclient.types import DocumentIdentifierSlug, DocumentIdentifierValue, DocumentLock, DocumentURIString
  36from caselawclient.xquery_type_dicts import (
  37    CheckContentHashUniqueByUriDict,
  38    MarkLogicDocumentURIString,
  39    MarkLogicDocumentVersionURIString,
  40    MarkLogicPrivilegeURIString,
  41)
  42
  43from .content_hash import validate_content_hash
  44from .errors import (
  45    DocumentNotFoundError,
  46    GatewayTimeoutError,
  47    MarklogicAPIError,
  48    MarklogicBadRequestError,
  49    MarklogicCheckoutConflictError,
  50    MarklogicCommunicationError,
  51    MarklogicNotPermittedError,
  52    MarklogicResourceLockedError,
  53    MarklogicResourceNotCheckedOutError,
  54    MarklogicResourceNotFoundError,
  55    MarklogicResourceUnmanagedError,
  56    MarklogicUnauthorizedError,
  57    MarklogicValidationFailedError,
  58)
  59
  60env = environ.Env()
  61
  62# Requests timeouts: https://requests.readthedocs.io/en/latest/user/advanced/
  63CONNECT_TIMEOUT = float(os.environ.get("CONNECT_TIMEOUT", "3.05"))
  64READ_TIMEOUT = float(os.environ.get("READ_TIMEOUT", "10.0"))
  65
  66ROOT_DIR = os.path.dirname(os.path.realpath(__file__))
  67DEFAULT_XSL_TRANSFORM = "accessible-html.xsl"
  68
  69try:
  70    VERSION = importlib.metadata.version("ds-caselaw-marklogic-api-client")
  71except importlib.metadata.PackageNotFoundError:
  72    VERSION = "0"
  73DEFAULT_USER_AGENT = f"ds-caselaw-marklogic-api-client/{VERSION}"
  74
  75DEBUG: bool = bool(os.getenv("DEBUG", default=False))
  76
  77
  78class NoResponse(Exception):
  79    """A requests HTTPError has no response. We expect this will never happen."""
  80
  81
  82class MultipartResponseLongerThanExpected(Exception):
  83    """
  84    MarkLogic has returned a multipart response with more than one part, where only a single part was expected.
  85    """
  86
  87
  88def get_multipart_strings_from_marklogic_response(
  89    response: requests.Response,
  90) -> list[str]:
  91    """
  92    Given a HTTP response from a MarkLogic server, extract the text content from each part of the response.
  93
  94    :param response: A multipart HTTP response
  95
  96    :return: A list of the text within each part of the response
  97    """
  98    if not (response.content):
  99        return []
 100
 101    multipart_data = decoder.MultipartDecoder.from_response(response)
 102
 103    return [part.text for part in multipart_data.parts]
 104
 105
 106def get_multipart_bytes_from_marklogic_response(
 107    response: requests.Response,
 108) -> list[bytes]:
 109    if not (response.content):
 110        return []
 111
 112    multipart_data = decoder.MultipartDecoder.from_response(response)
 113
 114    return [part.content for part in multipart_data.parts]
 115
 116
 117def get_single_string_from_marklogic_response(
 118    response: requests.Response,
 119) -> str:
 120    """
 121    Given a HTTP response from a MarkLogic server, assuming the response contains a single part, extract the text
 122    content of the response.
 123
 124    :param response: A multipart HTTP response
 125
 126    :return: The text of the response
 127
 128    :raises MultipartResponseLongerThanExpected: If the response from MarkLogic has more than one part
 129    """
 130    parts = get_multipart_strings_from_marklogic_response(response)
 131    part_count = len(parts)
 132
 133    if part_count == 0:
 134        # TODO: This should strictly speaking be None, but fixing this involves refactoring a lot of other stuff which
 135        # relies on "" being falsy.
 136        return ""
 137
 138    if part_count > 1:
 139        raise MultipartResponseLongerThanExpected(
 140            f"Response returned {part_count} multipart items, expected 1",
 141        )
 142
 143    return parts[0]
 144
 145
 146def get_single_bytestring_from_marklogic_response(
 147    response: requests.Response,
 148) -> bytes:
 149    parts = get_multipart_bytes_from_marklogic_response(response)
 150    part_count = len(parts)
 151
 152    if part_count == 0:
 153        # TODO: This should strictly speaking be None, but fixing this involves refactoring a lot of other stuff which
 154        # relies on "" being falsy.
 155        return b""
 156
 157    if part_count > 1:
 158        raise MultipartResponseLongerThanExpected(
 159            f"Response returned {part_count} multipart items, expected 1",
 160        )
 161
 162    return parts[0]
 163
 164
 165class MarklogicApiClient:
 166    """
 167    The base class for interacting with a MarkLogic instance.
 168    """
 169
 170    http_error_classes: dict[int, Type[MarklogicAPIError]] = {
 171        400: MarklogicBadRequestError,
 172        401: MarklogicUnauthorizedError,
 173        403: MarklogicNotPermittedError,
 174        404: MarklogicResourceNotFoundError,
 175        504: GatewayTimeoutError,
 176    }
 177    error_code_classes: dict[str, Type[MarklogicAPIError]] = {
 178        "XDMP-DOCNOTFOUND": MarklogicResourceNotFoundError,
 179        "XDMP-LOCKCONFLICT": MarklogicResourceLockedError,
 180        "XDMP-LOCKED": MarklogicResourceLockedError,
 181        "DLS-UNMANAGED": MarklogicResourceUnmanagedError,
 182        "DLS-NOTCHECKEDOUT": MarklogicResourceNotCheckedOutError,
 183        "DLS-CHECKOUTCONFLICT": MarklogicCheckoutConflictError,
 184        "SEC-PRIVDNE": MarklogicNotPermittedError,
 185        "XDMP-VALIDATE.*": MarklogicValidationFailedError,
 186        "FCL-DOCUMENTNOTFOUND.*": DocumentNotFoundError,
 187    }
 188
 189    default_http_error_class = MarklogicCommunicationError
 190
 191    def __init__(
 192        self,
 193        host: str,
 194        username: str,
 195        password: str,
 196        use_https: bool,
 197        user_agent: str = DEFAULT_USER_AGENT,
 198    ) -> None:
 199        self.host = host
 200        self.username = username
 201        self.password = password
 202        self.base_url = f"{'https' if use_https else 'http'}://{self.host}:8011"
 203        # Apply auth / common headers to the session
 204        self.session = requests.Session()
 205        self.session.auth = HTTPBasicAuth(username, password)
 206        self.session.headers.update({"User-Agent": user_agent})
 207        self.user_agent = user_agent
 208
 209    def get_press_summaries_for_document_uri(
 210        self,
 211        uri: DocumentURIString,
 212    ) -> list[PressSummary]:
 213        """
 214        Returns a list of PressSummary objects associated with a given Document URI
 215        """
 216        vars: query_dicts.GetComponentsForDocumentDict = {
 217            "parent_uri": uri,
 218            "component": "pressSummary",
 219        }
 220        response = self._send_to_eval(vars, "get_components_for_document.xqy")
 221        uris = get_multipart_strings_from_marklogic_response(response)
 222        return [
 223            PressSummary(DocumentURIString(uri.strip("/").strip(".xml")), self) for uri in uris
 224        ]  # TODO: Migrate this strip behaviour into proper manipulation of a MarkLogicURIString
 225
 226    def get_document_by_uri(
 227        self,
 228        uri: DocumentURIString,
 229        search_query: Optional[str] = None,
 230    ) -> Document:
 231        document_type_class = self.get_document_type_from_uri(uri)
 232        return document_type_class(uri, self, search_query=search_query)
 233
 234    def get_document_type_from_uri(self, uri: DocumentURIString) -> Type[Document]:
 235        vars: query_dicts.DocumentCollectionsDict = {
 236            "uri": self._format_uri_for_marklogic(uri),
 237        }
 238        response = self._send_to_eval(vars, "document_collections.xqy")
 239        collections = get_multipart_strings_from_marklogic_response(response)
 240
 241        if DOCUMENT_COLLECTION_URI_JUDGMENT in collections:
 242            return Judgment
 243        if DOCUMENT_COLLECTION_URI_PRESS_SUMMARY in collections:
 244            return PressSummary
 245        return Document
 246
 247    def _get_error_code_class(self, error_code: str) -> Type[MarklogicAPIError]:
 248        """
 249        Get the exception type for a MarkLogic error code, or the first part of one
 250        """
 251        for regex, error in self.error_code_classes.items():
 252            if re.fullmatch(regex, error_code):
 253                return error
 254        print(f"No error code match found for {error_code}")
 255        return self.default_http_error_class
 256
 257    def _path_to_request_url(self, path: str) -> str:
 258        return f"{self.base_url}/{path.lstrip('/')}"
 259
 260    @classmethod
 261    def _get_error_code(cls, content_as_xml: Optional[str]) -> str:
 262        logging.warning(
 263            "XMLTools is deprecated and will be removed in later versions. "
 264            "Use methods from MarklogicApiClient.Client instead.",
 265        )
 266        if not content_as_xml:
 267            return "Unknown error, Marklogic returned a null or empty response"
 268        try:
 269            xml = fromstring(content_as_xml)
 270            return str(
 271                xml.find(
 272                    "message-code",
 273                    namespaces={"": "http://marklogic.com/xdmp/error"},
 274                ).text
 275            )
 276        except (ParseError, TypeError, AttributeError):
 277            return "Unknown error, Marklogic returned a null or empty response"
 278
 279    def _raise_for_status(self, response: requests.Response) -> None:
 280        try:
 281            response.raise_for_status()
 282        except requests.exceptions.HTTPError as e:
 283            if e.response is None:
 284                raise NoResponse
 285            status_code = e.response.status_code
 286            new_error_class = self.http_error_classes.get(
 287                status_code,
 288                self.default_http_error_class,
 289            )
 290            try:
 291                response_body = json.dumps(response.json(), indent=4)
 292            except requests.JSONDecodeError:
 293                response_body = response.text
 294
 295            if new_error_class == self.default_http_error_class:
 296                # Attempt to decode the error code from the response
 297
 298                error_code = self._get_error_code(response.content.decode("utf-8"))
 299
 300                new_error_class = self._get_error_code_class(error_code)
 301
 302            new_exception = new_error_class(
 303                f"{e}. Response body:\n{response_body}",
 304            )
 305            new_exception.response = response
 306            raise new_exception
 307
 308    def _format_uri_for_marklogic(
 309        self,
 310        uri: DocumentURIString,
 311    ) -> MarkLogicDocumentURIString:
 312        """
 313        MarkLogic requires a document URI that begins with a slash `/` and ends in `.xml`. This method ensures any takes
 314        a `DocumentURIString` and converts it to a MarkLogic-friendly `MarkLogicDocumentURIString`.
 315
 316        :return: A `MarkLogicDocumentURIString` at which the document at the given `DocumentURIString` can be located
 317            within MarkLogic.
 318        """
 319        return MarkLogicDocumentURIString(f"/{uri.lstrip('/').rstrip('/')}.xml")
 320
 321    def _xquery_path(self, xquery_file_name: str) -> str:
 322        return os.path.join(ROOT_DIR, "xquery", xquery_file_name)
 323
 324    def _send_to_eval(
 325        self,
 326        vars: query_dicts.MarkLogicAPIDict,
 327        xquery_file_name: str,
 328        timeout: tuple[float, float] = (CONNECT_TIMEOUT, READ_TIMEOUT),
 329    ) -> requests.Response:
 330        return self.eval(
 331            self._xquery_path(xquery_file_name),
 332            vars=json.dumps(vars),
 333            accept_header="application/xml",
 334            timeout=timeout,
 335        )
 336
 337    def _eval_and_decode(
 338        self,
 339        vars: query_dicts.MarkLogicAPIDict,
 340        xquery_file_name: str,
 341    ) -> str:
 342        response = self._send_to_eval(vars, xquery_file_name)
 343        return get_single_string_from_marklogic_response(response)
 344
 345    def _eval_as_bytes(
 346        self,
 347        vars: query_dicts.MarkLogicAPIDict,
 348        xquery_file_name: str,
 349    ) -> bytes:
 350        response = self._send_to_eval(vars, xquery_file_name)
 351        return get_single_bytestring_from_marklogic_response(response)
 352
 353    def prepare_request_kwargs(
 354        self,
 355        method: str,
 356        path: str,
 357        body: Optional[str] = None,
 358        data: Optional[dict[str, Any]] = None,
 359    ) -> dict[str, Any]:
 360        kwargs = dict(url=self._path_to_request_url(path))
 361        if data is not None:
 362            data = {k: v for k, v in data.items() if v is not None}
 363            if method == "GET":
 364                kwargs["params"] = data  # type: ignore
 365            else:
 366                kwargs["data"] = json.dumps(data)
 367        if body is not None:
 368            kwargs["data"] = body
 369        return kwargs
 370
 371    def make_request(
 372        self,
 373        method: str,
 374        path: str,
 375        headers: CaseInsensitiveDict[Union[str, Any]],
 376        body: Optional[str] = None,
 377        data: Optional[dict[str, Any]] = None,
 378    ) -> requests.Response:
 379        kwargs = self.prepare_request_kwargs(method, path, body, data)
 380        self.session.headers = headers
 381        response = self.session.request(method, **kwargs)
 382        # Raise relevant exception for an erroneous response
 383        self._raise_for_status(response)
 384        return response
 385
 386    def GET(self, path: str, headers: dict[str, Any], **data: Any) -> requests.Response:
 387        logging.warning("GET() is deprecated, use eval() or invoke()")
 388        return self.make_request("GET", path, headers, data)  # type: ignore
 389
 390    def POST(
 391        self,
 392        path: str,
 393        headers: dict[str, Any],
 394        **data: Any,
 395    ) -> requests.Response:
 396        logging.warning("POST() is deprecated, use eval() or invoke()")
 397        return self.make_request("POST", path, headers, data)  # type: ignore
 398
 399    def document_exists(self, document_uri: DocumentURIString) -> bool:
 400        uri = self._format_uri_for_marklogic(document_uri)
 401        vars: query_dicts.DocumentExistsDict = {
 402            "uri": uri,
 403        }
 404        decoded_response = self._eval_and_decode(vars, "document_exists.xqy")
 405
 406        if decoded_response == "true":
 407            return True
 408        if decoded_response == "false":
 409            return False
 410        raise RuntimeError("Marklogic response was neither true nor false")
 411
 412    def get_judgment_xml_bytestring(
 413        self,
 414        judgment_uri: DocumentURIString,
 415        version_uri: Optional[DocumentURIString] = None,
 416        show_unpublished: bool = False,
 417        search_query: Optional[str] = None,
 418    ) -> bytes:
 419        marklogic_document_uri = self._format_uri_for_marklogic(judgment_uri)
 420        marklogic_document_version_uri = (
 421            MarkLogicDocumentVersionURIString(
 422                self._format_uri_for_marklogic(version_uri),
 423            )
 424            if version_uri
 425            else None
 426        )
 427        show_unpublished = self.verify_show_unpublished(show_unpublished)
 428
 429        vars: query_dicts.GetJudgmentDict = {
 430            "uri": marklogic_document_uri,
 431            "version_uri": marklogic_document_version_uri,
 432            "show_unpublished": show_unpublished,
 433            "search_query": search_query,
 434        }
 435
 436        response = self._eval_as_bytes(vars, "get_judgment.xqy")
 437        if not response:
 438            raise MarklogicNotPermittedError(
 439                "The document is not published and show_unpublished was not set",
 440            )
 441
 442        return response
 443
 444    def get_judgment_xml(
 445        self,
 446        judgment_uri: DocumentURIString,
 447        version_uri: Optional[DocumentURIString] = None,
 448        show_unpublished: bool = False,
 449        search_query: Optional[str] = None,
 450    ) -> str:
 451        return self.get_judgment_xml_bytestring(
 452            judgment_uri,
 453            version_uri,
 454            show_unpublished,
 455            search_query=search_query,
 456        ).decode(encoding="utf-8")
 457
 458    def set_document_name(
 459        self,
 460        document_uri: DocumentURIString,
 461        content: str,
 462    ) -> requests.Response:
 463        uri = self._format_uri_for_marklogic(document_uri)
 464        vars: query_dicts.SetMetadataNameDict = {"uri": uri, "content": content}
 465        return self._send_to_eval(vars, "set_metadata_name.xqy")
 466
 467    def set_judgment_date(
 468        self,
 469        judgment_uri: DocumentURIString,
 470        content: str,
 471    ) -> requests.Response:
 472        warnings.warn(
 473            "set_judgment_date() is deprecated, use set_document_work_expression_date()",
 474            DeprecationWarning,
 475            stacklevel=2,
 476        )
 477        return self.set_document_work_expression_date(judgment_uri, content)
 478
 479    def set_document_work_expression_date(
 480        self,
 481        document_uri: DocumentURIString,
 482        content: str,
 483    ) -> requests.Response:
 484        uri = self._format_uri_for_marklogic(document_uri)
 485        vars: query_dicts.SetMetadataWorkExpressionDateDict = {
 486            "uri": uri,
 487            "content": content,
 488        }
 489
 490        return self._send_to_eval(vars, "set_metadata_work_expression_date.xqy")
 491
 492    def set_judgment_citation(
 493        self,
 494        judgment_uri: DocumentURIString,
 495        content: str,
 496    ) -> requests.Response:
 497        uri = self._format_uri_for_marklogic(judgment_uri)
 498        vars: query_dicts.SetMetadataCitationDict = {
 499            "uri": uri,
 500            "content": content.strip(),
 501        }
 502
 503        return self._send_to_eval(vars, "set_metadata_citation.xqy")
 504
 505    def set_document_court(
 506        self,
 507        document_uri: DocumentURIString,
 508        content: str,
 509    ) -> requests.Response:
 510        uri = self._format_uri_for_marklogic(document_uri)
 511        vars: query_dicts.SetMetadataCourtDict = {"uri": uri, "content": content}
 512
 513        return self._send_to_eval(vars, "set_metadata_court.xqy")
 514
 515    def set_document_jurisdiction(
 516        self,
 517        document_uri: DocumentURIString,
 518        content: str,
 519    ) -> requests.Response:
 520        uri = self._format_uri_for_marklogic(document_uri)
 521        vars: query_dicts.SetMetadataJurisdictionDict = {"uri": uri, "content": content}
 522        return self._send_to_eval(vars, "set_metadata_jurisdiction.xqy")
 523
 524    def set_document_court_and_jurisdiction(
 525        self,
 526        document_uri: DocumentURIString,
 527        content: str,
 528    ) -> requests.Response:
 529        if "/" in content:
 530            court, jurisdiction = re.split("\\s*/\\s*", content)
 531            self.set_document_court(document_uri, court)
 532            return self.set_document_jurisdiction(document_uri, jurisdiction)
 533        self.set_document_court(document_uri, content)
 534        return self.set_document_jurisdiction(document_uri, "")
 535
 536    def set_judgment_this_uri(
 537        self,
 538        judgment_uri: DocumentURIString,
 539    ) -> requests.Response:
 540        uri = self._format_uri_for_marklogic(judgment_uri)
 541        content_with_id = f"https://caselaw.nationalarchives.gov.uk/id/{judgment_uri.lstrip('/')}"
 542        content_without_id = f"https://caselaw.nationalarchives.gov.uk/{judgment_uri.lstrip('/')}"
 543        content_with_xml = f"https://caselaw.nationalarchives.gov.uk/{judgment_uri.lstrip('/')}/data.xml"
 544        vars: query_dicts.SetMetadataThisUriDict = {
 545            "uri": uri,
 546            "content_with_id": content_with_id,
 547            "content_without_id": content_without_id,
 548            "content_with_xml": content_with_xml,
 549        }
 550
 551        return self._send_to_eval(vars, "set_metadata_this_uri.xqy")
 552
 553    def save_locked_judgment_xml(
 554        self,
 555        judgment_uri: DocumentURIString,
 556        judgment_xml: bytes,
 557        annotation: VersionAnnotation,
 558    ) -> requests.Response:
 559        """assumes the judgment is already locked, does not unlock/check in
 560        note this version assumes the XML is raw bytes, rather than a tree..."""
 561
 562        validate_content_hash(judgment_xml)
 563        uri = self._format_uri_for_marklogic(judgment_uri)
 564
 565        annotation.set_calling_function("save_locked_judgment_xml")
 566        annotation.set_calling_agent(self.user_agent)
 567
 568        vars: query_dicts.UpdateLockedJudgmentDict = {
 569            "uri": uri,
 570            "judgment": judgment_xml.decode("utf-8"),
 571            "annotation": annotation.as_json,
 572        }
 573
 574        return self._send_to_eval(vars, "update_locked_judgment.xqy")
 575
 576    def insert_document_xml(
 577        self,
 578        document_uri: DocumentURIString,
 579        document_xml: Element,
 580        document_type: type[Document],
 581        annotation: VersionAnnotation,
 582    ) -> requests.Response:
 583        """
 584        Insert a new XML document into MarkLogic.
 585
 586        :param document_uri: The URI to insert the document at
 587        :param document_xml: The XML of the document to insert
 588        :param document_type: The type class of the document
 589        :param annotation: Annotations to record alongside this version
 590
 591        :return: The response object from MarkLogic
 592        """
 593        xml = ElementTree.tostring(document_xml)
 594
 595        uri = self._format_uri_for_marklogic(document_uri)
 596
 597        annotation.set_calling_function("insert_document_xml")
 598        annotation.set_calling_agent(self.user_agent)
 599
 600        vars: query_dicts.InsertDocumentDict = {
 601            "uri": uri,
 602            "type_collection": document_type.type_collection_name,
 603            "document": xml.decode("utf-8"),
 604            "annotation": annotation.as_json,
 605        }
 606
 607        return self._send_to_eval(vars, "insert_document.xqy")
 608
 609    def update_document_xml(
 610        self,
 611        document_uri: DocumentURIString,
 612        document_xml: Element,
 613        annotation: VersionAnnotation,
 614    ) -> requests.Response:
 615        """
 616        Updates an existing XML document in MarkLogic with a new version.
 617
 618        This uses `dls:document-checkout-update-checkin` to perform this in a single operation.
 619
 620        :param document_uri: The URI of the document to update
 621        :param document_xml: The new XML content of the document
 622        :param annotation: Annotations to record alongside this version
 623
 624        :return: The response object from MarkLogic
 625        """
 626        xml = ElementTree.tostring(document_xml)
 627
 628        uri = self._format_uri_for_marklogic(document_uri)
 629
 630        annotation.set_calling_function("update_document_xml")
 631        annotation.set_calling_agent(self.user_agent)
 632
 633        vars: query_dicts.UpdateDocumentDict = {
 634            "uri": uri,
 635            "judgment": xml.decode("utf-8"),
 636            "annotation": annotation.as_json,
 637        }
 638
 639        return self._send_to_eval(vars, "update_document.xqy")
 640
 641    def list_judgment_versions(
 642        self,
 643        judgment_uri: DocumentURIString,
 644    ) -> requests.Response:
 645        uri = self._format_uri_for_marklogic(judgment_uri)
 646        vars: query_dicts.ListJudgmentVersionsDict = {"uri": uri}
 647
 648        return self._send_to_eval(vars, "list_judgment_versions.xqy")
 649
 650    def checkout_judgment(
 651        self,
 652        judgment_uri: DocumentURIString,
 653        annotation: str = "",
 654        expires_at_midnight: bool = False,
 655        timeout_seconds: int = -1,
 656    ) -> requests.Response:
 657        """If timeout_seconds is -1, the lock never times out"""
 658        uri = self._format_uri_for_marklogic(judgment_uri)
 659        vars: query_dicts.CheckoutJudgmentDict = {
 660            "uri": uri,
 661            "annotation": annotation,
 662            "timeout": timeout_seconds,
 663        }
 664
 665        if expires_at_midnight:
 666            timeout = self.calculate_seconds_until_midnight()
 667            vars["timeout"] = timeout
 668
 669        return self._send_to_eval(vars, "checkout_judgment.xqy")
 670
 671    def checkin_judgment(self, judgment_uri: DocumentURIString) -> requests.Response:
 672        uri = self._format_uri_for_marklogic(judgment_uri)
 673        vars: query_dicts.CheckinJudgmentDict = {"uri": uri}
 674
 675        return self._send_to_eval(vars, "checkin_judgment.xqy")
 676
 677    def get_judgment_checkout_status(
 678        self,
 679        judgment_uri: DocumentURIString,
 680    ) -> requests.Response:
 681        uri = self._format_uri_for_marklogic(judgment_uri)
 682        vars: query_dicts.GetJudgmentCheckoutStatusDict = {"uri": uri}
 683
 684        return self._send_to_eval(vars, "get_judgment_checkout_status.xqy")
 685
 686    def get_judgment_checkout_status_message(
 687        self,
 688        judgment_uri: DocumentURIString,
 689    ) -> Optional[str]:
 690        """Return the annotation of the lock or `None` if there is no lock."""
 691        response = self.get_judgment_checkout_status(judgment_uri)
 692        if not response.content:
 693            return None
 694        content = decoder.MultipartDecoder.from_response(response).parts[0].text
 695        if content == "":
 696            return None
 697        response_xml = ElementTree.fromstring(content)
 698        return str(
 699            response_xml.find(
 700                "dls:annotation",
 701                namespaces={"dls": "http://marklogic.com/xdmp/dls"},
 702            ).text
 703        )
 704
 705    def get_judgment_version(
 706        self,
 707        judgment_uri: DocumentURIString,
 708        version: int,
 709    ) -> requests.Response:
 710        uri = self._format_uri_for_marklogic(judgment_uri)
 711        vars: query_dicts.GetJudgmentVersionDict = {"uri": uri, "version": str(version)}
 712
 713        return self._send_to_eval(vars, "get_judgment_version.xqy")
 714
 715    def validate_document(self, document_uri: DocumentURIString) -> bool:
 716        vars: query_dicts.ValidateDocumentDict = {
 717            "uri": self._format_uri_for_marklogic(document_uri),
 718        }
 719        response = self._send_to_eval(vars, "validate_document.xqy")
 720        content = decoder.MultipartDecoder.from_response(response).parts[0].text
 721        xml = ElementTree.fromstring(content)
 722        return (
 723            len(
 724                xml.findall(
 725                    ".//error:error",
 726                    {"error": "http://marklogic.com/xdmp/error"},
 727                ),
 728            )
 729            == 0
 730        )
 731
 732    def has_unique_content_hash(self, judgment_uri: DocumentURIString) -> bool:
 733        """
 734        Returns True if the content hash for this document is unique (not shared with other documents).
 735        """
 736        uri = self._format_uri_for_marklogic(judgment_uri)
 737        vars: CheckContentHashUniqueByUriDict = {"uri": uri}
 738        return self._eval_and_decode(vars, "check_content_hash_unique_by_uri.xqy") == "true"
 739
 740    def eval(
 741        self,
 742        xquery_path: str,
 743        vars: str,
 744        accept_header: str = "multipart/mixed",
 745        timeout: tuple[float, float] = (CONNECT_TIMEOUT, READ_TIMEOUT),
 746    ) -> requests.Response:
 747        headers = {
 748            "Content-type": "application/x-www-form-urlencoded",
 749            "Accept": accept_header,
 750        }
 751        data = {
 752            "xquery": Path(xquery_path).read_text(),
 753            "vars": vars,
 754        }
 755        path = "LATEST/eval"
 756
 757        if DEBUG:
 758            print(f"Sending {vars} to {xquery_path}")
 759
 760        response = self.session.request(
 761            "POST",
 762            url=self._path_to_request_url(path),
 763            headers=headers,
 764            data=data,
 765            timeout=timeout,
 766        )
 767        # Raise relevant exception for an erroneous response
 768        self._raise_for_status(response)
 769        return response
 770
 771    def invoke(
 772        self,
 773        module: str,
 774        vars: str,
 775        accept_header: str = "multipart/mixed",
 776    ) -> requests.Response:
 777        headers = {
 778            "Content-type": "application/x-www-form-urlencoded",
 779            "Accept": accept_header,
 780        }
 781        data = {
 782            "module": module,
 783            "vars": vars,
 784        }
 785        path = "LATEST/invoke"
 786        response = self.session.request(
 787            "POST",
 788            url=self._path_to_request_url(path),
 789            headers=headers,
 790            data=data,
 791        )
 792        # Raise relevant exception for an erroneous response
 793        self._raise_for_status(response)
 794        return response
 795
 796    def advanced_search(self, search_parameters: SearchParameters) -> requests.Response:
 797        """
 798        Performs a search on the entire document set.
 799
 800        :param query:
 801        :param court:
 802        :param judge:
 803        :param party:
 804        :param neutral_citation:
 805        :param document_name:
 806        :param consignment_number:
 807        :param specific_keyword:
 808        :param order:
 809        :param date_from:
 810        :param date_to:
 811        :param page:
 812        :param page_size:
 813        :param show_unpublished: If True, both published and unpublished documents will be returned
 814        :param only_unpublished: If True, will only return published documents. Ignores the value of show_unpublished
 815        :param collections:
 816        :return:
 817        """
 818        module = "/judgments/search/search-v2.xqy"  # as stored on Marklogic
 819        search_parameters.show_unpublished = self.verify_show_unpublished(
 820            search_parameters.show_unpublished,
 821        )
 822        vars = json.dumps(search_parameters.as_marklogic_payload())
 823        return self.invoke(module, vars)
 824
 825    def eval_xslt(
 826        self,
 827        judgment_uri: DocumentURIString,
 828        version_uri: Optional[DocumentURIString] = None,
 829        show_unpublished: bool = False,
 830        xsl_filename: str = DEFAULT_XSL_TRANSFORM,
 831        query: Optional[str] = None,
 832    ) -> requests.Response:
 833        marklogic_document_uri = self._format_uri_for_marklogic(judgment_uri)
 834        marklogic_document_version_uri = (
 835            MarkLogicDocumentVersionURIString(
 836                self._format_uri_for_marklogic(version_uri),
 837            )
 838            if version_uri
 839            else None
 840        )
 841
 842        image_location = os.getenv("XSLT_IMAGE_LOCATION", "")
 843
 844        show_unpublished = self.verify_show_unpublished(show_unpublished)
 845
 846        vars: query_dicts.XsltTransformDict = {
 847            "uri": marklogic_document_uri,
 848            "version_uri": marklogic_document_version_uri,
 849            "show_unpublished": show_unpublished,
 850            "img_location": image_location,
 851            "xsl_filename": xsl_filename,
 852            "query": query,
 853        }
 854
 855        return self._send_to_eval(vars, "xslt_transform.xqy")
 856
 857    def accessible_judgment_transformation(
 858        self,
 859        judgment_uri: DocumentURIString,
 860        version_uri: Optional[DocumentURIString] = None,
 861        show_unpublished: bool = False,
 862    ) -> requests.Response:
 863        return self.eval_xslt(
 864            judgment_uri,
 865            version_uri,
 866            show_unpublished,
 867            xsl_filename=DEFAULT_XSL_TRANSFORM,
 868        )
 869
 870    def original_judgment_transformation(
 871        self,
 872        judgment_uri: DocumentURIString,
 873        version_uri: Optional[DocumentURIString] = None,
 874        show_unpublished: bool = False,
 875    ) -> requests.Response:
 876        return self.eval_xslt(
 877            judgment_uri,
 878            version_uri,
 879            show_unpublished,
 880            xsl_filename="as-handed-down.xsl",
 881        )
 882
 883    def get_property(self, judgment_uri: DocumentURIString, name: str) -> str:
 884        uri = self._format_uri_for_marklogic(judgment_uri)
 885        vars: query_dicts.GetPropertyDict = {
 886            "uri": uri,
 887            "name": name,
 888        }
 889        return self._eval_and_decode(vars, "get_property.xqy")
 890
 891    def get_property_as_node(self, judgment_uri: DocumentURIString, name: str) -> Optional[etree._Element]:
 892        uri = self._format_uri_for_marklogic(judgment_uri)
 893        vars: query_dicts.GetPropertyAsNodeDict = {
 894            "uri": uri,
 895            "name": name,
 896        }
 897        value = self._eval_and_decode(vars, "get_property_as_node.xqy")
 898        if not value:
 899            return None
 900        return etree.fromstring(value)
 901
 902    def get_version_annotation(self, judgment_uri: DocumentURIString) -> str:
 903        uri = self._format_uri_for_marklogic(judgment_uri)
 904        vars: query_dicts.GetVersionAnnotationDict = {
 905            "uri": uri,
 906        }
 907        return self._eval_and_decode(vars, "get_version_annotation.xqy")
 908
 909    def get_version_created_datetime(self, judgment_uri: DocumentURIString) -> datetime:
 910        uri = self._format_uri_for_marklogic(judgment_uri)
 911        vars: query_dicts.GetVersionCreatedDict = {
 912            "uri": uri,
 913        }
 914        return datetime.strptime(
 915            self._eval_and_decode(vars, "get_version_created.xqy"),
 916            "%Y-%m-%dT%H:%M:%S.%f%z",
 917        )
 918
 919    def set_property(
 920        self,
 921        judgment_uri: DocumentURIString,
 922        name: str,
 923        value: str,
 924    ) -> requests.Response:
 925        uri = self._format_uri_for_marklogic(judgment_uri)
 926        vars: query_dicts.SetPropertyDict = {
 927            "uri": uri,
 928            "value": value,
 929            "name": name,
 930        }
 931
 932        return self._send_to_eval(vars, "set_property.xqy")
 933
 934    def set_property_as_node(
 935        self,
 936        judgment_uri: DocumentURIString,
 937        name: str,
 938        value: etree._Element,
 939    ) -> requests.Response:
 940        """Given a root node, set the value of the MarkLogic property for a document to the _contents_ of that root node. The root node itself is discarded."""
 941        uri = self._format_uri_for_marklogic(judgment_uri)
 942        vars: query_dicts.SetPropertyAsNodeDict = {
 943            "uri": uri,
 944            "value": etree.tostring(value).decode(),
 945            "name": name,
 946        }
 947
 948        return self._send_to_eval(vars, "set_property_as_node.xqy")
 949
 950    def set_boolean_property(
 951        self,
 952        judgment_uri: DocumentURIString,
 953        name: str,
 954        value: bool,
 955    ) -> requests.Response:
 956        uri = self._format_uri_for_marklogic(judgment_uri)
 957        string_value = "true" if value else "false"
 958        vars: query_dicts.SetBooleanPropertyDict = {
 959            "uri": uri,
 960            "value": string_value,
 961            "name": name,
 962        }
 963        """
 964        Set a property within MarkLogic which is specifically a boolean.
 965
 966        Since XML has no concept of boolean, the actual value in the database is set to `"true"` or `"false"`.
 967        """
 968        return self._send_to_eval(vars, "set_boolean_property.xqy")
 969
 970    def get_boolean_property(self, judgment_uri: DocumentURIString, name: str) -> bool:
 971        """
 972        Get a property from MarkLogic which is specifically a boolean.
 973
 974        :return: `True` if the property exists and has a value of `"true"`, otherwise `False`
 975        """
 976        content = self.get_property(judgment_uri, name)
 977        return content == "true"
 978
 979    def set_datetime_property(
 980        self,
 981        judgment_uri: DocumentURIString,
 982        name: str,
 983        value: datetime,
 984    ) -> requests.Response:
 985        """Set a property within MarkLogic which is specifically a datetime."""
 986        uri = self._format_uri_for_marklogic(judgment_uri)
 987        vars: query_dicts.SetDatetimePropertyDict = {
 988            "uri": uri,
 989            "value": value.isoformat(),
 990            "name": name,
 991        }
 992        return self._send_to_eval(vars, "set_datetime_property.xqy")
 993
 994    def get_datetime_property(self, judgment_uri: DocumentURIString, name: str) -> Optional[datetime]:
 995        """
 996        Get a property from MarkLogic which is specifically a datetime.
 997
 998        :return: A datetime with the value of the property, or `None` if it does not exist
 999        """
1000        content = self.get_property(judgment_uri, name)
1001
1002        if content:
1003            return isoparse(content)
1004
1005        return None
1006
1007    def set_published(
1008        self,
1009        judgment_uri: DocumentURIString,
1010        published: bool,
1011    ) -> requests.Response:
1012        return self.set_boolean_property(judgment_uri, "published", published)
1013
1014    def get_published(self, judgment_uri: DocumentURIString) -> bool:
1015        return self.get_boolean_property(judgment_uri, "published")
1016
1017    def get_last_modified(self, judgment_uri: DocumentURIString) -> str:
1018        uri = self._format_uri_for_marklogic(judgment_uri)
1019        vars: query_dicts.GetLastModifiedDict = {
1020            "uri": uri,
1021        }
1022
1023        response = self._send_to_eval(vars, "get_last_modified.xqy")
1024
1025        if not response.text:
1026            return ""
1027
1028        content = str(decoder.MultipartDecoder.from_response(response).parts[0].text)
1029        return content
1030
1031    def delete_judgment(self, judgment_uri: DocumentURIString) -> requests.Response:
1032        uri = self._format_uri_for_marklogic(judgment_uri)
1033        vars: query_dicts.DeleteJudgmentDict = {"uri": uri}
1034        return self._send_to_eval(vars, "delete_judgment.xqy")
1035
1036    def copy_document(
1037        self,
1038        old: DocumentURIString,
1039        new: DocumentURIString,
1040    ) -> requests.Response:
1041        old_uri = self._format_uri_for_marklogic(old)
1042        new_uri = self._format_uri_for_marklogic(new)
1043
1044        vars: query_dicts.CopyDocumentDict = {
1045            "old_uri": old_uri,
1046            "new_uri": new_uri,
1047        }
1048        return self._send_to_eval(vars, "copy_document.xqy")
1049
1050    def break_checkout(self, judgment_uri: DocumentURIString) -> requests.Response:
1051        uri = self._format_uri_for_marklogic(judgment_uri)
1052        vars: query_dicts.BreakJudgmentCheckoutDict = {
1053            "uri": uri,
1054        }
1055        return self._send_to_eval(vars, "break_judgment_checkout.xqy")
1056
1057    def user_has_privilege(
1058        self,
1059        username: str,
1060        privilege_uri: MarkLogicPrivilegeURIString,
1061        privilege_action: str,
1062    ) -> requests.Response:
1063        vars: query_dicts.UserHasPrivilegeDict = {
1064            "user": username,
1065            "privilege_uri": privilege_uri,
1066            "privilege_action": privilege_action,
1067        }
1068        return self._send_to_eval(vars, "user_has_privilege.xqy")
1069
1070    def user_can_view_unpublished_judgments(self, username: str) -> bool:
1071        if self.user_has_admin_role(username):
1072            return True
1073
1074        check_privilege = self.user_has_privilege(
1075            username,
1076            MarkLogicPrivilegeURIString(
1077                "https://caselaw.nationalarchives.gov.uk/custom/privileges/can-view-unpublished-documents",
1078            ),
1079            "execute",
1080        )
1081        return get_single_string_from_marklogic_response(check_privilege).lower() == "true"
1082
1083    def user_has_role(self, username: str, role: str) -> requests.Response:
1084        vars: query_dicts.UserHasRoleDict = {
1085            "user": username,
1086            "role": role,
1087        }
1088        return self._send_to_eval(vars, "user_has_role.xqy")
1089
1090    def user_has_admin_role(self, username: str) -> bool:
1091        check_role = self.user_has_role(
1092            username,
1093            "admin",
1094        )
1095        multipart_data = decoder.MultipartDecoder.from_response(check_role)
1096        result = str(multipart_data.parts[0].text)
1097        return result.lower() == "true"
1098
1099    def calculate_seconds_until_midnight(self, now: Optional[datetime] = None) -> int:
1100        """
1101        Get timedelta until end of day on the datetime passed, or current time.
1102        https://stackoverflow.com/questions/45986035/seconds-until-end-of-day-in-python
1103        """
1104        if not now:
1105            now = datetime.now()
1106        tomorrow = now + timedelta(days=1)
1107        difference = datetime.combine(tomorrow, time.min) - now
1108
1109        return difference.seconds
1110
1111    def verify_show_unpublished(self, show_unpublished: bool) -> bool:
1112        if show_unpublished and not self.user_can_view_unpublished_judgments(
1113            self.username,
1114        ):
1115            return False
1116        return show_unpublished
1117
1118    def get_properties_for_search_results(
1119        self,
1120        judgment_uris: list[DocumentURIString],
1121    ) -> str:
1122        uris = [self._format_uri_for_marklogic(judgment_uri) for judgment_uri in judgment_uris]
1123        vars: query_dicts.GetPropertiesForSearchResultsDict = {"uris": uris}
1124        response = self._send_to_eval(vars, "get_properties_for_search_results.xqy")
1125        return get_single_string_from_marklogic_response(response)
1126
1127    def search_and_decode_response(self, search_parameters: SearchParameters) -> bytes:
1128        response = self.advanced_search(search_parameters)
1129        return get_single_bytestring_from_marklogic_response(response)
1130
1131    def search_judgments_and_decode_response(
1132        self,
1133        search_parameters: SearchParameters,
1134    ) -> bytes:
1135        search_parameters.collections = [DOCUMENT_COLLECTION_URI_JUDGMENT]
1136        return self.search_and_decode_response(search_parameters)
1137
1138    def update_document_uri(self, old_uri: DocumentURIString, new_citation: NeutralCitationString) -> DocumentURIString:
1139        """
1140        Move the document at old_uri to the correct location based on the neutral citation
1141        The new neutral citation *must* not already exist (that is handled elsewhere)
1142        This might not be needed; changing the URI/neutral citation is vanishingly rare
1143        """
1144        return move.update_document_uri(old_uri, new_citation, api_client=self)
1145
1146    def get_combined_stats_table(self) -> list[list[Any]]:
1147        """Run the combined statistics table xquery and return the result as a list of lists, each representing a table
1148        row."""
1149        results: list[list[Any]] = json.loads(
1150            get_single_string_from_marklogic_response(
1151                self._send_to_eval({}, "get_combined_stats_table.xqy"),
1152            ),
1153        )
1154
1155        return results
1156
1157    def get_highest_enrichment_version(self) -> tuple[int, int]:
1158        """This gets the highest enrichment version in the database,
1159        so if nothing has been enriched with the most recent version of enrichment,
1160        this won't reflect that change."""
1161        table = json.loads(
1162            get_single_string_from_marklogic_response(
1163                self._send_to_eval(
1164                    {},
1165                    "get_highest_enrichment_version.xqy",
1166                ),
1167            ),
1168        )
1169
1170        return (int(table[1][1]), int(table[1][2]))
1171
1172    def get_pending_enrichment_for_version(
1173        self,
1174        target_enrichment_version: tuple[int, int],
1175        target_parser_version: tuple[int, int],
1176        maximum_records: int = 1000,
1177    ) -> list[list[Any]]:
1178        """Retrieve documents which are not yet enriched with a given version."""
1179        vars: query_dicts.GetPendingEnrichmentForVersionDict = {
1180            "target_enrichment_major_version": target_enrichment_version[0],
1181            "target_enrichment_minor_version": target_enrichment_version[1],
1182            "target_parser_major_version": target_parser_version[0],
1183            "target_parser_minor_version": target_parser_version[1],
1184            "maximum_records": maximum_records,
1185        }
1186        results: list[list[Any]] = json.loads(
1187            get_single_string_from_marklogic_response(
1188                self._send_to_eval(
1189                    vars,
1190                    "get_pending_enrichment_for_version.xqy",
1191                ),
1192            ),
1193        )
1194
1195        return results
1196
1197    def get_recently_enriched(
1198        self,
1199    ) -> list[list[Any]]:
1200        """Retrieve documents which are not yet enriched with a given version."""
1201        results: list[list[Any]] = json.loads(
1202            get_single_string_from_marklogic_response(
1203                self._send_to_eval(
1204                    {},
1205                    "get_recently_enriched.xqy",
1206                ),
1207            ),
1208        )
1209
1210        return results
1211
1212    def get_highest_parser_version(self) -> tuple[int, int]:
1213        """This gets the highest parser version in the database, so if nothing has been parsed with the most recent version of the parser, this won't reflect that change."""
1214        table = json.loads(
1215            get_single_string_from_marklogic_response(
1216                self._send_to_eval(
1217                    {},
1218                    "get_highest_parser_version.xqy",
1219                ),
1220            ),
1221        )
1222
1223        return (int(table[1][1]), int(table[1][2]))
1224
1225    def get_documents_pending_parse_for_version(
1226        self,
1227        target_version: tuple[int, int],
1228        maximum_records: int = 1000,
1229    ) -> list[list[Any]]:
1230        """Retrieve a list of documents which are not yet parsed with a given version."""
1231        vars: query_dicts.GetPendingParseForVersionDocumentsDict = {
1232            "target_major_version": target_version[0],
1233            "target_minor_version": target_version[1],
1234            "maximum_records": maximum_records,
1235        }
1236        results: list[list[Any]] = json.loads(
1237            get_single_string_from_marklogic_response(
1238                self._send_to_eval(
1239                    vars,
1240                    "get_pending_parse_for_version_documents.xqy",
1241                ),
1242            ),
1243        )
1244
1245        return results
1246
1247    def get_count_pending_parse_for_version(
1248        self,
1249        target_version: tuple[int, int],
1250    ) -> int:
1251        """Get the total number of documents which are not yet parsed with a given version."""
1252        vars: query_dicts.GetPendingParseForVersionCountDict = {
1253            "target_major_version": target_version[0],
1254            "target_minor_version": target_version[1],
1255        }
1256        results = json.loads(
1257            get_single_string_from_marklogic_response(
1258                self._send_to_eval(
1259                    vars,
1260                    "get_pending_parse_for_version_count.xqy",
1261                ),
1262            ),
1263        )
1264
1265        return int(results[1][0])
1266
1267    def get_recently_parsed(
1268        self,
1269    ) -> list[list[Any]]:
1270        """Retrieve documents which are not yet enriched with a given version."""
1271        results: list[list[Any]] = json.loads(
1272            get_single_string_from_marklogic_response(
1273                self._send_to_eval(
1274                    {},
1275                    "get_recently_parsed.xqy",
1276                ),
1277            ),
1278        )
1279
1280        return results
1281
1282    def get_locked_documents(
1283        self,
1284    ) -> list[DocumentLock]:
1285        """Retrieve all currently locked documents."""
1286        results = [
1287            DocumentLock.from_string(lock)
1288            for lock in get_multipart_strings_from_marklogic_response(
1289                self._send_to_eval({}, "get_locked_documents.xqy")
1290            )
1291        ]
1292
1293        return sorted(results, key=lambda lock: lock.timestamp)
1294
1295    def get_missing_fclid(
1296        self,
1297        maximum_records: int = 50,
1298    ) -> list[str]:
1299        """Retrieve the URIs of published documents which do not have an identifier in the `fclid` schema."""
1300        vars: query_dicts.GetMissingFclidDict = {
1301            "maximum_records": maximum_records,
1302        }
1303
1304        results: list[str] = get_multipart_strings_from_marklogic_response(
1305            self._send_to_eval(
1306                vars,
1307                "get_missing_fclid.xqy",
1308            )
1309        )
1310
1311        return results
1312
1313    def resolve_from_identifier_slug(
1314        self, identifier_slug: DocumentIdentifierSlug, published_only: bool = True
1315    ) -> IdentifierResolutions:
1316        """Given a PUI/EUI url, look up the precomputed slug and return the
1317        MarkLogic document URIs which match that slug. Multiple returns should be anticipated"""
1318        vars: query_dicts.ResolveFromIdentifierSlugDict = {
1319            "identifier_slug": identifier_slug,
1320            "published_only": int(published_only),
1321        }
1322        raw_results: list[str] = get_multipart_strings_from_marklogic_response(
1323            self._send_to_eval(
1324                vars,
1325                "resolve_from_identifier_slug.xqy",
1326            ),
1327        )
1328        return IdentifierResolutions.from_marklogic_output(raw_results)
1329
1330    def resolve_from_identifier_value(
1331        self, identifier_value: DocumentIdentifierValue, published_only: bool = True
1332    ) -> IdentifierResolutions:
1333        """Given a PUI/EUI url, look up the precomputed slug and return the
1334        MarkLogic document URIs which match that slug. Multiple returns should be anticipated"""
1335        vars: query_dicts.ResolveFromIdentifierValueDict = {
1336            "identifier_value": identifier_value,
1337            "published_only": int(published_only),
1338        }
1339        raw_results: list[str] = get_multipart_strings_from_marklogic_response(
1340            self._send_to_eval(
1341                vars,
1342                "resolve_from_identifier_value.xqy",
1343            ),
1344        )
1345        return IdentifierResolutions.from_marklogic_output(raw_results)
1346
1347    def get_next_document_sequence_number(self) -> int:
1348        """Increment the MarkLogic sequence number by one and return the value."""
1349        return int(self._eval_and_decode({}, "get_next_document_sequence_number.xqy"))