Python for AWS & DevOps - Interview Questions & Answers

import boto3

# 1. IAM Role (recommended in AWS environments — EC2, Lambda, ECS)
# No credentials needed — SDK picks them up automatically from instance metadata
s3 = boto3.client("s3")

# 2. Environment variables
# AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN (optional)
import os
os.environ["AWS_ACCESS_KEY_ID"] = "AKIA..."
os.environ["AWS_SECRET_ACCESS_KEY"] = "..."

# 3. AWS credentials file (~/.aws/credentials)
# [default]
# aws_access_key_id = AKIA...
# aws_secret_access_key = ...

# 4. Explicit credentials (avoid hardcoding — use Secrets Manager instead)
client = boto3.client(
    "s3",
    region_name="us-east-1",
    aws_access_key_id="AKIA...",
    aws_secret_access_key="...",
)

# 5. Named profile
session = boto3.Session(profile_name="dev-account")
s3 = session.client("s3")

import boto3

# client — low level, returns dicts
client = boto3.client("s3")
response = client.list_buckets()
buckets = response["Buckets"]          # List of dicts
for b in buckets:
    print(b["Name"], b["CreationDate"])

# resource — high level, returns objects
s3 = boto3.resource("s3")
for bucket in s3.buckets.all():        # Bucket objects
    print(bucket.name, bucket.creation_date)

# Mixing both — get a resource's underlying client
bucket = s3.Bucket("my-bucket")
low_level_client = bucket.meta.client

import boto3
from pathlib import Path

s3 = boto3.client("s3", region_name="us-east-1")
BUCKET = "my-company-artifacts"

# --- Upload ---
# upload_file — reads from disk (streams, handles large files)
s3.upload_file(
    Filename="build/app.zip",
    Bucket=BUCKET,
    Key="releases/v1.2.0/app.zip",
    ExtraArgs={"ServerSideEncryption": "AES256"},
)

# put_object — upload bytes/strings directly (small objects)
s3.put_object(
    Bucket=BUCKET,
    Key="config/app.json",
    Body=b'{"env": "prod"}',
    ContentType="application/json",
)

# --- Download ---
s3.download_file(BUCKET, "releases/v1.2.0/app.zip", "/tmp/app.zip")

# get_object — read content into memory
response = s3.get_object(Bucket=BUCKET, Key="config/app.json")
content = response["Body"].read().decode("utf-8")

# --- List objects (paginated) ---
paginator = s3.get_paginator("list_objects_v2")
pages = paginator.paginate(Bucket=BUCKET, Prefix="releases/")

for page in pages:
    for obj in page.get("Contents", []):
        print(obj["Key"], obj["Size"], obj["LastModified"])

# --- Delete ---
s3.delete_object(Bucket=BUCKET, Key="releases/v1.0.0/app.zip")

# Delete multiple objects at once (batch)
s3.delete_objects(
    Bucket=BUCKET,
    Delete={"Objects": [{"Key": "old/file1.txt"}, {"Key": "old/file2.txt"}]},
)

import json
import logging
import os
import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event: dict, context) -> dict:
    """
    event   — trigger payload (dict shape varies by trigger source)
    context — runtime info provided by Lambda
    """
    # --- context attributes ---
    logger.info("Function name:    %s", context.function_name)
    logger.info("Request ID:       %s", context.aws_request_id)
    logger.info("Memory (MB):      %s", context.memory_limit_in_mb)
    logger.info("Time remaining:   %s ms", context.get_remaining_time_in_millis())

    # --- read environment variables ---
    table_name = os.environ["DYNAMODB_TABLE"]

    # --- business logic ---
    try:
        body = json.loads(event.get("body", "{}"))
        user_id = body["user_id"]

        dynamodb = boto3.resource("dynamodb")
        table = dynamodb.Table(table_name)
        table.put_item(Item={"user_id": user_id, "status": "active"})

        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"message": "User created", "user_id": user_id}),
        }

    except KeyError as e:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": f"Missing field: {e}"}),
        }
    except Exception as e:
        logger.exception("Unexpected error")
        return {"statusCode": 500, "body": json.dumps({"error": "Internal error"})}

import boto3
import time

ec2 = boto3.client("ec2", region_name="us-east-1")

# --- Launch an instance ---
response = ec2.run_instances(
    ImageId="ami-0c55b159cbfafe1f0",   # Amazon Linux 2
    InstanceType="t3.micro",
    MinCount=1,
    MaxCount=1,
    KeyName="my-key-pair",
    SecurityGroupIds=["sg-0123456789abcdef0"],
    SubnetId="subnet-0123456789abcdef0",
    TagSpecifications=[{
        "ResourceType": "instance",
        "Tags": [
            {"Key": "Name", "Value": "web-server-01"},
            {"Key": "Environment", "Value": "prod"},
        ],
    }],
    UserData="""#!/bin/bash
yum update -y
yum install -y nginx
systemctl start nginx
""",
)
instance_id = response["Instances"][0]["InstanceId"]
print(f"Launched: {instance_id}")

# --- Wait until running ---
waiter = ec2.get_waiter("instance_running")
waiter.wait(InstanceIds=[instance_id])
print("Instance is running")

# --- Describe / filter instances ---
response = ec2.describe_instances(
    Filters=[
        {"Name": "tag:Environment", "Values": ["prod"]},
        {"Name": "instance-state-name", "Values": ["running"]},
    ]
)

