caselawclient.models.utilities.move

 1from typing import TYPE_CHECKING, Any, Optional
 2
 3import ds_caselaw_utils as caselawutils
 4from ds_caselaw_utils.types import NeutralCitationString
 5
 6from caselawclient.errors import MarklogicAPIError
 7from caselawclient.models.utilities.aws import copy_assets
 8from caselawclient.types import DocumentURIString
 9
10if TYPE_CHECKING:
11    from caselawclient.Client import MarklogicApiClient
12
13
14class NeutralCitationToUriError(Exception):
15    pass
16
17
18class MoveJudgmentError(Exception):
19    pass
20
21
22def update_document_uri(
23    source_uri: DocumentURIString, target_citation: NeutralCitationString, api_client: "MarklogicApiClient"
24) -> DocumentURIString:
25    """
26    Move the document at source_uri to the correct location based on the neutral citation
27    The new neutral citation *must* not already exist (that is handled elsewhere)
28
29    :param source_uri: The URI with the contents of the document to be written. (possibly a failure url)
30    :param target_citation: The NCN (implying an unused URL) where the document will be written to
31    :param api_client: An instance of MarklogicApiClient used to make the search request
32    :return: The URL associated with the `target_citation`
33    """
34    new_ncn_based_uri = caselawutils.neutral_url(target_citation)
35    new_uri: Optional[DocumentURIString] = DocumentURIString(new_ncn_based_uri) if new_ncn_based_uri else None
36    if new_uri is None:
37        raise NeutralCitationToUriError(
38            f"Unable to form new URI for {source_uri} from neutral citation: {target_citation}",
39        )
40
41    if api_client.document_exists(new_uri):
42        raise MoveJudgmentError(
43            f"The URI {new_uri} generated from {target_citation} already exists, you cannot move this judgment to a"
44            f" pre-existing Neutral Citation Number.",
45        )
46
47    try:
48        api_client.copy_document(source_uri, new_uri)
49        set_metadata(source_uri, new_uri, api_client)
50        copy_assets(source_uri, new_uri)
51        api_client.set_judgment_this_uri(new_uri)
52    except MarklogicAPIError as e:
53        raise MoveJudgmentError(
54            f"Failure when attempting to copy judgment from {source_uri} to {new_uri}: {e}",
55        )
56
57    try:
58        api_client.delete_judgment(source_uri)
59    except MarklogicAPIError as e:
60        raise MoveJudgmentError(
61            f"Failure when attempting to delete judgment from {source_uri}: {e}",
62        )
63
64    return new_uri
65
66
67def set_metadata(old_uri: DocumentURIString, new_uri: DocumentURIString, api_client: Any) -> None:
68    source_organisation = api_client.get_property(old_uri, "source-organisation")
69    source_name = api_client.get_property(old_uri, "source-name")
70    source_email = api_client.get_property(old_uri, "source-email")
71    transfer_consignment_reference = api_client.get_property(
72        old_uri,
73        "transfer-consignment-reference",
74    )
75    transfer_received_at = api_client.get_property(old_uri, "transfer-received-at")
76    for key, value in [
77        ("source-organisation", source_organisation),
78        ("source-name", source_name),
79        ("source-email", source_email),
80        ("transfer-consignment-reference", transfer_consignment_reference),
81        ("transfer-received-at", transfer_received_at),
82    ]:
83        if value is not None:
84            api_client.set_property(new_uri, key, value)
85
86    """
87    `published` is a boolean property and set differently, technically
88    these failures should be unpublished but copy the property just in case.
89    """
90    published = api_client.get_published(old_uri)
91    api_client.set_boolean_property(new_uri, "published", bool(published))