Started Flask move with message_status

app.py

@@ -0,0 +1,7 @@
+from app import create_app
+
+app = create_app()
+
+if __name__ == "__main__":
+    # TODO: Update this to check for the environment specifically to decide when to run in debug mode
+    app.run(debug=True)

app/__init__.py

@@ -0,0 +1,9 @@
+from flask import Flask
+from app.routes import api
+
+def create_app():
+    app = Flask(__name__)
+
+    app.register_blueprint(api, url_prefix="/api")
+
+    return app

app/routes.py

@@ -0,0 +1,15 @@
+from flask import Blueprint, request, jsonify
+import app.services.message_status.main as message_status
+
+api = Blueprint("api", __name__)
+
+@api.route("/message-status/create", methods=["POST"])
+def create_message_status():
+    data = request.json
+    result = message_status.create_message_status(data)
+    return jsonify(result)
+
+@api.route("/message-status/health-check", methods=["GET"])
+def message_status_health_check():
+    result = message_status.health_check()
+    return result

app/services/message_status/main.py

@@ -0,0 +1,25 @@
+import json
+import logging
+from app.services.message_status import status_recorder, request_verifier
+from flask import jsonify
+
+def create_message_status(req_body, headers):
+    logging.info("MessageStatus HTTP trigger function. Processing callback from NHS Notify service.")
+    logging.debug(req_body)
+
+    if not request_verifier.verify_headers(headers):
+        status_code = 401
+        body = {"status": "error"}
+    elif request_verifier.verify_signature(headers, req_body):
+        body_dict = json.loads(req_body)
+        status_recorder.save_statuses(body_dict)
+        status_code = 200
+        body = {"status": "success"}
+    else:
+        status_code = 403
+        body = {"status": "error"}
+
+    return body, status_code
+
+def health_check():
+    return jsonify({"status": "healthy"}), 200

app/services/message_status/request_verifier.py

@@ -0,0 +1,34 @@
+import hashlib
+import hmac
+import os
+
+API_KEY_HEADER_NAME = 'x-api-key'
+SIGNATURE_HEADER_NAME = 'x-hmac-sha256-signature'
+
+
+def verify_headers(headers: dict) -> bool:
+    if (headers.get(API_KEY_HEADER_NAME) is None or
+            headers.get(API_KEY_HEADER_NAME) != os.getenv('NOTIFY_API_KEY')):
+        return False
+
+    if headers.get(SIGNATURE_HEADER_NAME) is None:
+        return False
+
+    return True
+
+
+def verify_signature(headers: dict, body: str) -> bool:
+    expected_signature = hmac.new(
+        bytes(signature_secret(), 'ASCII'),
+        msg=bytes(body, 'ASCII'),
+        digestmod=hashlib.sha256
+    ).hexdigest()
+
+    return hmac.compare_digest(
+        expected_signature,
+        headers[SIGNATURE_HEADER_NAME],
+    )
+
+
+def signature_secret() -> str:
+    return f"{os.getenv('APPLICATION_ID')}.{os.getenv('NOTIFY_API_KEY')}"

app/services/message_status/status_recorder.py

@@ -0,0 +1,34 @@
+import app.utils.datastore as datastore
+import json
+import logging
+import app.services.message_status.status_validator as status_validator
+
+
+def save_statuses(request_body: dict) -> None:
+    for data in request_body["data"]:
+        status_data = status_params(data)
+        valid, message = status_validator.validate(status_data)
+        if valid:
+            is_channel_status = "channelStatus" in data["attributes"]
+            datastore.create_status_record(status_data, is_channel_status)
+        else:
+            logging.error(f"Validation failed: {message}")
+
+    return None
+
+
+def status_params(status_data: dict):
+    try:
+        attributes = status_data["attributes"]
+        meta = status_data["meta"]
+        status = attributes.get("channelStatus", attributes.get("messageStatus"))
+        return {
+            "details": json.dumps(status_data),
+            "idempotency_key": meta["idempotencyKey"],
+            "message_id": attributes["messageId"],
+            "message_reference": attributes["messageReference"],
+            "status": status,
+        }
+    except KeyError as e:
+        logging.error(f"Missing key: {e}")
+        return None

app/services/message_status/status_validator.py

@@ -0,0 +1,56 @@
+import json
+import uuid
+
+SUCCESS_MESSAGE = "Validation successful"
+FIELDS = {
+    "details": "json",
+    "idempotency_key": str,
+    "message_reference": uuid.UUID,
+    "status": str,
+}
+
+
+def validate(status_params: dict) -> tuple[bool, str]:
+    for field, expected_type in FIELDS.items():
+        if field not in status_params:
+            return False, missing_field_message(field)
+        if not validator_for_type(expected_type)(status_params[field]):
+            return False, invalid_type_message(field, expected_type)
+
+    return True, SUCCESS_MESSAGE
+
+
+def validator_for_type(type) -> callable:
+    if type == str:
+        return valid_string
+    if type == uuid.UUID:
+        return valid_uuid
+    if type == "json":
+        return valid_json
+    return lambda: False
+
+
+def valid_string(value):
+    return isinstance(value, str)
+
+
+def valid_uuid(value):
+    try:
+        return uuid.UUID(value)
+    except ValueError:
+        return False
+
+
+def valid_json(value):
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return False
+
+
+def invalid_type_message(field, expected_type):
+    return f"Invalid type for field {field}. Expected {expected_type}"
+
+
+def missing_field_message(field):
+    return f"Missing required field: {field}"

app/utils/datastore.py