for reservation in response["Reservations"]:
    for inst in reservation["Instances"]:
        print(inst["InstanceId"], inst["PrivateIpAddress"])

# --- Stop / Start / Terminate ---
ec2.stop_instances(InstanceIds=[instance_id])
ec2.start_instances(InstanceIds=[instance_id])
ec2.terminate_instances(InstanceIds=[instance_id])

import boto3
import json
import os

# --- Secrets Manager ---
def get_secret(secret_name: str, region: str = "us-east-1") -> dict:
    client = boto3.client("secretsmanager", region_name=region)
    response = client.get_secret_value(SecretId=secret_name)

    # Secret can be a JSON string or a plain string
    if "SecretString" in response:
        return json.loads(response["SecretString"])
    else:
        # Binary secret
        return response["SecretBinary"]

# Usage — never hardcode DB credentials
db_creds = get_secret("prod/myapp/postgres")
conn_str = f"postgresql://{db_creds['username']}:{db_creds['password']}@{db_creds['host']}/mydb"


# --- SSM Parameter Store ---
ssm = boto3.client("ssm", region_name="us-east-1")

# Single parameter (SecureString is decrypted automatically with WithDecryption=True)
response = ssm.get_parameter(
    Name="/myapp/prod/api_key",
    WithDecryption=True,
)
api_key = response["Parameter"]["Value"]

# Fetch all parameters for an environment at once
response = ssm.get_parameters_by_path(
    Path="/myapp/prod/",
    Recursive=True,
    WithDecryption=True,
)
config = {p["Name"].split("/")[-1]: p["Value"] for p in response["Parameters"]}
# {"api_key": "...", "db_host": "...", "feature_flag": "true"}


# --- Best practice pattern — cache at Lambda cold start ---
_cache: dict = {}

def get_param(name: str) -> str:
    if name not in _cache:
        ssm = boto3.client("ssm")
        _cache[name] = ssm.get_parameter(Name=name, WithDecryption=True)["Parameter"]["Value"]
    return _cache[name]

import subprocess
import shlex

# --- Basic run (recommended — raises on failure) ---
result = subprocess.run(
    ["git", "pull", "origin", "main"],
    capture_output=True,   # Capture stdout and stderr
    text=True,             # Decode bytes to str
    check=True,            # Raise CalledProcessError if exit code != 0
)
print(result.stdout)

# --- Capture output and check return code manually ---
result = subprocess.run(
    ["docker", "ps", "--format", "{{.Names}}"],
    capture_output=True,
    text=True,
)
if result.returncode == 0:
    containers = result.stdout.strip().splitlines()
else:
    print("Error:", result.stderr)

# --- Run shell commands (use shlex.split to avoid injection) ---
cmd = "kubectl get pods -n production -o json"
result = subprocess.run(shlex.split(cmd), capture_output=True, text=True, check=True)

# --- Piping commands ---
ps = subprocess.Popen(["ps", "aux"], stdout=subprocess.PIPE)
grep = subprocess.Popen(
    ["grep", "nginx"],
    stdin=ps.stdout,
    stdout=subprocess.PIPE,
    text=True,
)
ps.stdout.close()
output, _ = grep.communicate()

# --- Timeout (prevent hanging in CI) ---
try:
    subprocess.run(["make", "test"], timeout=300, check=True)
except subprocess.TimeoutExpired:
    print("Tests timed out after 5 minutes")
except subprocess.CalledProcessError as e:
    print(f"Tests failed with exit code {e.returncode}")
    print(e.stderr)

import json
import yaml      # pip install pyyaml
from pathlib import Path

# --- JSON ---
# Read
with open("config.json") as f:
    config = json.load(f)

# Write (pretty-printed)
with open("output.json", "w") as f:
    json.dump(config, f, indent=2, default=str)  # default=str handles datetime etc.

# Parse string
data = json.loads('{"env": "prod", "replicas": 3}')

# --- YAML ---
# Read single document
with open("deployment.yaml") as f:
    manifest = yaml.safe_load(f)   # Always use safe_load, never yaml.load

# Read multiple documents (Kubernetes multi-resource files)
with open("k8s-resources.yaml") as f:
    docs = list(yaml.safe_load_all(f))

# Write
with open("output.yaml", "w") as f:
    yaml.dump(manifest, f, default_flow_style=False, sort_keys=False)

# --- Practical: patch a Kubernetes manifest ---
with open("deployment.yaml") as f:
    deploy = yaml.safe_load(f)

# Update image tag from CI pipeline
image_tag = "v1.2.3"
deploy["spec"]["template"]["spec"]["containers"][0]["image"] = f"myapp:{image_tag}"
deploy["spec"]["replicas"] = 3

with open("deployment.yaml", "w") as f:
    yaml.dump(deploy, f, default_flow_style=False)

# --- Practical: generate Terraform variable files from Python dicts ---
tf_vars = {"region": "us-east-1", "instance_type": "t3.medium", "min_capacity": 2}
with open("terraform.tfvars.json", "w") as f:
    json.dump(tf_vars, f, indent=2)

import logging
import json
import sys
from datetime import datetime

# --- Basic structured logging (JSON format for log aggregators) ---
class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)


def get_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(JsonFormatter())
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
        logger.propagate = False
    return logger


# --- Usage ---
logger = get_logger(__name__)

