Answer:

• List ([]): Mutable (can be modified after creation).
• Tuple (()): Immutable (cannot be changed once created).
• DevOps Use Case: Use a list for dynamic data (e.g., list of IPs retrieved from AWS API that you need to filter). Use a tuple for fixed data (e.g., a hardcoded set of allowed HTTP response codes (200, 201, 204)).

my_list = [1, 2, 3]
my_tuple = (1, 2, 3)

my_list[0] = 10  # Works
# my_tuple[0] = 10 # Raises TypeError

Answer:

• ==: Checks for value equality (if the elements inside are identical).
• is: Checks for identity reference (if both variables point to the exact same object in memory).

a = [1, 2]
b = [1, 2]

print(a == b)  # True (same values)
print(a is b)  # False (different locations in memory)

Answer:

• String (str): Character sequences (e.g., IP addresses, log logs).
• Integer (int): Whole numbers (e.g., ports, retry counts).
• Float (float): Decimals (e.g., system load average).
• Boolean (bool): True or False.
• List (list): Ordered, mutable sequence of items.
• Tuple (tuple): Ordered, immutable sequence of items.
• Dictionary (dict): Key-value mappings (highly used for JSON configurations).
• Set (set): Unordered collection of unique items (useful for deduplicating data).

Answer: A Dictionary stores data in key-value pairs. Since most cloud infrastructure APIs (AWS, GCP, Kubernetes) return responses in JSON format, Python dictionaries are the native structure used to parse, manipulate, and generate configuration parameters.

server = {
    "name": "web-server",
    "ip": "10.0.0.1"
}
print(server["ip"])

Answer: You can iterate through the list using a standard for loop:

instances = ["web", "app", "db"]

for instance in instances:
    print(instance)

Answer:

• for: Iterates over a defined sequence (list, range, dictionary keys).
• while: Loops as long as a boolean condition remains True.

# For Loop
for i in range(5):
    print(i)

# While Loop (waiting for something to finish)
while True:
    # Do health check
    break

