Python Interview Questions

my_list  = [1, 2, 2, 3]      # Ordered, mutable, allows duplicates
my_tuple = (1, 2, 2, 3)      # Ordered, immutable
my_set   = {1, 2, 3}         # Unordered, no duplicates
my_dict  = {"name": "Alice", "age": 30}  # Key-value pairs

# When to use each:
# list   → ordered sequence that changes (shopping cart)
# tuple  → fixed data (coordinates, RGB values)
# set    → membership testing, deduplication
# dict   → lookup by key (configuration, JSON-like data)

# Mutable — same object modified in place
lst = [1, 2, 3]
print(id(lst))   # e.g., 140234567890
lst.append(4)
print(id(lst))   # SAME id — modified in place

# Immutable — new object created
s = "hello"
print(id(s))     # e.g., 140234567000
s = s + " world"
print(id(s))     # DIFFERENT id — new string object

# Gotcha: mutable default arguments
def add_item(item, lst=[]):   # BAD — shared across calls!
    lst.append(item)
    return lst

def add_item(item, lst=None): # GOOD
    if lst is None:
        lst = []
    lst.append(item)
    return lst

import functools
import time

# Basic decorator
def timer(func):
    @functools.wraps(func)  # Preserves original function metadata
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        print(f"{func.__name__} took {end - start:.4f}s")
        return result
    return wrapper

@timer
def slow_function(n):
    """Simulate work."""
    return sum(range(n))

slow_function(1_000_000)
# Output: slow_function took 0.0234s

# Decorator with arguments
def retry(max_attempts=3, delay=1.0):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=5, delay=0.5)
def call_external_api():
    # Retries automatically on failure
    ...

# Regular function — loads all values into memory
def get_squares_list(n):
    return [x**2 for x in range(n)]   # Creates list of n items in memory

# Generator — yields one value at a time (lazy)
def get_squares_gen(n):
    for x in range(n):
        yield x**2

# Memory comparison
import sys
lst = get_squares_list(1_000_000)
gen = get_squares_gen(1_000_000)

print(sys.getsizeof(lst))  # ~8 MB
print(sys.getsizeof(gen))  # 112 bytes!

# Generator expressions (like list comprehensions but lazy)
squares = (x**2 for x in range(1000))   # Generator
squares_list = [x**2 for x in range(1000)]  # List

# Infinite generator (impossible with a list)
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

fib = fibonacci()
print([next(fib) for _ in range(10)])
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

def demo(*args, **kwargs):
    print(f"args:   {args}")    # tuple
    print(f"kwargs: {kwargs}")  # dict

demo(1, 2, 3, name="Alice", role="DevOps")
# args:   (1, 2, 3)
# kwargs: {'name': 'Alice', 'role': 'DevOps'}

# Passing to another function (forwarding)
def log_call(func):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__} with {args} {kwargs}")
        return func(*args, **kwargs)
    return wrapper

# Unpacking with * and **
numbers = [1, 2, 3]
print(*numbers)        # same as print(1, 2, 3)

config = {"sep": "-", "end": "\n"}
print(*numbers, **config)   # 1-2-3

import threading
import multiprocessing

# CPU-bound: multiprocessing wins (bypasses GIL — each process has its own GIL)
def cpu_work(n):
    return sum(i**2 for i in range(n))

# I/O-bound: threading works fine (GIL released during sleep/network/file I/O)
import requests
import concurrent.futures

urls = ["https://api.example.com/data"] * 10

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as pool:
    responses = list(pool.map(requests.get, urls))

# Python 3.13+ introduces a "free-threaded" mode (experimental)
# that removes the GIL — but this is still stabilizing.

# Current best practice:
# CPU-bound → multiprocessing.Pool
# I/O-bound → asyncio or threading
# Mixed     → asyncio + run_in_executor for CPU parts

import copy

original = {"name": "Alice", "scores": [90, 85, 78]}

# Shallow copy — copies the outer object, inner objects are shared
shallow = copy.copy(original)
shallow["name"] = "Bob"       # Does NOT affect original
shallow["scores"].append(100) # DOES affect original (same list object!)

print(original["name"])    # Alice (unchanged)
print(original["scores"])  # [90, 85, 78, 100] ← affected!

# Deep copy — recursively copies all nested objects
original = {"name": "Alice", "scores": [90, 85, 78]}
deep = copy.deepcopy(original)
deep["scores"].append(100)

print(original["scores"])  # [90, 85, 78] ← NOT affected

# Without context manager (fragile)
f = open("file.txt", "r")
try:
    data = f.read()
finally:
    f.close()   # Must manually close — could be missed