logger.info("Deployment started", extra={"version": "v1.2.3"})
logger.warning("High CPU detected", extra={"instance": "i-0abc123", "cpu_pct": 87})

try:
    result = deploy_service()
except Exception:
    logger.exception("Deployment failed")   # Logs traceback automatically
    sys.exit(1)

# --- Lambda-specific: logging level from env var ---
import os
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
logging.getLogger().setLevel(getattr(logging, LOG_LEVEL))

# --- Log levels guideline for DevOps scripts ---
# DEBUG   → detailed tracing (disabled in prod)
# INFO    → normal operations (deployment steps, resource counts)
# WARNING → recoverable issue (retrying, degraded mode)
# ERROR   → operation failed but script continues
# CRITICAL → script must abort

#!/usr/bin/env python3
"""
deploy.py — deployment automation CLI
Usage:
  python deploy.py deploy --env prod --version v1.2.3
  python deploy.py rollback --env staging --steps 1
  python deploy.py status --env prod
"""

import argparse
import sys


def deploy(args):
    print(f"Deploying version {args.version} to {args.env}")
    if args.dry_run:
        print("[DRY RUN] No changes made")


def rollback(args):
    print(f"Rolling back {args.steps} step(s) in {args.env}")


def status(args):
    print(f"Checking status in {args.env}")


def main():
    parser = argparse.ArgumentParser(
        description="Deployment automation tool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--dry-run", action="store_true", help="Preview without making changes")

    subparsers = parser.add_subparsers(dest="command", required=True)

    # deploy subcommand
    deploy_parser = subparsers.add_parser("deploy", help="Deploy a version")
    deploy_parser.add_argument("--env", required=True, choices=["dev", "staging", "prod"])
    deploy_parser.add_argument("--version", required=True, help="Version tag, e.g. v1.2.3")
    deploy_parser.set_defaults(func=deploy)

    # rollback subcommand
    rollback_parser = subparsers.add_parser("rollback", help="Roll back deployments")
    rollback_parser.add_argument("--env", required=True, choices=["dev", "staging", "prod"])
    rollback_parser.add_argument("--steps", type=int, default=1, help="Number of versions to roll back")
    rollback_parser.set_defaults(func=rollback)

    # status subcommand
    status_parser = subparsers.add_parser("status", help="Show deployment status")
    status_parser.add_argument("--env", required=True)
    status_parser.set_defaults(func=status)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()

import docker   # pip install docker

client = docker.from_env()  # Uses DOCKER_HOST or local socket

# --- Build an image ---
image, build_logs = client.images.build(
    path=".",
    tag="myapp:v1.2.3",
    rm=True,            # Remove intermediate containers
    buildargs={"APP_ENV": "production"},
)
for log in build_logs:
    if "stream" in log:
        print(log["stream"], end="")

# --- Run a container ---
container = client.containers.run(
    "myapp:v1.2.3",
    name="myapp-test",
    environment={"DATABASE_URL": "postgresql://localhost/test"},
    ports={"8080/tcp": 8080},
    detach=True,       # Non-blocking
    remove=True,       # Auto-remove when stopped
)

# Wait for it to finish
result = container.wait()
print("Exit code:", result["StatusCode"])
print(container.logs().decode())

# --- One-shot execution (blocking) ---
output = client.containers.run(
    "alpine:3.18",
    command="echo 'hello from container'",
    remove=True,
)
print(output.decode())

# --- List running containers ---
for c in client.containers.list():
    print(c.name, c.status, c.image.tags)

# --- Push to ECR ---
import base64, boto3

ecr = boto3.client("ecr", region_name="us-east-1")
token = ecr.get_authorization_token()["authorizationData"][0]
username, password = base64.b64decode(token["authorizationToken"]).decode().split(":")
registry = token["proxyEndpoint"]

client.images.push(
    f"{registry}/myapp",
    tag="v1.2.3",
    auth_config={"username": username, "password": password},
)

from kubernetes import client, config, watch   # pip install kubernetes

# Load kubeconfig — works locally and in-cluster
try:
    config.load_incluster_config()   # Running inside a pod
except config.ConfigException:
    config.load_kube_config()        # Local ~/.kube/config

v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()

# --- List pods in a namespace ---
pods = v1.list_namespaced_pod(namespace="production")
for pod in pods.items:
    print(pod.metadata.name, pod.status.phase)

# --- Scale a deployment ---
apps_v1.patch_namespaced_deployment_scale(
    name="myapp",
    namespace="production",
    body={"spec": {"replicas": 5}},
)

# --- Create a ConfigMap ---
cm = client.V1ConfigMap(
    metadata=client.V1ObjectMeta(name="app-config", namespace="production"),
    data={"LOG_LEVEL": "INFO", "FEATURE_FLAG": "true"},
)
v1.create_namespaced_config_map(namespace="production", body=cm)

# --- Apply a manifest from a YAML file ---
import yaml
from kubernetes.utils import create_from_yaml

with open("deployment.yaml") as f:
    manifest = yaml.safe_load(f)

k8s_client = client.ApiClient()
create_from_yaml(k8s_client, yaml_objects=[manifest], namespace="production")

# --- Watch pod events (streaming) ---
w = watch.Watch()
for event in w.stream(v1.list_namespaced_pod, namespace="production", timeout_seconds=60):
    print(f"{event['type']}: {event['object'].metadata.name}")
    if event["object"].status.phase == "Running":
        w.stop()