@@ -0,0 +1,86 @@
+import logging
+import os
+import psycopg2
+import app.utils.schema_initialiser as schema_initialiser
+import time
+from typing import Tuple
+
+
+INSERT_BATCH_MESSAGE = """
+    INSERT INTO batch_messages (
+        batch_id,
+        details,
+        message_reference,
+        nhs_number,
+        recipient_id,
+        status
+    ) VALUES (
+        %(batch_id)s,
+        %(details)s,
+        %(message_reference)s,
+        %(nhs_number)s,
+        %(recipient_id)s,
+        %(status)s
+    ) RETURNING batch_id, message_reference"""
+
+
+INSERT_STATUS = """
+    INSERT INTO {status_table} (
+        idempotency_key,
+        message_id,
+        message_reference,
+        details,
+        status
+    ) VALUES (
+        %(idempotency_key)s,
+        %(message_id)s,
+        %(message_reference)s,
+        %(details)s,
+        %(status)s
+    ) RETURNING idempotency_key"""
+
+
+def create_batch_message_record(batch_message_data: dict) -> Tuple[str, str] | None | bool:
+    try:
+        with connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute(INSERT_BATCH_MESSAGE, batch_message_data)
+                return cur.fetchone()
+
+    except psycopg2.Error as e:
+        logging.error("Error creating batch message record")
+        logging.error(f"{type(e).__name__} : {e}")
+        return False
+
+
+def create_status_record(status_data: dict, is_channel_status=False) -> bool | str:
+    status_table = "channel_statuses" if is_channel_status else "message_statuses"
+    statement = INSERT_STATUS.format(status_table=status_table)
+    try:
+        with connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute(statement, status_data)
+
+                return cur.fetchone()[0]
+
+    except psycopg2.Error as e:
+        logging.error("Error creating message status record")
+        logging.error(f"{type(e).__name__} : {e}")
+        return False
+
+
+def connection() -> psycopg2.extensions.connection:
+    start = time.time()
+    conn = psycopg2.connect(
+        dbname=os.environ["DATABASE_NAME"],
+        user=os.environ["DATABASE_USER"],
+        host=os.environ["DATABASE_HOST"],
+        password=os.environ["DATABASE_PASSWORD"],
+        sslmode=os.getenv("DATABASE_SSLMODE", "require"),
+    )
+    end = time.time()
+    logging.debug(f"Connected to database in {(end - start)}s")
+
+    schema_initialiser.check_and_initialise_schema(conn)
+
+    return conn

app/utils/format_date.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+
+CRYSTAL_REPORT_EXTRACT_FORMAT = "%dM%mM%Y"
+
+def _to_format(date_str: str, desired_format: str) -> str | None:
+    try:
+        # If we can parse the date string in the desired format, we can return it as-is
+        datetime.strptime(date_str, desired_format)
+        return date_str
+    except ValueError:
+        pass
+
+    try:
+        return datetime.strptime(date_str, CRYSTAL_REPORT_EXTRACT_FORMAT).strftime(desired_format)
+    except ValueError:
+        return None
+
+
+def to_date_of_birth(date_str: str) -> str | None:
+    return _to_format(date_str, "%Y-%m-%d")
+
+
+def to_human_readable_date(date_str: str) -> str | None:
+    return _to_format(date_str, "%A %d %B %Y")

app/utils/format_time.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+import logging
+
+def to_human_readable_twelve_hours(raw_time: str) -> str | None:
+    """
+    Convert a time string like '11:56:00' to '11:56am' or '12:56pm'.
+    """
+    try:
+        if raw_time is None:
+            logging.error("Attempted to convert time format, but input is None.")
+            return None
+
+        raw_time = raw_time.strip()
+
+        time_obj = datetime.strptime(raw_time, "%H:%M:%S")
+        return time_obj.strftime("%-I:%M%p").lower()
+
+    except (ValueError, TypeError):
+        logging.error(f"Invalid time format: {raw_time}")
+        return None

app/utils/schema_initialiser.py

@@ -0,0 +1,25 @@
+import os
+import psycopg2
+import logging
+
+
+SCHEMA_FILE_PATH = f"{os.path.dirname(__file__)}/database/schema.sql"
+
+# FIXME: This could be replaced with a version number query from a migrations table.
+SCHEMA_CHECK = """
+    SELECT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name = 'channel_statuses')
+"""
+
+
+def check_and_initialise_schema(conn: psycopg2.extensions.connection):
+    if bool(os.getenv("SCHEMA_INITIALISED")):
+        return
+
+    with conn.cursor() as cur:
+        cur.execute(SCHEMA_CHECK)
+        if not bool(cur.fetchone()[0]):
+            logging.debug("Initialising schema")
+            cur.execute(open(SCHEMA_FILE_PATH, "r").read())
+
+    conn.commit()
+    os.environ["SCHEMA_INITIALISED"] = "true"

app/utils/uuid_generator.py

@@ -0,0 +1,21 @@
+import hashlib
+import uuid
+
+
+def recipient_id(message_data: dict) -> str:
+    str_val: str = ",".join(list(filter(None, (
+        message_data["appointment_date"],
+        message_data["appointment_time"],
+        message_data["date_of_birth"],
+        message_data["nhs_number"],
+    ))))
+    return reference_uuid(str_val)
+
+
+def uuid4_str() -> str:
+    return str(uuid.uuid4())
+
+
+def reference_uuid(val) -> str:
+    str_val = str(val)
+    return str(uuid.UUID(hashlib.md5(str_val.encode()).hexdigest()))

app_tests/README.md

@@ -0,0 +1 @@
+This directory contains the tests for the Flask app. This should be renamed "tests" when the previous tests directory is no longer needed (i.e. when we switch over from non Flask to Flask)