# With context manager (safe)
with open("file.txt", "r") as f:
    data = f.read()
# f.close() called automatically, even on exception

# Creating your own context manager (class-based)
class DatabaseConnection:
    def __enter__(self):
        self.conn = connect_to_db()
        return self.conn

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.conn.close()
        return False  # Don't suppress exceptions

with DatabaseConnection() as db:
    db.execute("SELECT 1")

# Creating context managers with @contextmanager (simpler)
from contextlib import contextmanager

@contextmanager
def managed_resource():
    resource = acquire_resource()
    try:
        yield resource          # Code inside 'with' block runs here
    finally:
        release_resource(resource)  # Always runs

with managed_resource() as r:
    r.do_something()

# List comprehension — creates a full list in memory
squares = [x**2 for x in range(10) if x % 2 == 0]
# [0, 4, 16, 36, 64]

# Dict comprehension
word_lengths = {word: len(word) for word in ["aws", "kubernetes", "docker"]}
# {'aws': 3, 'kubernetes': 10, 'docker': 6}

# Set comprehension
unique_lengths = {len(word) for word in ["cat", "bat", "door", "mat"]}
# {3, 4}

# Generator expression (lazy — use () not [])
squares_gen = (x**2 for x in range(10_000_000))
# No memory allocated yet — computed on demand

# Rule of thumb:
# List comprehension → need all results at once, or reuse multiple times
# Generator expression → single-pass consumption, large/infinite datasets

# Generator expressions compose well (pipeline)
lines = (line.strip() for line in open("huge.log"))
errors = (line for line in lines if "ERROR" in line)
first_10 = list(itertools.islice(errors, 10))
# Memory: only one line loaded at a time

from datetime import datetime

dt = datetime.now()
print(repr(dt))  # datetime.datetime(2024, 1, 15, 14, 30, 45, 123456)
print(str(dt))   # 2024-01-15 14:30:45.123456

class ServerConfig:
    def __init__(self, host, port):
        self.host = host
        self.port = port

    def __repr__(self):
        # Should be unambiguous — ideally recreatable
        return f"ServerConfig(host='{self.host}', port={self.port})"

    def __str__(self):
        # Human-friendly
        return f"{self.host}:{self.port}"

cfg = ServerConfig("localhost", 8080)
print(repr(cfg))  # ServerConfig(host='localhost', port=8080)
print(str(cfg))   # localhost:8080
print(cfg)        # localhost:8080  (uses __str__)

class Order:
    TAX_RATE = 0.08

    def __init__(self, amount):
        self.amount = amount

    def total(self):
        """Instance method — has access to self and the instance."""
        return self.amount * (1 + self.TAX_RATE)

    @classmethod
    def from_dict(cls, data: dict):
        """Class method — has access to cls (the class itself, not the instance).
        Useful as alternative constructors."""
        return cls(data["amount"])

    @staticmethod
    def is_valid_amount(amount: float) -> bool:
        """Static method — no access to self or cls.
        A utility function that logically belongs to the class."""
        return amount > 0

# Usage
order = Order(100)
order2 = Order.from_dict({"amount": 200})    # classmethod
print(Order.is_valid_amount(-5))             # False — staticmethod

import sys

x = [1, 2, 3]
print(sys.getrefcount(x))  # 2 (x + the getrefcount argument)

y = x          # Reference count becomes 3
del x          # Reference count becomes 2
y = None       # Reference count becomes 0 → object freed

# Circular reference — neither object reaches refcount 0
a = {}
b = {}
a["other"] = b
b["other"] = a

# Python's gc module detects and cleans these up
import gc
gc.collect()  # Manually trigger cycle detection

# __slots__ — prevents __dict__ creation, saves 40-50% memory per instance
class Point:
    __slots__ = ["x", "y"]  # No __dict__, no arbitrary attributes
    def __init__(self, x, y):
        self.x = x
        self.y = y

# For millions of objects, this matters:
import sys
class PointDict:
    def __init__(self, x, y): self.x = x; self.y = y
class PointSlots:
    __slots__ = ["x", "y"]
    def __init__(self, x, y): self.x = x; self.y = y

print(sys.getsizeof(PointDict(1,2)))   # ~48 bytes + dict overhead
print(sys.getsizeof(PointSlots(1,2)))  # ~56 bytes total (no dict)

import asyncio
import aiohttp   # pip install aiohttp

# WITHOUT asyncio — sequential (slow)
# Each request waits for the previous one to complete
# 10 requests × 200ms each = 2000ms total

# WITH asyncio — concurrent (fast)
# All requests fire "simultaneously" — 10 requests ≈ 200ms total

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)  # Run all concurrently
    return results

