diff --git a/scripts/put_pointers_from_files.py b/scripts/put_pointers_from_files.py new file mode 100644 index 000000000..12a901b14 --- /dev/null +++ b/scripts/put_pointers_from_files.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python +# Put pointers from the provided files into the pointers table +# This will overwrite the pointer if it already exists in the table +import json +import os + +import fire +from aws_session_assume import get_boto_session + +from nrlf.core.dynamodb.model import DocumentPointer +from nrlf.core.logger import logger +from nrlf.producer.fhir.r4.model import DocumentReference + +logger.setLevel("ERROR") + +SKIP_PROD_WARNING = os.getenv("SKIP_PROD_WARNING", "false") + + +def _put_pointers_from_files( + *filenames, env: str = "dev", table_name: str | None = None +): + if env == "prod" and SKIP_PROD_WARNING != "true": + confirmation = input( + "\nWARNING - This command will modify the PROD environment. Continue? [y/n] " + ) + if confirmation != "y": + return "Exiting at user request" + + docrefs: list[DocumentReference] = [] + print("Reading docrefs from files...") + for filename in filenames: + with open(filename) as f: + docref_json = json.load(f) + docref = DocumentReference.model_validate(docref_json) + docrefs.append(docref) + + session = get_boto_session(env) + dynamodb = session.resource("dynamodb") + if not table_name: + table_name = f"nhsd-nrlf--{env}-pointers-table" + table = dynamodb.Table(table_name) + + for docref in docrefs: + try: + print(f"Putting {docref.id}....") + pointer = DocumentPointer.from_document_reference(docref) + table.put_item(Item=pointer.model_dump()) + except Exception as e: + print(f"Unable to put pointer for {docref.id}. Error: {e}") + + +if __name__ == "__main__": + fire.Fire(_put_pointers_from_files) diff --git a/scripts/set_pointer_supersede.py b/scripts/set_pointer_supersede.py new file mode 100644 index 000000000..ab6ff56f5 --- /dev/null +++ b/scripts/set_pointer_supersede.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python +# Set supersede info on a pointer +import json +import os + +import aws_session_assume +import fire + +from nrlf.core.dynamodb.model import DocumentPointer +from nrlf.core.logger import logger +from nrlf.producer.fhir.r4.model import ( + DocumentReference, + DocumentReferenceRelatesTo, + Identifier, + Reference, +) + +logger.setLevel("ERROR") + +SKIP_PROD_WARNING = os.getenv("SKIP_PROD_WARNING", "false") + + +def _set_pointer_supersede_info( + pointer_id: str, + supersede_pointer_id: str, + delete_superseded: bool = False, + env: str = "dev", + table_name: str | None = None, +): + if env == "prod" and SKIP_PROD_WARNING != "true": + confirmation = input( + "\nWARNING - This command will modify the PROD environment. Continue? [y/n] " + ) + if confirmation != "y": + return "Exiting at user request" + + session = aws_session_assume.get_boto_session(env) + dynamodb = session.resource("dynamodb") + + if not table_name: + table_name = f"nhsd-nrlf--{env}-pointers-table" + table = dynamodb.Table(table_name) + + print( + f"Setting pointer {pointer_id} in {table_name} to supersede {supersede_pointer_id}...." + ) + + try: + doc_key = f"D#{pointer_id}" + print(f"Getting {pointer_id}...") + result = table.get_item( + Key={"pk": doc_key, "sk": doc_key}, + ) + except Exception as e: + print(f"Unable to get pointer. Error: {e}") + return + + if "Item" not in result: + print(f"Unable to set superseded info. Pointer {pointer_id} not found.") + return + + item = result["Item"] + + try: + pointer = DocumentPointer.model_validate({"_from_dynamo": True, **item}) + doc_ref = DocumentReference.model_validate_json(pointer.document) + except Exception as e: + print(f"Could not validate pointer from table. Error: {e}") + return + + if not doc_ref.relatesTo: + doc_ref.relatesTo = [] + + existing_supersedes = [ + relates_to for relates_to in doc_ref.relatesTo if relates_to.code == "replaces" + ] + if existing_supersedes: + print( + f"Unable to add supersede info as pointer is already superseding a pointer: {existing_supersedes}" + ) + return + + doc_ref.relatesTo.append( + DocumentReferenceRelatesTo( + code="replaces", + target=Reference( + type="DocumentReference", + identifier=Identifier(value=supersede_pointer_id), + ), + ) + ) + + print(f"Adding supersede info to {pointer_id}...") + updated_pointer = DocumentPointer.from_document_reference(doc_ref) + table.put_item( + Item=updated_pointer.dict(exclude_none=True, exclude={"_from_dynamo"}) + ) + + if delete_superseded: + print(f"Deleting superseded {supersede_pointer_id}...") + table.delete_item( + Key={"pk": f"D#{supersede_pointer_id}", "sk": f"D#{supersede_pointer_id}"} + ) + + +if __name__ == "__main__": + fire.Fire(_set_pointer_supersede_info)