Provisioning Cloud Resources

import boto3
 
def provision_queue(sqs, name: str) -> str:
    try:
        resp = sqs.get_queue_url(QueueName=name)
        return resp["QueueUrl"]
    except sqs.exceptions.QueueDoesNotExist:
        resp = sqs.create_queue(
            QueueName=name,
            Attributes={"VisibilityTimeout": "60"},
        )
        return resp["QueueUrl"]

"""provision.py - idempotent AWS resource bootstrap."""
from __future__ import annotations
 
import json
import logging
from dataclasses import dataclass
 
import boto3
from botocore.exceptions import ClientError
 
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("provision")
 
@dataclass(frozen=True)
class ProvisionSpec:
    bucket: str
    queue: str
    tags: dict[str, str]
 
def ensure_bucket(s3, name: str, tags: dict[str, str]) -> str:
    buckets = {b["Name"] for b in s3.list_buckets().get("Buckets", [])}
    if name not in buckets:
        s3.create_bucket(Bucket=name)
        log.info(json.dumps({"action": "create", "resource": "s3", "name": name}))
    s3.put_bucket_tagging(Bucket=name, Tagging={"TagSet": [{"Key": k, "Value": v} for k, v in tags.items()]})
    return name
 
def ensure_queue(sqs, name: str) -> str:
    try:
        return sqs.get_queue_url(QueueName=name)["QueueUrl"]
    except ClientError as exc:
        if exc.response["Error"]["Code"] != "AWS.SimpleQueueService.NonExistentQueue":
            raise
    url = sqs.create_queue(QueueName=name)["QueueUrl"]
    log.info(json.dumps({"action": "create", "resource": "sqs", "name": name}))
    return url
 
def main() -> None:
    spec = ProvisionSpec(
        bucket="company-events-dev",
        queue="company-events-dev",
        tags={"Environment": "dev", "Owner": "platform"},
    )
    session = boto3.Session()
    s3 = session.client("s3")
    sqs = session.client("sqs")
    ensure_bucket(s3, spec.bucket, spec.tags)
    ensure_queue(sqs, spec.queue)
    log.info(json.dumps({"status": "ok", "bucket": spec.bucket, "queue": spec.queue}))
 
if __name__ == "__main__":
    main()

from botocore.config import Config
 
retry_config = Config(retries={"max_attempts": 10, "mode": "standard"})
s3 = boto3.client("s3", config=retry_config)