For work, I have had an interesting project where I had to pass through multiple AWS Backup Plans into a Python class to make all my Backup Plans; these backup plans had so many options that needed configuring that the most logical way of passing the variables through was with a dictionary. It's a bit like a Struct in GoLang.
When approaching this task, the first milestone was to get Python to read through the dictionary. I did this by hardcoding the dictionaries into my Python class and choosing a toggle to select that dictionary to see if Python was reading my dictionaries correctly.
The code below shows the passing of the toggle backup_type and the hard coding of the dictionary.
class BackupPlanCreator:
def __init__(self, vault_name: str, base_tags: dict, backup_type: str):
self.vault_name = vault_name
self.base_tags = base_tags
self.backup_type = backup_type
self.backup_vault = aws.backup.Vault(
resource_name="backupVault",
name=self.vault_name,
tags=self.base_tags,
)
self.backup_configs = {
"hourly": {
"name": "bkp-hourly-ret7-jc",
"rule_name": "hourly-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 * * * ? *)",
"delete_after_days": 7,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
"daily": {
"name": "bkp-daily-ret14-jc",
"rule_name": "daily-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 5 ? * * *)",
"delete_after_days": 30,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
"weekly": {
"name": "bkp-weekly-ret365-jc",
"rule_name": "weekly-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 6 1 * ? *)",
"delete_after_days": 365,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
"monthly": {
"name": "bkp-monthly-ret365-jc",
"rule_name": "monthly-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 6 1 * ? *)",
"delete_after_days": 365,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
}
Below is the same class reading through the dictionary and compiling the backup_rules for the ultimate plan.
config = self.backup_configs.get(self.backup_type)
if not config:
raise ValueError(
"Invalid backup type specified. Choose from: 'hourly', 'daily', 'weekly', 'monthly'."
)
backup_rules = [
{
"rule_name": config["rule_name"],
"target_vault_name": self.backup_vault.name,
"completion_window": config["completion_window"],
"copy_actions": [
{
"destination_vault_arn": self.backup_vault.arn,
"lifecycle": {
"cold_storage_after": 0,
"delete_after": config["delete_after_days"],
"opt_in_to_archive_for_supported_resources": False,
},
}
],
"enable_continuous_backup": False,
"lifecycle": {
"cold_storage_after": 0,
"delete_after": config["delete_after_days"],
"opt_in_to_archive_for_supported_resources": False,
},
"schedule": config["schedule"],
"schedule_expression_timezone": "UTC",
"start_window": config["start_window"],
}
]
advanced_backup_settings = [
{
"backup_options": {
"WindowsVSS": (
"enabled" if config["enable_windows_vss"] else "disabled"
),
},
"resource_type": "EC2",
}
]
As you can see, a for loop is not currently needed, as I am only seeing if Python can read through and select the correct dictionary. I then pass backup_type to the Python Backup Plan Function as a string. This one is daily to choose the daily backup plan out of the dictionary.
backup_creator = backup.BackupPlanCreator(vault_name, base_tags, "daily")
This is great and shows me that I can process a dictionary. However, I want to take this a step further and pass the dictionary around. To achieve this, I moved the dictionary out of the class and into a variable to see if I could pass it through and get the same result.
import pulumi
import pulumi_aws as aws
import jc_aws_backup as backup
backup_configs = {
"hourly": {
"name": "bkp-hourly-ret7-jc",
"rule_name": "hourly-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 * * * ? *)",
"delete_after_days": 7,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
"daily": {
"name": "bkp-daily-ret14-jc",
"rule_name": "daily-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 5 ? * * *)",
"delete_after_days": 30,
"start_window": 60,
"completion_window": 125,
"enable_windows_vss": True,
},
"weekly": {
"name": "bkp-weekly-ret365-jc",
"rule_name": "weekly-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 6 1 * ? *)",
"delete_after_days": 365,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
"monthly": {
"name": "bkp-monthly-ret365-jc",
"rule_name": "monthly-backup-rule",
"target_vault_name": self.backup_vault.name,
"schedule": "cron(0 6 1 * ? *)",
"delete_after_days": 365,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": True,
},
}
backup_creator = backup.BackupPlanCreator(vault_name, base_tags, backup_configs, "daily")
import pulumi
import pulumi_aws as aws
class CorityBackupPlanCreator:
def __init__(
self, vault_name: str, base_tags: dict, backup_configs: dict, backup_type: str
):
self.base_tags = base_tags
self.backup_type = backup_type
self.backup_configs = backup_configs
self.backup_vault = aws.backup.Vault(
vault_name,
tags=self.base_tags,
)
config = self.backup_configs.get(self.backup_type)
if not config:
raise ValueError(
"Invalid backup type specified. Choose from: 'hourly', 'daily', 'weekly', 'monthly'."
)
backup_rules = []
for _, backup_details in self.backup_configs.items():
backup_rules.append(
aws.backup.PlanRuleArgs(
rule_name=backup_details["rule_name"],
target_vault_name=self.backup_vault.name,
schedule=backup_details["schedule"],
lifecycle=aws.backup.PlanRuleLifecycleArgs(
delete_after=backup_details["delete_after_days"]
),
start_window=backup_details["start_window"],
completion_window=backup_details["completion_window"],
recovery_point_tags=self.base_tags,
)
)
advanced_backup_settings = []
for _, backup_details in self.backup_configs.items():
advanced_backup_settings.append(
aws.backup.PlanAdvancedBackupSettingArgs(
backup_options={
"WindowsVSS": (
"enabled"
if backup_details["enable_windows_vss"]
else "disabled"
)
},
resource_type="EC2",
)
)
self.plan_resource = aws.backup.Plan(
resource_name=backup_details["rule_name"],
rules=backup_rules,
advanced_backup_settings=advanced_backup_settings,
tags=self.base_tags,
)
pulumi.export("backupPlanId", self.plan_resource.id)
What I also did here was add a for loop because my logic will change later, and I will want all four backup plans made at once from the dictionary. Yet this showed me that I could pass around a dictionary, and If I could pass around a dictionary, that meant I could also make it come from the central pulumi.yaml file and load it as an json object into the class. This is because when working with Pulumi, the idea you want to get to is to pass everything through one central yaml file that is linked to each environment stack. That is always the end goal. I then changed my main.py code around a bit like this to achieve this.
import pulumi
import pulumi_aws as aws
import json
import jc_aws_backup as backup
backup_configs_json = pulumi.Config().require("backupConfigs")
vault_name = f"bkp-vault-{project_profile}-{project_short_region}"[:42]
try:
backup_configs = json.loads(backup_configs_json)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in backupConfigs: {e}")
backup_creator = backup.BackupPlanCreator(
vault_name=vault_name,
base_tags=base_tags,
backup_type="daily"
backup_configs=backup_configs,
)
I also added some Error Handling. This is because when I first passed through the json from the yaml file, I got a panic error shown below, and it was a pretty unhelpful error in figuring out where in the json it was failing.
Diagnostics:
pulumi:pulumi:Stack (jcontent-dev-v2-jcontent_dev_v2):
error: Program failed with an unhandled exception:
Traceback (most recent call last):
File "C:\DevOps\infrastructure.pulumi\iac-development\jcontent-dev-v2\__main__.py", line 70, in <module>
backup_configs = json.loads(backup_configs_json)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Python312\Lib\json\__init__.py", line 346, in loads
return _default_decoder.decode(s)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Python312\Lib\json\decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Python312\Lib\json\decoder.py", line 355, in raw_decode
raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 23 column 27 (char 602)
By creating my Error handling, I get it to show me where exactly in the json object it is failing.
I have also supplied the yaml file below to show you how to pass in the object. This stumped me for a little while, and having an example would have helped.
encryptionsalt: VALUE
config:
aws:region: us-east-2
environment: jc_testing_pulumi
backupConfigs: |
{
"hourly": {
"name": "bkp-hourly-ret7-jc",
"rule_name": "hourly-backup-rule",
"schedule": "cron(0 * * * ? *)",
"delete_after_days": 7,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": true
},
"daily": {
"name": "bkp-daily-ret14-jc",
"rule_name": "daily-backup-rule",
"schedule": "cron(0 5 ? * * *)",
"delete_after_days": 30,
"start_window": 60,
"completion_window": 125,
"enable_windows_vss": true
},
"weekly": {
"name": "bkp-weekly-ret365-jc",
"rule_name": "weekly-backup-rule",
"schedule": "cron(0 6 ? * 2 *)",
"delete_after_days": 365,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": true
},
"monthly": {
"name": "bkp-monthly-ret365-jc",
"rule_name": "monthly-backup-rule",
"schedule": "cron(0 6 1 * ? *)",
"delete_after_days": 365,
"start_window": 60,
"completion_window": 120,
"enable_windows_vss": true
}
}
This now prepares me for my final hurdle and task; what I actually want to happen is for the backup plan class to make all four backup plans in one run and name them accordingly. How I achieved this was first to remove backup_type from my Class and to change my for loop around to include the aws.backup.plan() function. Yet, I still needed it to name each plan for me. The code below shows how I achieved this.
import json
import pulumi
import pulumi_aws as aws
class BackupPlanCreator:
def __init__(self, vault_name: str, base_tags: dict, backup_configs: dict):
self.base_tags = base_tags
self.backup_configs = backup_configs
self.backup_vault = aws.backup.Vault(
vault_name,
tags=self.base_tags,
)
for backup_name, backup_details in self.backup_configs.items():
backup_plan = aws.backup.Plan(
resource_name=f"{backup_name}-backup-plan",
rules=[
{
"rule_name": backup_details["rule_name"],
"target_vault_name": self.backup_vault.name,
"schedule": backup_details["schedule"],
"lifecycle": {
"delete_after": backup_details["delete_after_days"],
},
"start_window": backup_details["start_window"],
"completion_window": backup_details["completion_window"],
"recovery_point_tags": self.base_tags,
}
],
advanced_backup_settings=[
{
"backup_options": {
"WindowsVSS": (
"enabled"
if backup_details["enable_windows_vss"]
else "disabled"
),
},
"resource_type": "EC2",
}
],
tags=self.base_tags,
)
The call to this function is similar to before, without the backup_type part being passed.
import pulumi
import pulumi_aws as aws
import json
import jc_aws_backup as backup
backup_configs_json = pulumi.Config().require("backupConfigs")
vault_name = f"bkp-vault-{project_profile}-{project_short_region}"[:42]
try:
backup_configs = json.loads(backup_configs_json)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in backupConfigs: {e}")
backup_creator = backup.BackupPlanCreator(
vault_name=vault_name,
base_tags=base_tags,
backup_configs=backup_configs,
)
And that’s it. That’s how I have managed to pass around a dictionary and evolve that into an actual, useful json object to then make multiple plans for one backup plan run.
I will build on this dictionary now and put in some functionality that you see in Terraform. If you want to use default values, you can, or if you're going to override those values, you can. The thing with my dictionary right now is everything is being passed through, but essentially, the engineer will not want to do that. They will probably only want to configure certain parts of the dictionary from the yaml file.
Yet this is a cliffhanger moment for another blog post for another time.
Till next time, happy coding.
