Disaster recovery utils

aws/authentication/generate_aws_resource.py

@@ -26,26 +26,34 @@ def get_session_for_stage(stage: Stage | str) -> Session:
             print(f"User: {identity['Arn'].split('/')[2]}")
             break
         except botocore.exceptions.ProfileNotFound:
-            input(f"Couldn't find stage {stage_profile} in your awscli credentials file.\n"
-                  f">> Please run `aws configure sso` and set up profile {stage_profile} to set up your credentials."
-                  f"Hit enter to try again")
+            input(
+                f"Couldn't find stage {stage_profile} in your awscli credentials file.\n"
+                f">> Please run `aws configure sso` and set up profile {stage_profile} to set up your credentials."
+                f"Hit enter to try again"
+            )
         except botocore.exceptions.ClientError:
             # Thrown when profile manually configured in ~/.aws/credentials
-            input(f"\nInvalid Access key ID / Secret access key / Session token for profile {stage_profile}.\n"
-                  f"Update the credentials in your .aws/credentials file for the {stage_profile} profile.\n"
-                  f"Hit enter to try again")
+            input(
+                f"\nInvalid Access key ID / Secret access key / Session token for profile {stage_profile}.\n"
+                f"Update the credentials in your .aws/credentials file for the {stage_profile} profile.\n"
+                f"Hit enter to try again"
+            )
         except botocore.exceptions.TokenRetrievalError:
             # Thrown when using SSO and credentials have expired
-            input(f"\nInvalid or expired credentials for profile {stage_profile}.\n"
-                  f"aws sso login --profile {stage_profile}; || - Run to refresh.\n"
-                  f"If you have a {stage_profile} AWS SSO profile, hit enter to run this command and log in:")
+            input(
+                f"\nInvalid or expired credentials for profile {stage_profile}.\n"
+                f"aws sso login --profile {stage_profile}; || - Run to refresh.\n"
+                f"If you have a {stage_profile} AWS SSO profile, hit enter to run this command and log in:"
+            )
             subprocess.Popen(["aws", "sso", "login", "--profile", stage_profile]).wait()
             input("\nHit enter to try running the script again")
         except botocore.exceptions.NoRegionError:
             # Thrown when there is no region configured for the profile
-            input(f"\nNo region configured for profile {stage_profile}.\n"
-                  f"Set a default region in your .aws/config file for the {stage_profile} profile.\n"
-                  f"Hit enter to try again")
+            input(
+                f"\nNo region configured for profile {stage_profile}.\n"
+                f"Set a default region in your .aws/config file for the {stage_profile} profile.\n"
+                f"Hit enter to try again"
+            )
     return session
 
 
@@ -61,11 +69,11 @@ def generate_aws_service(service_name: str, stage: Stage) -> Any:
     # Info on resources: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/resources.html
     # Info on clients: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/clients.html
     valid_resources = ["dynamodb"]
-    valid_clients = ["ssm", "rds", "es", "opensearch"]
+    valid_clients = ["ssm", "lambda", "logs", "rds", "es", "opensearch", "sns"]
     if service_name in valid_resources:
-        service: ServiceResource = session.resource(service_name) # type: ignore
+        service: ServiceResource = session.resource(service_name)  # type: ignore
     elif service_name in valid_clients:
-        service: ServiceResource = session.client(service_name) # type: ignore
+        service: ServiceResource = session.client(service_name)  # type: ignore
     else:
         raise ValueError(f"Valid service names are {valid_resources + valid_clients}")
     return service