import requests
import json
import smtplib
import boto3
from email.mime.text import MIMEText

# --- Slack webhook ---
def send_slack_alert(message: str, webhook_url: str, channel: str = "#devops-alerts"):
    payload = {
        "channel": channel,
        "text": message,
        "attachments": [{
            "color": "danger",
            "fields": [{"title": "Environment", "value": "production", "short": True}],
        }],
    }
    resp = requests.post(webhook_url, json=payload, timeout=10)
    resp.raise_for_status()

# --- AWS SNS (for PagerDuty/email/SMS fan-out) ---
def send_sns_alert(message: str, subject: str, topic_arn: str):
    sns = boto3.client("sns", region_name="us-east-1")
    sns.publish(
        TopicArn=topic_arn,
        Message=message,
        Subject=subject,
    )

# --- AWS SES email ---
def send_email(to: str, subject: str, body: str, sender: str = "[email protected]"):
    ses = boto3.client("ses", region_name="us-east-1")
    ses.send_email(
        Source=sender,
        Destination={"ToAddresses": [to]},
        Message={
            "Subject": {"Data": subject},
            "Body": {"Text": {"Data": body}},
        },
    )

# --- Practical: wrap a deployment with alerts ---
def deploy_with_alerting(version: str):
    try:
        run_deployment(version)
        send_slack_alert(f":white_check_mark: Deployed {version} successfully")
        send_sns_alert(
            f"Deployment of {version} succeeded",
            subject="Deploy Success",
            topic_arn="arn:aws:sns:us-east-1:123456789012:deploy-notifications",
        )
    except Exception as e:
        send_slack_alert(f":red_circle: Deployment of {version} FAILED: {e}")
        raise

import boto3
from datetime import datetime, timedelta, timezone

cw = boto3.client("cloudwatch", region_name="us-east-1")

# --- Publish custom metrics ---
cw.put_metric_data(
    Namespace="MyApp/Production",
    MetricData=[
        {
            "MetricName": "DeploymentDuration",
            "Dimensions": [
                {"Name": "Environment", "Value": "prod"},
                {"Name": "Service", "Value": "api"},
            ],
            "Value": 45.3,           # seconds
            "Unit": "Seconds",
            "Timestamp": datetime.now(timezone.utc),
        },
        {
            "MetricName": "ActiveUsers",
            "Value": 1250,
            "Unit": "Count",
        },
    ],
)

# --- Query existing metrics ---
now = datetime.now(timezone.utc)
response = cw.get_metric_statistics(
    Namespace="AWS/EC2",
    MetricName="CPUUtilization",
    Dimensions=[{"Name": "InstanceId", "Value": "i-0abc123def456"}],
    StartTime=now - timedelta(hours=1),
    EndTime=now,
    Period=300,        # 5-minute granularity
    Statistics=["Average", "Maximum"],
)
for dp in sorted(response["Datapoints"], key=lambda x: x["Timestamp"]):
    print(f"{dp['Timestamp']}: avg={dp['Average']:.1f}%, max={dp['Maximum']:.1f}%")

# --- Create a CloudWatch alarm from Python ---
cw.put_metric_alarm(
    AlarmName="HighCPU-api-server",
    MetricName="CPUUtilization",
    Namespace="AWS/EC2",
    Statistic="Average",
    Period=300,
    EvaluationPeriods=2,
    Threshold=80.0,
    ComparisonOperator="GreaterThanThreshold",
    Dimensions=[{"Name": "InstanceId", "Value": "i-0abc123def456"}],
    AlarmActions=["arn:aws:sns:us-east-1:123456789012:ops-alerts"],
    TreatMissingData="notBreaching",
)

import boto3

ec2 = boto3.client("ec2", region_name="us-east-1")
s3  = boto3.client("s3")

# --- Method 1: Manual pagination (fragile, avoid) ---
all_instances = []
response = ec2.describe_instances(MaxResults=100)
while True:
    for r in response["Reservations"]:
        all_instances.extend(r["Instances"])
    next_token = response.get("NextToken")
    if not next_token:
        break
    response = ec2.describe_instances(MaxResults=100, NextToken=next_token)

# --- Method 2: boto3 Paginator (recommended) ---
paginator = ec2.get_paginator("describe_instances")
pages = paginator.paginate(
    Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
)
all_instances = [
    inst
    for page in pages
    for r in page["Reservations"]
    for inst in r["Instances"]
]

# --- S3 list objects ---
s3_paginator = s3.get_paginator("list_objects_v2")
all_keys = []
for page in s3_paginator.paginate(Bucket="my-bucket", Prefix="logs/2024/"):
    for obj in page.get("Contents", []):
        all_keys.append(obj["Key"])

print(f"Found {len(all_keys)} objects")

# --- Paginate with page size limit ---
for page in s3_paginator.paginate(
    Bucket="my-bucket",
    PaginationConfig={"MaxItems": 500, "PageSize": 100},
):
    for obj in page.get("Contents", []):
        process(obj)

# See all paginators for a service
ec2.meta.service_model.operation_names
client.can_paginate("describe_instances")  # True if paginator exists

import boto3
import concurrent.futures
from typing import List, Dict

ec2 = boto3.client("ec2", region_name="us-east-1")

REGIONS = ["us-east-1", "us-west-2", "eu-west-1", "ap-southeast-1"]