urls = ["https://api.example.com/user/1",
        "https://api.example.com/user/2",
        "https://api.example.com/user/3"]

results = asyncio.run(fetch_all(urls))

# asyncio vs threading vs multiprocessing
# I/O-bound, many connections  → asyncio (best — low overhead)
# I/O-bound, simple            → threading (easier)
# CPU-bound computation        → multiprocessing (bypasses GIL)

from typing import Optional, List, Dict, Union, Tuple
from dataclasses import dataclass

# Basic type hints
def greet(name: str, times: int = 1) -> str:
    return f"Hello, {name}! " * times

# Complex types
def process_users(users: List[Dict[str, str]]) -> Dict[str, int]:
    return {u["name"]: len(u["email"]) for u in users}

# Optional (can be None)
def find_user(user_id: int) -> Optional[str]:
    return db.get(user_id)   # Returns str or None

# Python 3.10+ — union with |
def parse_value(val: str | int | None) -> float:
    return float(val) if val is not None else 0.0

# dataclass — combines type hints with automatic __init__, __repr__
@dataclass
class ServiceConfig:
    host: str
    port: int
    timeout: float = 30.0
    tags: List[str] = None

    def __post_init__(self):
        if self.tags is None:
            self.tags = []

cfg = ServiceConfig(host="localhost", port=8080)
print(cfg)  # ServiceConfig(host='localhost', port=8080, timeout=30.0, tags=[])

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

print(a == b)  # True  — same values
print(a is b)  # False — different objects
print(a is c)  # True  — same object (c points to a)

# Integer caching — CPython caches small integers (-5 to 256)
x = 256
y = 256
print(x is y)  # True  — same cached object

x = 257
y = 257
print(x is y)  # False — outside cache range, different objects

# Correct usage of `is`
if response is None:    # ✅ Correct — None is a singleton
    handle_empty()

if response == None:    # ⚠️ Works but less Pythonic
    handle_empty()

import logging

def read_config(path: str) -> dict:
    try:
        # Code that may raise an exception
        with open(path) as f:
            return json.load(f)

    except FileNotFoundError:
        # Specific exception — file doesn't exist
        logging.error(f"Config file not found: {path}")
        return {}

    except json.JSONDecodeError as e:
        # Another specific exception — bad JSON
        logging.error(f"Invalid JSON in {path}: {e}")
        raise ValueError(f"Config file is malformed: {path}") from e

    except (PermissionError, OSError) as e:
        # Multiple exception types in one clause
        raise RuntimeError(f"Cannot read config: {e}") from e

    else:
        # Runs ONLY if no exception was raised in try
        logging.info(f"Config loaded successfully from {path}")

    finally:
        # ALWAYS runs — cleanup code (even if exception propagates)
        logging.debug("read_config complete")

# Custom exceptions
class InsufficientFundsError(ValueError):
    def __init__(self, balance: float, amount: float):
        self.balance = balance
        self.amount = amount
        super().__init__(f"Cannot withdraw {amount}: balance is {balance}")

def withdraw(balance: float, amount: float) -> float:
    if amount > balance:
        raise InsufficientFundsError(balance, amount)
    return balance - amount

from abc import ABC, abstractmethod

# ABSTRACTION — abstract base class defines interface, not implementation
class NotificationService(ABC):
    @abstractmethod
    def send(self, message: str, recipient: str) -> bool:
        """Must be implemented by subclasses."""
        ...

    def send_with_retry(self, message, recipient, max_retries=3):
        """Concrete method available to all subclasses."""
        for attempt in range(max_retries):
            if self.send(message, recipient):
                return True
        return False

# INHERITANCE — subclass extends/overrides base class
class EmailService(NotificationService):
    def __init__(self, smtp_host: str):
        self._smtp_host = smtp_host   # ENCAPSULATION — _ prefix = private by convention

    def send(self, message: str, recipient: str) -> bool:
        # Concrete implementation
        print(f"Email to {recipient}: {message}")
        return True

class SlackService(NotificationService):
    def __init__(self, webhook_url: str):
        self.__webhook = webhook_url   # __ name mangling = stronger encapsulation

    def send(self, message: str, recipient: str) -> bool:
        print(f"Slack to #{recipient}: {message}")
        return True

# POLYMORPHISM — same interface, different behaviour
services: list[NotificationService] = [
    EmailService("smtp.gmail.com"),
    SlackService("https://hooks.slack.com/..."),
]

alert = "Deployment complete!"
for service in services:
    service.send(alert, "devops-team")  # Calls the right implementation automatically

# Regular function
def double(x):
    return x * 2