aws/sns/publish_message.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Dict, Optional
+
+from pydantic import BaseModel, ConfigDict
+
+from aws.authentication.generate_aws_resource import generate_aws_service
+from enums.enums import Stage
+
+
+class EntityEventSns(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    eventType: str
+    sourceDomain: str
+    sourceSystem: str
+    version: str
+    correlationId: str
+    dateTime: datetime
+    user: User
+    entityId: str
+    eventData: EventData
+
+    class EventData(BaseModel):
+        oldData: Optional[Dict]
+        newData: Dict
+
+    class User(BaseModel):
+        name: str
+        email: str
+
+
+def create_sns_message(
+    event_type: str,
+    user: EntityEventSns.User,
+    entity_id: str,
+    event_data: EntityEventSns.EventData,
+) -> EntityEventSns:
+    return EntityEventSns(
+        id=entity_id,
+        eventType=event_type,
+        sourceDomain="mtfh-api",
+        sourceSystem="mtfh-api",
+        version="v1",
+        correlationId=str(uuid.uuid4()),
+        dateTime=datetime.now(timezone.utc),
+        user=user,
+        entityId=entity_id,
+        eventData=event_data,
+    )
+
+
+def main():
+    sns_client = generate_aws_service("sns", Stage.DISASTER_RECOVERY)
+    event: EntityEventSns = create_sns_message(
+        user=EntityEventSns.User(name="System", email="system@example.com"),
+        event_type="TenureCreatedEvent",
+        entity_id="de414b82-24c0-aaed-64e7-c51f79b1a08d",
+        event_data=EntityEventSns.EventData(oldData={}, newData={}),
+    )
+    topic_arn = "arn:aws:sns:eu-west-2:851725205572:cautionaryalerts.fifo"
+    print(event.model_dump_json())
+    sns_client.publish(
+        Message=event.model_dump_json(),
+        TopicArn=topic_arn,
+        MessageGroupId="fake",
+    )
+    print("Published event")
+
+
+if __name__ == "__main__":
+    main()

aws/ssm/copy_ssm_parameters.py

@@ -70,16 +70,20 @@ def migrate_parameters(parameters: list[ParameterTypeDef]):
         print(parameter["Name"])
 
         # Check if parameter already exists in target account - skip if values match
-        existing_param = ssm_target.get_parameter(
-            Name=parameter["Name"], WithDecryption=True
-        ).get("Parameter")
-        if existing_param and "Value" in existing_param:
-            print(f"Param {parameter['Name']} already: {existing_param['Value']}")
-            if existing_param["Value"] == parameter["Value"]:
-                print("Values match, skipping...")
-                continue
-            else:
-                print("Values differ.")
+        try:
+            existing_param = ssm_target.get_parameter(
+                Name=parameter["Name"], WithDecryption=True
+            ).get("Parameter")
+            if existing_param and "Value" in existing_param:
+                print(f"Param {parameter['Name']} already: {existing_param['Value']}")
+                if existing_param["Value"] == parameter["Value"]:
+                    print("Values match, skipping...")
+                    continue
+                else:
+                    print("Values differ.")
+        except ssm_target.exceptions.ParameterNotFound:
+            print("Parameter not found in target account, proceeding to create.")
+            pass
 
         # Ask for confirmation before overwriting
         if not confirm(f"Copy parameter {parameter['Name']} to target account?"):

enums/enums.py

@@ -9,14 +9,16 @@ class Stage(Enum):
     HOUSING_PRODUCTION = "housing-production"
     HOUSING_STAGING = "housing-staging"
     HOUSING_DEVELOPMENT = "housing-development"
-    BASE_DEVELOPMENT = "base-development"
-    BASE_STAGING = "base-staging"
-    BASE_PRODUCTION = "base-production"
+    BASE_DEVELOPMENT = "development-apis"
+    BASE_STAGING = "staging-apis"
+    BASE_PRODUCTION = "production-apis"
     DISASTER_RECOVERY = "disaster-recovery"
 
     def to_env_name(self) -> Environment:
         value = self.value
         for stage_str in ["development", "staging", "production"]:
             if stage_str in value:
                 return cast(Environment, stage_str)
+        if value == "disaster-recovery":
+            return "production"
         raise ValueError(f"Stage {self.value} not recognised")