# --- Sequential (slow — waits for each region) ---
def get_instances_sequential() -> List[Dict]:
    results = []
    for region in REGIONS:
        client = boto3.client("ec2", region_name=region)
        paginator = client.get_paginator("describe_instances")
        for page in paginator.paginate():
            for r in page["Reservations"]:
                for inst in r["Instances"]:
                    results.append({**inst, "Region": region})
    return results

# --- Parallel (fast — all regions queried simultaneously) ---
def get_instances_in_region(region: str) -> List[Dict]:
    client = boto3.client("ec2", region_name=region)
    paginator = client.get_paginator("describe_instances")
    instances = []
    for page in paginator.paginate():
        for r in page["Reservations"]:
            for inst in r["Instances"]:
                instances.append({**inst, "Region": region})
    return instances

def get_all_instances_parallel() -> List[Dict]:
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=len(REGIONS)) as executor:
        futures = {executor.submit(get_instances_in_region, r): r for r in REGIONS}
        for future in concurrent.futures.as_completed(futures):
            region = futures[future]
            try:
                results.extend(future.result())
            except Exception as e:
                print(f"Region {region} failed: {e}")
    return results

# --- Parallel S3 uploads ---
def upload_file(args):
    bucket, key, local_path = args
    boto3.client("s3").upload_file(local_path, bucket, key)
    return key

files = [("my-bucket", f"logs/{f}", f"/tmp/{f}") for f in ["a.log", "b.log", "c.log"]]

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool:
    uploaded = list(pool.map(upload_file, files))
    print(f"Uploaded {len(uploaded)} files")

#!/usr/bin/env python3
"""
Cleanup script: deletes EBS snapshots older than 30 days not attached to any AMI.
Run as a Lambda on a schedule or manually.
"""

import boto3
import logging
from datetime import datetime, timezone, timedelta

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

MAX_AGE_DAYS = 30


def get_ami_snapshot_ids(ec2) -> set:
    """Return all snapshot IDs used by registered AMIs."""
    snap_ids = set()
    paginator = ec2.get_paginator("describe_images")
    for page in paginator.paginate(Owners=["self"]):
        for image in page["Images"]:
            for bd in image.get("BlockDeviceMappings", []):
                if "Ebs" in bd and "SnapshotId" in bd["Ebs"]:
                    snap_ids.add(bd["Ebs"]["SnapshotId"])
    return snap_ids


def cleanup_old_snapshots(dry_run: bool = True) -> dict:
    ec2 = boto3.client("ec2", region_name="us-east-1")
    cutoff = datetime.now(timezone.utc) - timedelta(days=MAX_AGE_DAYS)

    ami_snap_ids = get_ami_snapshot_ids(ec2)
    deleted, skipped, errors = [], [], []

    paginator = ec2.get_paginator("describe_snapshots")
    for page in paginator.paginate(OwnerIds=["self"]):
        for snap in page["Snapshots"]:
            snap_id = snap["SnapshotId"]
            start_time = snap["StartTime"]

            # Skip recent snapshots
            if start_time > cutoff:
                skipped.append(snap_id)
                continue

            # Skip AMI-linked snapshots
            if snap_id in ami_snap_ids:
                logger.info("Skipping AMI snapshot: %s", snap_id)
                skipped.append(snap_id)
                continue

            if dry_run:
                logger.info("[DRY RUN] Would delete: %s (%s, %d GB)",
                            snap_id, start_time.date(), snap["VolumeSize"])
                deleted.append(snap_id)
            else:
                try:
                    ec2.delete_snapshot(SnapshotId=snap_id)
                    logger.info("Deleted: %s", snap_id)
                    deleted.append(snap_id)
                except ec2.exceptions.ClientError as e:
                    logger.error("Failed to delete %s: %s", snap_id, e)
                    errors.append(snap_id)

    return {"deleted": len(deleted), "skipped": len(skipped), "errors": len(errors)}


def lambda_handler(event, context):
    dry_run = event.get("dry_run", False)
    summary = cleanup_old_snapshots(dry_run=dry_run)
    logger.info("Summary: %s", summary)
    return summary


if __name__ == "__main__":
    import argparse
    p = argparse.ArgumentParser()
    p.add_argument("--dry-run", action="store_true")
    args = p.parse_args()
    print(cleanup_old_snapshots(dry_run=args.dry_run))

import boto3
from boto3.dynamodb.conditions import Key, Attr
from decimal import Decimal

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
table = dynamodb.Table("deployments")

# --- Put item ---
table.put_item(Item={
    "service_id": "api-gateway",        # Partition key
    "deploy_time": "2024-01-15T14:30Z", # Sort key
    "version": "v1.2.3",
    "status": "success",
    "duration_s": Decimal("45.3"),      # DynamoDB uses Decimal, not float
    "metadata": {"region": "us-east-1", "triggered_by": "CI"},
})

# --- Get item ---
response = table.get_item(
    Key={"service_id": "api-gateway", "deploy_time": "2024-01-15T14:30Z"}
)
item = response.get("Item")

# --- Query (uses index — efficient) ---
response = table.query(
    KeyConditionExpression=Key("service_id").eq("api-gateway"),
    FilterExpression=Attr("status").eq("success"),
    Limit=10,
    ScanIndexForward=False,   # Sort descending by sort key
)
for item in response["Items"]:
    print(item["deploy_time"], item["version"])