Answer: Functions are used to enforce DRY (Don't Repeat Yourself) principles. They bundle blocks of code to make scripts modular, testable, and reusable.

def backup():
    print("Backup started")

backup()  # Reusable call

Answer:

• Parameter: The variable placeholder declared in the function definition.
• Argument: The actual value passed to the function when calling it.

def deploy(env): # env is the parameter
    pass

deploy("prod")  # "prod" is the argument

Answer: Use the with open(...) context manager, which ensures that file descriptors are automatically closed even if the script raises an error.

with open("app.log", "r") as file:
    print(file.read())

Answer: The os module provides standard interfaces to interact with the underlying host OS. It is used for listing directory contents, checking file paths, executing system operations, and loading environment variables.

import os

print(os.getcwd())  # Get current working directory

Answer:

• os.path.exists(path): Returns True if either a directory or a file exists at the path.
• os.path.isfile(path): Returns True only if a regular file exists at that path (returns False if the path points to a directory).

import os

os.path.exists("/var/log")   # True (it is a directory)
os.path.isfile("/var/log")   # False (it is not a file)

Answer: os.listdir(path) returns a list containing the names of the entries (both files and subdirectories) inside the specified directory, excluding . and ...

import os
print(os.listdir("/tmp"))  # Output: ['file1.txt', 'cache_dir']

Answer:

• os.remove(path): Deletes a file. Raises an error if the path points to a directory.
• os.rmdir(path): Deletes an empty directory. Raises an error if the directory contains files or subdirectories.

Answer: Use os.makedirs(path, exist_ok=True). Setting exist_ok=True ensures the script does not raise an exception if the target directories already exist.

import os
os.makedirs("/opt/app/logs/prod", exist_ok=True)

Answer:

• os.environ["VAR"]: Directly accesses the environment variable dictionary. If "VAR" is missing, it raises a KeyError and crashes.
• os.getenv("VAR", default): Safely queries the variable. If missing, it returns None (or the specified default value) without crashing.

import os

# Safe check with fallback value
db_port = os.getenv("DB_PORT", "5432")

Answer:

• shutil.copy(src, dst): Copies the file contents and permissions. It does not preserve file metadata (like creation and modification times).
• shutil.copy2(src, dst): Copies the file contents, permissions, and preserves all metadata (timestamps).

Answer: shutil.move(src, dst) recursively moves a file or directory to another destination. It functions like the Linux mv command.

Answer: Use shutil.copytree(src, dst). It copies the entire directory tree from src to a new directory named dst (which must not exist beforehand unless dirs_exist_ok=True is set).

import shutil
shutil.copytree("/var/log/nginx", "/backup/nginx_logs", dirs_exist_ok=True)

Answer: Use shutil.rmtree(path). This deletes the directory and all of its contents (files, subdirectories, links) recursively. Use with caution!

import shutil
shutil.rmtree("/tmp/cache_folder")

Answer: shutil.disk_usage(path) returns a named tuple containing total, used, and free bytes on the given path.

import shutil

total, used, free = shutil.disk_usage("/")
print(f"Used Space: {used / (1024**3):.2f} GB")

Answer:

• datetime.now(): Returns the current local date and time based on the host system's timezone settings.
• datetime.utcnow(): Returns the current date and time in UTC (Coordinated Universal Time). In cloud/DevOps automation, always use UTC to avoid discrepancies between servers located in different regions.

Answer: Subtracting two datetime objects returns a timedelta object, which represents the duration between them.

from datetime import datetime

date1 = datetime(2026, 6, 1)
date2 = datetime(2026, 6, 5)

diff = date2 - date1
print(diff.days)  # Output: 4

Answer:

• subprocess.run(): Synchronous/blocking. It runs the command, waits for it to complete, and returns a CompletedProcess instance.
• subprocess.Popen(): Asynchronous/non-blocking. It spawns the command in a background process immediately and allows the Python script to continue running concurrently, interacting with standard streams in real time.

Answer: If the command string includes untrusted user inputs, shell=True exposes the system to shell injection vulnerabilities (where an attacker inserts extra characters like ; rm -rf / to execute arbitrary commands). If possible, pass arguments as a list with shell=False.

Answer: Pass stdout=subprocess.PIPE and stderr=subprocess.PIPE. You can then check the exit status using returncode.

import subprocess

result = subprocess.run("echo 'hello'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print("Exit Status:", result.returncode)
print("Output:", result.stdout.decode().strip())

Answer: sys.argv is a list containing the command-line arguments passed to the active Python script.

• sys.argv[0]: The name of the script itself.
• sys.argv[1]: The first argument passed after the script name.

# Save as test.py
import sys

print(f"Script Name: {sys.argv[0]}")
if len(sys.argv) > 1:
    print(f"First Arg: {sys.argv[1]}")

Running python3 test.py prod will output:

Script Name: test.py
First Arg: prod

Answer: sys.exit(code) raises the SystemExit exception to terminate the script execution. Passing 0 indicates a successful termination, while any non-zero value (typically 1) signals a failure to the calling shell or CI/CD workflow runner.

Answer: Boto3 looks for credentials in the following order:

1. Explicit parameters passed to the boto3.client() constructor.
2. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN).
3. Shared credential files (~/.aws/credentials and ~/.aws/config).
4. IAM Role credentials assigned to the EC2 instance or ECS task metadata service.

Answer:

• Client: Low-level interface. Maps directly to the underlying AWS HTTP query APIs. Always returns structured dictionary outputs.
• Resource: High-level, object-oriented interface. Wraps resources as Python objects with attributes and actions (e.g., bucket.objects.all()). Not all AWS services are supported by resources.

Answer: Use the filter() method with appropriate criteria:

import boto3

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

# Filter instances that are stopped and have the tag Name=TestServer
instances = ec2.instances.filter(
    Filters=[
        {"Name": "instance-state-name", "Values": ["stopped"]},
        {"Name": "tag:Name", "Values": ["TestServer"]}
    ]
)

# Start them
for instance in instances:
    print(f"Starting instance: {instance.id}")
    instance.start()

Answer: AWS APIs paginate responses when resource counts are large. You can handle pagination using a Boto3 Paginated Query:

import boto3

client = boto3.client("s3")
paginator = client.get_paginator("list_objects_v2")

# Create a page iterator
page_iterator = paginator.paginate(Bucket="my-large-s3-bucket")

for page in page_iterator:
    if "Contents" in page:
        for obj in page["Contents"]:
            print(obj["Key"])

Answer: Call AWS Security Token Service (STS) assume_role to fetch temporary credentials, then initialize a new Boto3 session:

import boto3

sts = boto3.client("sts")

assumed_role = sts.assume_role(
    RoleArn="arn:aws:iam::123456789012:role/TargetDevOpsRole",
    RoleSessionName="AssumedRoleSession"
)

# Extract credentials
creds = assumed_role["Credentials"]

# Create session with temporary credentials
session = boto3.Session(
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"]
)

# Use resources within the context of the assumed role
s3 = session.resource("s3")

Answer:

import os
import shutil
import boto3
from datetime import datetime

# Configurations
THRESHOLD = 90.0
LOG_DIR = "/var/log/app"
BACKUP_FILE = f"/tmp/logs_backup_{datetime.now().strftime('%Y%m%d')}"
S3_BUCKET = "devops-incident-backups"

# 1. Check disk usage
total, used, free = shutil.disk_usage("/")
percent_used = (used / total) * 100

if percent_used > THRESHOLD:
    print(f"Disk alert: {percent_used:.2f}% used! Commencing cleanup...")
    
    # 2. Compress log directory
    if os.path.exists(LOG_DIR):
        archive_path = shutil.make_archive(BACKUP_FILE, 'gztar', LOG_DIR)
        print(f"Logs compressed to: {archive_path}")
        
        # 3. Upload to S3
        s3 = boto3.client("s3")
        try:
            s3.upload_file(archive_path, S3_BUCKET, os.path.basename(archive_path))
            print("Compressed archive uploaded to S3 successfully.")
            
            # 4. Cleanup old logs (Optional / Safe truncation)
            for file in os.listdir(LOG_DIR):
                file_path = os.path.join(LOG_DIR, file)
                if os.path.isfile(file_path):
                    open(file_path, 'w').close()  # Truncate content keeping file descriptors open
            
            # Remove temporary archive
            os.remove(archive_path)
        except Exception as e:
            print(f"Backup upload failed: {e}")

Answer:

import boto3

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

# Find stopped instances
stopped_instances = ec2.instances.filter(
    Filters=[{"Name": "instance-state-name", "Values": ["stopped"]}]
)

instance_ids = [instance.id for instance in stopped_instances]

if instance_ids:
    print(f"Found stopped instances: {instance_ids}. Starting them now...")
    # Start the instances
    ec2.instances.filter(InstanceIds=instance_ids).start()
else:
    print("No stopped instances found.")

Answer:

import subprocess
import os

tf_dir = "/home/dev/DevOps/DevOps/12-Python-for-DevOps/terra-automate/terraform"
log_path = "/var/log/terraform_deploy.log"

def run_terraform():
    print("Initiating Terraform deployment...")
    try:
        # Run apply command
        process = subprocess.run(
            f"terraform -chdir={tf_dir} apply -auto-approve",
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=True
        )
        
        # Log success
        with open(log_path, "a") as log:
            log.write(f"\n--- SUCCESS: {os.uname()[1]} ---\n")
            log.write(process.stdout.decode())
        print("Terraform applied successfully.")
        
    except subprocess.CalledProcessError as e:
        # Log failure details
        with open(log_path, "a") as log:
            log.write(f"\n--- FAILURE: {os.uname()[1]} ---\n")
            log.write(e.stderr.decode())
        print(f"Terraform deployment failed. Error logged to {log_path}.")

run_terraform()

Answer:

log_file = "app.log"
report_file = "error_report.md"

errors = []

# 1. Read and find errors
with open(log_file, "r") as f:
    for line in f:
        if "ERROR" in line:
            errors.append(line.strip())

# 2. Write Markdown report
with open(report_file, "w") as f:
    f.write("# 📋 DevOps Log Incident Report\n\n")
    f.write(f"**Source Log File:** `{log_file}`  \n")
    f.write(f"**Total Incidents Found:** {len(errors)}  \n\n")
    f.write("## ⚠️ Identified Exceptions:\n")
    if errors:
        for idx, error in enumerate(errors, 1):
            f.write(f"{idx}. `{error}`\n")
    else:
        f.write("*No errors detected in the current log range.*\n")

print(f"Report compiled successfully at {report_file}")

Answer:

import sys

def deploy(environment):
    print(f"Initializing deployment configuration for: {environment.upper()}")
    
    if environment == "dev":
        print("Running: terraform apply -var-file=dev.tfvars")
    elif environment == "stage":
        print("Running: terraform apply -var-file=stage.tfvars")
    elif environment == "prod":
        print("CRITICAL: Executing Production deployment pipeline...")
        print("Running: terraform apply -var-file=prod.tfvars")
    else:
        print(f"Error: Unknown environment '{environment}'!")
        sys.exit(1)

if __name__ == "__main__":
    # Ensure environment argument is provided
    if len(sys.argv) < 2:
        print("Usage: python deploy.py <dev|stage|prod>")
        sys.exit(1)
        
    target_env = sys.argv[1].lower()
    deploy(target_env)

Answer:

import os
import shutil
import boto3
from datetime import datetime, timezone, timedelta

# Config
BACKUP_DIR = "/opt/app/data"
S3_BUCKET = "my-company-backups"
date_suffix = datetime.now().strftime("%Y_%m_%d")
archive_name = f"/tmp/backup_{date_suffix}"

# 1. Create timestamped backup
archive_file = shutil.make_archive(archive_name, 'zip', BACKUP_DIR)
s3_key = os.path.basename(archive_file)

# Initialize AWS clients
s3 = boto3.client("s3")
s3_res = boto3.resource("s3")

try:
    # 2. Upload to S3
    s3.upload_file(archive_file, S3_BUCKET, s3_key)
    print(f"Uploaded {s3_key} to S3 bucket.")
    os.remove(archive_file) # Remove local copy
    
    # 3. Purge objects older than 30 days
    bucket = s3_res.Bucket(S3_BUCKET)
    now_utc = datetime.now(timezone.utc)
    expiration_limit = now_utc - timedelta(days=30)
    
    for obj in bucket.objects.all():
        if obj.last_modified < expiration_limit:
            print(f"Purging expired backup object: {obj.key} (Modified: {obj.last_modified})")
            obj.delete()
            
except Exception as e:
    print(f"Backup operation failed: {e}")