# Equivalent lambda
double = lambda x: x * 2

# Lambdas shine as inline callbacks
numbers = [5, 2, 8, 1, 9, 3]
numbers.sort(key=lambda x: -x)   # Sort descending
# [9, 8, 5, 3, 2, 1]

users = [{"name": "Charlie", "age": 30}, {"name": "Alice", "age": 25}]
users.sort(key=lambda u: u["name"])   # Sort by name

# With filter() and map()
evens = list(filter(lambda x: x % 2 == 0, range(10)))  # [0, 2, 4, 6, 8]
squares = list(map(lambda x: x**2, [1, 2, 3, 4]))       # [1, 4, 9, 16]

class Temperature:
    def __init__(self, celsius: float):
        self._celsius = celsius   # Internal storage

    @property
    def celsius(self) -> float:
        """Getter."""
        return self._celsius

    @celsius.setter
    def celsius(self, value: float):
        """Setter — validates before storing."""
        if value < -273.15:
            raise ValueError(f"Temperature below absolute zero: {value}")
        self._celsius = value

    @celsius.deleter
    def celsius(self):
        del self._celsius

    @property
    def fahrenheit(self) -> float:
        """Computed property — no setter needed."""
        return self._celsius * 9/5 + 32

t = Temperature(100)
print(t.fahrenheit)   # 212.0 — looks like an attribute, runs code
t.celsius = -274      # Raises ValueError
t.celsius = 0
print(t.fahrenheit)   # 32.0

# Create virtual environment
python -m venv venv

# Activate (Linux/macOS)
source venv/bin/activate

# Activate (Windows)
venv\Scripts\activate

# Install packages (only in this venv, not globally)
pip install fastapi uvicorn

# Freeze dependencies
pip freeze > requirements.txt

# Deactivate
deactivate

# Modern tooling
# uv — ultra-fast alternative to pip + venv (recommended)
pip install uv
uv venv
uv pip install fastapi

# poetry — dependency management + virtual envs
poetry init
poetry add fastapi
poetry shell

# myapp/__init__.py — expose public API
from .config import settings
from .models.user import User

__version__ = "1.0.0"
__all__ = ["User", "settings"]  # Controls what 'from myapp import *' exports

# Usage
from myapp import User           # Works because of __init__.py
from myapp.models.order import Order  # Direct import also works

# Reading a file
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()            # Entire file as string
    # OR
    lines = f.readlines()         # List of lines
    # OR (memory efficient for large files)
    for line in f:                # Iterator — one line at a time
        process(line.strip())

# Writing a file
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
    f.writelines(["line1\n", "line2\n"])

# Appending
with open("log.txt", "a") as f:
    f.write("New log entry\n")

# Binary mode — for images, PDFs, executables
with open("image.png", "rb") as f:
    data = f.read()   # bytes, not str

with open("copy.png", "wb") as f:
    f.write(data)

import threading
import multiprocessing
import time

# I/O-bound task — threading is fine (GIL released during I/O)
def download_file(url):
    import requests
    return requests.get(url).content

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(download_file, urls))

# CPU-bound task — multiprocessing bypasses GIL
def compute_intensive(n):
    return sum(i**2 for i in range(n))

with multiprocessing.Pool(processes=4) as pool:
    results = pool.map(compute_intensive, [10**7] * 8)

# ProcessPoolExecutor — modern way to use multiprocessing
from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(compute_intensive, [10**7] * 8))

from collections import Counter, defaultdict, OrderedDict, deque

# Counter — count occurrences
words = ["aws", "docker", "aws", "kubernetes", "docker", "aws"]
count = Counter(words)
print(count)                    # Counter({'aws': 3, 'docker': 2, 'kubernetes': 1})
print(count.most_common(2))     # [('aws', 3), ('docker', 2)]
count.update(["aws", "helm"])   # Add more counts

# defaultdict — dict with default value for missing keys (no KeyError)
graph = defaultdict(list)
graph["A"].append("B")   # No need to check if "A" exists
graph["A"].append("C")
print(graph)  # defaultdict(<class 'list'>, {'A': ['B', 'C']})

# Versus regular dict:
try:
    regular = {}
    regular["A"].append("B")  # KeyError!
except KeyError:
    regular["A"] = ["B"]      # Must manually initialize

# deque — double-ended queue with O(1) append/pop from both ends
queue = deque(maxlen=5)      # Fixed-size sliding window
for i in range(10):
    queue.append(i)          # Auto-evicts oldest when full
print(queue)                 # deque([5, 6, 7, 8, 9], maxlen=5)

queue.appendleft(99)         # Add to left
queue.rotate(2)              # Rotate elements

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",
)