# --- Update item ---
table.update_item(
    Key={"service_id": "api-gateway", "deploy_time": "2024-01-15T14:30Z"},
    UpdateExpression="SET #s = :s, retry_count = retry_count + :inc",
    ExpressionAttributeNames={"#s": "status"},   # "status" is reserved word
    ExpressionAttributeValues={":s": "rolled_back", ":inc": 1},
)

# --- Conditional write (optimistic locking) ---
try:
    table.put_item(
        Item={"service_id": "api-gateway", "deploy_time": "2024-01-15T14:30Z", "status": "deploying"},
        ConditionExpression=Attr("service_id").not_exists(),   # Only if doesn't exist
    )
except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
    print("Item already exists — concurrent write prevented")

# --- Batch write ---
with table.batch_writer() as batch:
    for i in range(50):
        batch.put_item(Item={"service_id": f"svc-{i}", "deploy_time": "2024-01-15", "status": "ok"})

import re
import json
from collections import Counter, defaultdict
from pathlib import Path
from datetime import datetime

# --- Parse structured JSON logs ---
def parse_json_logs(log_file: str) -> list:
    errors = []
    with open(log_file) as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                entry = json.loads(line)
                if entry.get("level") in ("ERROR", "CRITICAL"):
                    errors.append(entry)
            except json.JSONDecodeError:
                pass   # Skip malformed lines
    return errors

# --- Parse unstructured logs with regex ---
LOG_PATTERN = re.compile(
    r'(?P<ip>\d+\.\d+\.\d+\.\d+) .* \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\w+) (?P<path>\S+) HTTP/[\d.]+" '
    r'(?P<status>\d+) (?P<size>\d+)'
)

def parse_access_log(log_file: str) -> dict:
    status_counts = Counter()
    slow_requests = []

    with open(log_file) as f:
        for line in f:
            m = LOG_PATTERN.match(line)
            if not m:
                continue
            status_counts[m.group("status")] += 1
            if int(m.group("size")) > 1_000_000:   # Response > 1 MB
                slow_requests.append(m.group("path"))

    return {
        "total": sum(status_counts.values()),
        "5xx_errors": sum(v for k, v in status_counts.items() if k.startswith("5")),
        "status_breakdown": dict(status_counts.most_common()),
        "large_responses": slow_requests[:10],
    }

# --- Download and analyse CloudWatch logs ---
import boto3

def get_cloudwatch_errors(log_group: str, hours: int = 1) -> list:
    logs = boto3.client("logs", region_name="us-east-1")
    end_ms = int(datetime.utcnow().timestamp() * 1000)
    start_ms = end_ms - hours * 3600 * 1000

    response = logs.filter_log_events(
        logGroupName=log_group,
        startTime=start_ms,
        endTime=end_ms,
        filterPattern="ERROR",
    )
    return [e["message"] for e in response["events"]]

import subprocess
import json
import os
from pathlib import Path


def terraform(cmd: list[str], cwd: str, env_vars: dict = None) -> subprocess.CompletedProcess:
    env = {**os.environ, **(env_vars or {})}
    result = subprocess.run(
        ["terraform"] + cmd,
        cwd=cwd,
        capture_output=True,
        text=True,
        env=env,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Terraform failed:\n{result.stderr}")
    return result


def deploy_infrastructure(
    tf_dir: str,
    variables: dict,
    auto_approve: bool = False,
) -> dict:
    # Write variables to a tfvars file
    tfvars_path = Path(tf_dir) / "override.auto.tfvars.json"
    tfvars_path.write_text(json.dumps(variables, indent=2))

    try:
        # Init (idempotent)
        terraform(["init", "-input=false"], cwd=tf_dir)

        # Plan
        terraform(
            ["plan", "-input=false", "-out=tfplan", "-detailed-exitcode"],
            cwd=tf_dir,
        )

        # Apply
        apply_args = ["apply", "-input=false", "tfplan"]
        if auto_approve:
            apply_args.insert(1, "-auto-approve")
        terraform(apply_args, cwd=tf_dir)

        # Read outputs
        result = terraform(["output", "-json"], cwd=tf_dir)
        outputs = json.loads(result.stdout)
        return {k: v["value"] for k, v in outputs.items()}

    finally:
        tfvars_path.unlink(missing_ok=True)   # Clean up


# --- Usage in CI pipeline ---
outputs = deploy_infrastructure(
    tf_dir="infra/environments/prod",
    variables={
        "region": "us-east-1",
        "instance_type": "t3.medium",
        "app_version": os.environ["APP_VERSION"],
    },
    auto_approve=True,
)
print("ALB DNS:", outputs["alb_dns_name"])
print("RDS Endpoint:", outputs["rds_endpoint"])

# Pass Terraform output to Ansible or other tools
with open("inventory.json", "w") as f:
    json.dump({"all": {"hosts": outputs["instance_ips"]}}, f)

#!/usr/bin/env python3
"""
health_check.py — checks service health and alerts on failure.
"""

import requests
import boto3
import sys
import logging
import time
from dataclasses import dataclass, field
from typing import List

logger = logging.getLogger(__name__)

@dataclass
class HealthCheckResult:
    name: str
    url: str
    healthy: bool
    status_code: int = 0
    response_time_ms: float = 0
    error: str = ""

@dataclass
class HealthCheckConfig:
    name: str
    url: str
    expected_status: int = 200
    timeout_s: int = 5
    expected_text: str = ""


def check_endpoint(cfg: HealthCheckConfig) -> HealthCheckResult:
    start = time.monotonic()
    try:
        resp = requests.get(cfg.url, timeout=cfg.timeout_s)
        elapsed = (time.monotonic() - start) * 1000

        healthy = resp.status_code == cfg.expected_status
        if cfg.expected_text and cfg.expected_text not in resp.text:
            healthy = False

        return HealthCheckResult(
            name=cfg.name,
            url=cfg.url,
            healthy=healthy,
            status_code=resp.status_code,
            response_time_ms=round(elapsed, 1),
        )
    except requests.RequestException as e:
        return HealthCheckResult(name=cfg.name, url=cfg.url, healthy=False, error=str(e))


def run_health_checks(checks: List[HealthCheckConfig]) -> List[HealthCheckResult]:
    results = [check_endpoint(c) for c in checks]
    failed = [r for r in results if not r.healthy]

    for r in results:
        status = "OK" if r.healthy else "FAIL"
        logger.info("[%s] %s — %dms", status, r.name, r.response_time_ms)

    if failed:
        # Publish to CloudWatch
        cw = boto3.client("cloudwatch", region_name="us-east-1")
        cw.put_metric_data(
            Namespace="HealthChecks",
            MetricData=[{
                "MetricName": "UnhealthyEndpoints",
                "Value": len(failed),
                "Unit": "Count",
            }],
        )

    return results


ENDPOINTS = [
    HealthCheckConfig("API Gateway", "https://api.myapp.com/health"),
    HealthCheckConfig("Auth Service", "https://auth.myapp.com/ping", expected_text="pong"),
    HealthCheckConfig("Admin Panel", "https://admin.myapp.com/", expected_status=200),
]

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    results = run_health_checks(ENDPOINTS)
    failed = [r for r in results if not r.healthy]
    sys.exit(1 if failed else 0)

import os
from pathlib import Path

# --- Environment variables ---
APP_ENV   = os.environ.get("APP_ENV", "dev")
DB_HOST   = os.environ["DB_HOST"]           # Raises KeyError if missing
PORT      = int(os.environ.get("PORT", "8080"))

# --- pathlib — object-oriented paths (preferred over os.path) ---
BASE_DIR  = Path(__file__).parent.parent    # Two levels up from this file
CONFIG    = BASE_DIR / "config" / f"{APP_ENV}.yaml"
LOG_DIR   = Path("/var/log/myapp")

# Create directories
LOG_DIR.mkdir(parents=True, exist_ok=True)

# File operations
if CONFIG.exists():
    content = CONFIG.read_text(encoding="utf-8")
    data = CONFIG.read_bytes()

CONFIG.write_text("key: value\n")

# Iterate over files matching a pattern
for log_file in LOG_DIR.glob("*.log"):
    size_mb = log_file.stat().st_size / 1_048_576
    if size_mb > 100:
        print(f"Large log: {log_file.name} ({size_mb:.1f} MB)")

# --- Portable path joining (works on Windows and Linux) ---
# os.path style (older)
artifact_dir = os.path.join(os.getcwd(), "build", "artifacts")

# pathlib style (modern — preferred)
artifact_dir = Path.cwd() / "build" / "artifacts"

# --- Useful path operations ---
p = Path("/home/ubuntu/app/config/prod.yaml")
print(p.parent)       # /home/ubuntu/app/config
print(p.stem)         # prod
print(p.suffix)       # .yaml
print(p.name)         # prod.yaml
print(p.parts)        # ('/', 'home', 'ubuntu', 'app', 'config', 'prod.yaml')

# --- Temp files (useful in CI) ---
import tempfile
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp:
    tmp.write(b'{"key": "value"}')
    tmp_path = Path(tmp.name)
# Process tmp_path, then clean up
tmp_path.unlink()

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time

# --- Session with retry logic (production-grade) ---
def make_session(retries: int = 3, backoff: float = 0.5) -> requests.Session:
    session = requests.Session()
    retry = Retry(
        total=retries,
        backoff_factor=backoff,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST", "PUT"],
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

session = make_session()

# --- GitHub API example (useful in CI/CD) ---
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
session.headers.update({
    "Authorization": f"Bearer {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json",
})

# Get PR info
resp = session.get(
    "https://api.github.com/repos/myorg/myapp/pulls/42",
    timeout=10,
)
resp.raise_for_status()
pr = resp.json()
print(f"PR #{pr['number']}: {pr['title']} — {pr['state']}")

# Trigger a workflow dispatch
resp = session.post(
    "https://api.github.com/repos/myorg/myapp/actions/workflows/deploy.yml/dispatches",
    json={"ref": "main", "inputs": {"environment": "prod", "version": "v1.2.3"}},
    timeout=10,
)
resp.raise_for_status()

# --- Rate limiting pattern ---
def api_call_with_rate_limit(url: str, calls_per_second: int = 5):
    interval = 1.0 / calls_per_second
    start = time.monotonic()
    resp = session.get(url, timeout=10)
    resp.raise_for_status()
    elapsed = time.monotonic() - start
    if elapsed < interval:
        time.sleep(interval - elapsed)
    return resp.json()

import boto3
import json
import logging
import time
import signal
import sys

logger = logging.getLogger(__name__)
sqs = boto3.client("sqs", region_name="us-east-1")
QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"

# --- Graceful shutdown ---
running = True
def handle_sigterm(sig, frame):
    global running
    logger.info("Shutdown signal received — draining current batch...")
    running = False

signal.signal(signal.SIGTERM, handle_sigterm)
signal.signal(signal.SIGINT, handle_sigterm)


def process_message(body: dict) -> bool:
    """Returns True if processed successfully."""
    event_type = body.get("event_type")
    if event_type == "deploy_request":
        logger.info("Processing deploy: %s", body.get("version"))
        # ... business logic ...
        return True
    logger.warning("Unknown event type: %s", event_type)
    return False


def poll_and_process():
    while running:
        response = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,     # Process up to 10 at once
            WaitTimeSeconds=20,         # Long polling — reduces empty receive calls
            VisibilityTimeout=300,      # Message hidden for 5 min while processing
            MessageAttributeNames=["All"],
        )

        messages = response.get("Messages", [])
        if not messages:
            continue

        for msg in messages:
            receipt_handle = msg["ReceiptHandle"]
            try:
                body = json.loads(msg["Body"])
                success = process_message(body)

                if success:
                    # Delete after successful processing
                    sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=receipt_handle)
                    logger.info("Message processed and deleted: %s", msg["MessageId"])
                else:
                    # Leave in queue to be retried (visibility timeout expires)
                    logger.warning("Processing failed — message will retry")

            except Exception:
                logger.exception("Unhandled error for message %s", msg["MessageId"])
                # DLQ handles it after max receives exceeded


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    poll_and_process()
    logger.info("Worker stopped cleanly")

import boto3
import json

iam = boto3.client("iam")

# --- Create a policy ---
policy_document = {
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Action": ["s3:GetObject", "s3:PutObject"],
        "Resource": "arn:aws:s3:::my-bucket/*",
    }],
}

policy_response = iam.create_policy(
    PolicyName="S3ReadWriteMyBucket",
    PolicyDocument=json.dumps(policy_document),
    Description="Allows R/W access to my-bucket",
)
policy_arn = policy_response["Policy"]["Arn"]

# --- Create a role (for EC2 to assume) ---
trust_policy = {
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {"Service": "ec2.amazonaws.com"},
        "Action": "sts:AssumeRole",
    }],
}

role = iam.create_role(
    RoleName="EC2-S3Access-Role",
    AssumeRolePolicyDocument=json.dumps(trust_policy),
    Description="Allows EC2 instances to access S3",
)
role_arn = role["Role"]["Arn"]

# Attach the policy to the role
iam.attach_role_policy(RoleName="EC2-S3Access-Role", PolicyArn=policy_arn)

# Create instance profile (needed to attach role to EC2)
iam.create_instance_profile(InstanceProfileName="EC2-S3Access-Profile")
iam.add_role_to_instance_profile(
    InstanceProfileName="EC2-S3Access-Profile",
    RoleName="EC2-S3Access-Role",
)

# --- Assume a role (cross-account access) ---
sts = boto3.client("sts")
assumed = sts.assume_role(
    RoleArn="arn:aws:iam::999999999999:role/CrossAccountRole",
    RoleSessionName="ci-pipeline-deploy",
    DurationSeconds=3600,
)
creds = assumed["Credentials"]

# Use the assumed credentials
cross_account_client = boto3.client(
    "ec2",
    region_name="us-east-1",
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"],
)

import boto3
import time
import sys

codepipeline = boto3.client("codepipeline", region_name="us-east-1")
codebuild    = boto3.client("codebuild", region_name="us-east-1")

PIPELINE_NAME = "myapp-prod-pipeline"

# --- Trigger a pipeline execution ---
response = codepipeline.start_pipeline_execution(name=PIPELINE_NAME)
execution_id = response["pipelineExecutionId"]
print(f"Started pipeline execution: {execution_id}")

# --- Poll until pipeline completes ---
def wait_for_pipeline(name: str, execution_id: str, timeout_s: int = 1800) -> str:
    start = time.monotonic()
    while time.monotonic() - start < timeout_s:
        response = codepipeline.get_pipeline_execution(
            pipelineName=name,
            pipelineExecutionId=execution_id,
        )
        status = response["pipelineExecution"]["status"]
        print(f"Pipeline status: {status}")

        if status in ("Succeeded", "Failed", "Stopped", "Superseded"):
            return status

        time.sleep(30)

    raise TimeoutError(f"Pipeline did not complete within {timeout_s}s")


final_status = wait_for_pipeline(PIPELINE_NAME, execution_id)
if final_status != "Succeeded":
    sys.exit(1)


# --- Trigger CodeBuild directly ---
build_response = codebuild.start_build(
    projectName="myapp-unit-tests",
    environmentVariablesOverride=[
        {"name": "APP_VERSION", "value": "v1.2.3", "type": "PLAINTEXT"},
        {"name": "DEPLOY_ENV", "value": "staging", "type": "PLAINTEXT"},
    ],
)
build_id = build_response["build"]["id"]

# Poll until build completes
while True:
    builds = codebuild.batch_get_builds(ids=[build_id])
    build = builds["builds"][0]
    phase = build["currentPhase"]
    status = build["buildStatus"]
    print(f"Build phase: {phase}, status: {status}")

    if status in ("SUCCEEDED", "FAILED", "STOPPED", "TIMED_OUT", "FAULT"):
        break
    time.sleep(15)

print("Build logs:", build["logs"]["deepLink"])
if status != "SUCCEEDED":
    sys.exit(1)