Python Practice Questions: 25 Real Problems & Solutions

def clean_numeric_value(val, default: float = 0.0) -> float:
    """
    Defensively coerces raw inputs into float values, stripping currency
    symbols and whitespace. Returns default value on invalid inputs.
    """
    if val is None:
        return default
    
    # If already an integer or float, return as float directly
    if isinstance(val, (int, float)):
        return float(val)
    
    # Clean string representation
    cleaned = str(val).strip().replace("$", "").replace(",", "")
    if not cleaned:
        return default
    
    try:
        return float(cleaned)
    except (ValueError, TypeError):
        return default
 
# Test Cases
print(clean_numeric_value("$1,450.50"))  # Expected: 1450.5
print(clean_numeric_value(None))          # Expected: 0.0
print(clean_numeric_value("  "))          # Expected: 0.0
print(clean_numeric_value("invalid", -1.0)) # Expected: -1.0
print(clean_numeric_value(42))            # Expected: 42.0

def is_clean_palindrome(text: str) -> bool:
    """
    Validates if a string is a palindrome ignoring non-alphanumeric characters and casing.
    """
    if not isinstance(text, str):
        return False
    
    # Filter only alphanumeric characters and convert to lowercase
    cleaned = [char.lower() for char in text if char.isalnum()]
    
    # Compare sequence to its exact reverse using two-pointer or slice
    return cleaned == cleaned[::-1]
 
# Test Cases
print(is_clean_palindrome("A man, a plan, a canal: Panama!")) # True
print(is_clean_palindrome("Was it a car or a cat I saw?"))     # True
print(is_clean_palindrome("Topfolio Analytics"))              # False
print(is_clean_palindrome(""))                                # True (empty is valid)

import re
from typing import Optional
 
def normalize_us_phone(raw_phone: str) -> Optional[str]:
    """
    Extracts digits from raw input and formats standard US phone numbers.
    """
    if not raw_phone:
        return None
    
    # Extract only numeric digits using regex
    digits = re.sub(r"\D", "", str(raw_phone))
    
    # Handle optional leading US country code '1'
    if len(digits) == 11 and digits.startswith("1"):
        digits = digits[1:]
        
    if len(digits) != 10:
        return None
        
    area_code, prefix, line = digits[:3], digits[3:6], digits[6:]
    return f"({area_code}) {prefix}-{line}"
 
# Test Cases
print(normalize_us_phone("+1 (555) 234-5678")) # (555) 234-5678
print(normalize_us_phone("555.234.5678"))      # (555) 234-5678
print(normalize_us_phone("5552345678"))        # (555) 234-5678
print(normalize_us_phone("123-45"))            # None (invalid length)

from decimal import Decimal, ROUND_HALF_UP
 
def calculate_invoice_balance(items: list[tuple[str, str]], tax_rate: str) -> Decimal:
    """
    Calculates line-item invoice sum with exact tax rounding using Decimal.
    items: list of (item_name, price_as_string)
    tax_rate: tax percentage as string (e.g. '0.0825' for 8.25%)
    """
    subtotal = Decimal("0.00")
    for _, price_str in items:
        subtotal += Decimal(price_str)
        
    tax = subtotal * Decimal(tax_rate)
    total = subtotal + tax
    
    # Quantize to exactly 2 decimal places using standard commercial rounding
    cents = Decimal("0.01")
    return total.quantize(cents, rounding=ROUND_HALF_UP)
 
# Test Cases
invoice_items = [
    ("Database Server License", "1299.99"),
    ("Cloud Storage 1TB", "14.20"),
    ("Support Retainer", "499.50")
]
tax_pct = "0.0825" # 8.25%
 
print(calculate_invoice_balance(invoice_items, tax_pct)) # Expected: 1963.38

def print_department_summary(departments: list[dict]):
    """
    Prints a formatted summary table using advanced f-string specification.
    """
    print(f"{'Department':<20} | {'Headcount':>10} | {'Annual Budget':>15}")
    print("-" * 52)
    
    total_budget = 0
    total_headcount = 0
    
    for dept in departments:
        name = dept["name"]
        headcount = dept["headcount"]
        budget = dept["budget"]
        
        total_headcount += headcount
        total_budget += budget
        
        # Format: 20 chars left-aligned, 10 chars right-aligned, 15 chars right-aligned with commas
        print(f"{name:<20} | {headcount:>10,d} | ${budget:>14,.2f}")
        
    print("-" * 52)
    print(f"{'TOTAL':<20} | {total_headcount:>10,d} | ${total_budget:>14,.2f}")
 
# Test Case Data
depts = [
    {"name": "Data Analytics", "headcount": 14, "budget": 1420500.00},
    {"name": "Engineering", "headcount": 48, "budget": 5890000.50},
    {"name": "Marketing", "headcount": 8, "budget": 750200.25}
]
 
print_department_summary(depts)

def calculate_moving_averages(data: list[float], window_size: int) -> list[float]:
    """
    Computes rolling moving average using list slicing.
    """
    if window_size <= 0 or len(data) < window_size:
        return []
    
    averages = []
    # Slide across valid starting indices
    for i in range(len(data) - window_size + 1):
        window = data[i : i + window_size]
        avg = sum(window) / window_size
        averages.append(round(avg, 2))
        
    return averages
 
# Test Cases
daily_sales = [100.0, 120.0, 110.0, 150.0, 180.0, 200.0]
print(calculate_moving_averages(daily_sales, 3)) # 3-day rolling

from collections import defaultdict
 
def invert_dictionary(mapping: dict[str, str]) -> dict[str, list[str]]:
    """
    Inverts a dictionary, grouping multiple keys that share the same value.
    """
    inverted = defaultdict(list)
    for key, value in mapping.items():
        inverted[value].append(key)
        
    # Return as standard dict sorted for deterministic output
    return dict(sorted(inverted.items()))
 
# Test Cases
user_depts = {
    "alice": "Analytics",
    "bob": "Engineering",
    "charlie": "Analytics",
    "david": "Marketing",
    "elena": "Engineering"
}
 
print(invert_dictionary(user_depts))

def analyze_monthly_cohorts(january: set, february: set) -> dict:
    """
    Analyzes user retention using set operations.
    """
    retained = january.intersection(february)   # january & february
    churned = january.difference(february)       # january - february
    new_users = february.difference(january)     # february - january
    
    retention_rate = (len(retained) / len(january) * 100.0) if january else 0.0
    
    return {
        "retained_count": len(retained),
        "churned_count": len(churned),
        "new_users_count": len(new_users),
        "retention_rate_pct": round(retention_rate, 2)
    }
 
# Test Cases
jan_cohort = {"usr_101", "usr_102", "usr_103", "usr_104", "usr_105"}
feb_cohort = {"usr_103", "usr_104", "usr_105", "usr_106", "usr_107"}
 
metrics = analyze_monthly_cohorts(jan_cohort, feb_cohort)
for metric, val in metrics.items():
    print(f"{metric}: {val}")

def filter_slow_api_calls(logs: list[tuple], max_latency_ms: float) -> list[dict]:
    """
    Unpacks log tuples and filters requests by status code and latency threshold.
    """
    slow_requests = []
    
    for timestamp, method, endpoint, status, latency in logs:
        if 200 <= status <= 299 and latency > max_latency_ms:
            slow_requests.append({
                "timestamp": timestamp,
                "endpoint": f"{method} {endpoint}",
                "status": status,
                "latency_ms": latency
            })
            
    return slow_requests
 
# Test Cases
server_logs = [
    ("2024-04-10T10:00:01Z", "GET", "/api/v1/products", 200, 45.2),
    ("2024-04-10T10:00:03Z", "POST", "/api/v1/checkout", 201, 310.8),
    ("2024-04-10T10:00:05Z", "GET", "/api/v1/users/me", 500, 450.0),
    ("2024-04-10T10:00:08Z", "GET", "/api/v1/search", 200, 220.4)
]
 
print(filter_slow_api_calls(server_logs, 200.0))

import re
from collections import Counter
 
def get_top_k_words(text: str, k: int = 3) -> list[tuple[str, int]]:
    """
    Tokenizes text, strips punctuation, and extracts top K word frequencies.
    """
    if not text:
        return []
    
    # Extract lowercase words using regex word boundaries
    words = re.findall(r"\b\w+\b", text.lower())
    
    # Count occurrences using collections.Counter
    counts = Counter(words)
    
    # Return top K most common items
    return counts.most_common(k)
 
# Test Cases
paragraph = """
Data analysts use SQL and Python. Python is great for data cleaning,
while SQL is great for database aggregation. Data teams need both Python and SQL.
"""
 
print(get_top_k_words(paragraph, 3))

def flatten_and_filter_sales(store_matrix: list[list[float]], threshold: float) -> list[float]:
    """
    Flattens a 2D matrix into 1D and filters values >= threshold using comprehension.
    """
    return [
        revenue
        for store in store_matrix
        for revenue in store
        if revenue >= threshold
    ]
 
# Test Cases
quarterly_sales_matrix = [
    [120000.0, 95000.0, 140000.0, 180000.0],  # Store 1
    [85000.0, 78000.0, 90000.0, 88000.0],     # Store 2 (Underperforming)
    [210000.0, 240000.0, 195000.0, 280000.0]  # Store 3
]
 
print(flatten_and_filter_sales(quarterly_sales_matrix, 100000.0))

from typing import Callable
 
def transform_dataset(data: list, *transformers: Callable, **options) -> list:
    """
    Applies a chain of transformation functions to a dataset with optional configs.
    """
    result = list(data)
    
    # Apply each transformation function in sequence
    for fn in transformers:
        result = [fn(x) for x in result]
        
    # Apply optional keyword configurations
    if options.get("drop_none", False):
        result = [x for x in result if x is not None]
        
    round_digits = options.get("round_digits")
    if round_digits is not None:
        result = [round(x, round_digits) if isinstance(x, (int, float)) else x for x in result]
        
    return result
 
# Sample Transformer Functions
def to_celsius(f_temp):
    return (f_temp - 32) * 5 / 9 if f_temp is not None else None
 
def add_sensor_bias(temp):
    return temp + 0.45 if temp is not None else None
 
# Test Case
raw_temperatures = [72.5, 68.0, None, 85.2, 90.0]
 
clean_temps = transform_dataset(
    raw_temperatures,
    to_celsius,
    add_sensor_bias,
    drop_none=True,
    round_digits=1
)
 
print(clean_temps) # Expected Celsius conversions

def sort_employees_by_dept_and_salary(employees: list[dict]) -> list[dict]:
    """
    Sorts employees by department ASC, then salary DESC using a lambda tuple key.
    """
    # Key returns a tuple: (department_string, -salary_float)
    return sorted(
        employees,
        key=lambda emp: (emp["department"], -emp["salary"])
    )
 
# Test Cases
staff = [
    {"name": "Alice", "department": "Engineering", "salary": 115000},
    {"name": "Bob", "department": "Analytics", "salary": 92000},
    {"name": "Charlie", "department": "Engineering", "salary": 140000},
    {"name": "David", "department": "Analytics", "salary": 105000},
    {"name": "Elena", "department": "Marketing", "salary": 88000}
]
 
sorted_staff = sort_employees_by_dept_and_salary(staff)
for s in sorted_staff:
    print(f"{s['department']:<12} | ${s['salary']:>7} | {s['name']}")

from typing import Generator
import io
 
def stream_error_logs(log_stream: io.StringIO) -> Generator[str, None, None]:
    """
    Yields error log lines lazily without loading the entire stream into memory.
    """
    for line in log_stream:
        # Check if line contains common HTTP client/server error patterns
        # Example format: '2024-04-10 12:00:00 [404] /api/v1/missing'
        parts = line.strip().split()
        for token in parts:
            if token.startswith("[") and token.endswith("]") and token[1:-1].isdigit():
                status_code = int(token[1:-1])
                if 400 <= status_code <= 599:
                    yield line.strip()
                    break
 
# Simulated Log File Stream
mock_logs = io.StringIO("""
2024-04-10 10:00:01 [200] /api/v1/health
2024-04-10 10:00:05 [404] /api/v1/users/9999
2024-04-10 10:00:10 [200] /api/v1/products
2024-04-10 10:00:12 [500] /api/v1/payment/checkout
""")
 
# Consumer consumes one line at a time
for error_line in stream_error_logs(mock_logs):
    print(f"ALERT: {error_line}")

def flatten_taxonomy(tree: dict, parent_key: str = "", sep: str = " > ") -> dict[str, int]:
    """
    Recursively flattens nested dictionary paths into breadcrumb keys.
    """
    flat_dict = {}
    
    for key, value in tree.items():
        # Construct path
        new_key = f"{parent_key}{sep}{key}" if parent_key else key
        
        if isinstance(value, dict):
            # Recurse deeper into subtree
            flat_dict.update(flatten_taxonomy(value, new_key, sep=sep))
        else:
            # Base case: leaf node
            flat_dict[new_key] = value
            
    return flat_dict
 
# Test Cases
catalog_tree = {
    "Electronics": {
        "Audio": {
            "Headphones": 45,
            "Speakers": 12
        },
        "Displays": 30
    },
    "Home Office": {
        "Desks": 18
    }
}
 
flattened = flatten_taxonomy(catalog_tree)
for path, count in flattened.items():
    print(f"{path}: {count} items")

import csv
import io
 
def parse_csv_safely(csv_text: str) -> tuple[list[dict], list[dict]]:
    """
    Parses CSV text safely, returning (valid_records, rejected_records).
    """
    valid_rows = []
    rejected_rows = []
    
    # Use StringIO as mock file buffer
    file_obj = io.StringIO(csv_text.strip())
    reader = csv.DictReader(file_obj)
    
    for row_idx, row in enumerate(reader, start=2): # Start at 2 accounting for header
        try:
            order_id = int(row["order_id"])
            customer = row["customer_name"].strip()
            amount = float(row["total_amount"])
            
            if not customer:
                raise ValueError("Customer name cannot be blank")
                
            valid_rows.append({
                "order_id": order_id,
                "customer": customer,
                "total_amount": amount
            })
        except (ValueError, KeyError, TypeError) as err:
            rejected_rows.append({
                "row_number": row_idx,
                "raw_data": row,
                "error": str(err)
            })
            
    return valid_rows, rejected_rows
 
# Test Case
sample_csv = """order_id,customer_name,total_amount
1001,Elena Rostova,240.50
1002,,120.00
1003,Marcus Vance,invalid_price
1004,Chloe Bennett,89.99
"""
 
valids, rejects = parse_csv_safely(sample_csv)
print(f"Successfully Parsed ({len(valids)}):", valids)
print(f"Rejected Rows ({len(rejects)}):", rejects)

from datetime import datetime, timezone
 
class DataQualityValidationError(Exception):
    """Raised when incoming transactional data fails business validation checks."""
    pass
 
def validate_transaction_batch(transactions: list[dict]):
    """
    Validates transactional records against business integrity rules.
    """
    now = datetime.now(timezone.utc)
    
    for txn in transactions:
        txn_id = txn.get("id")
        amount = txn.get("amount", 0)
        txn_time = txn.get("timestamp")
        
        if amount < 0:
            raise DataQualityValidationError(
                f"Validation Failure: Transaction {txn_id} has negative amount ({amount})."
            )
            
        if txn_time and txn_time > now:
            raise DataQualityValidationError(
                f"Validation Failure: Transaction {txn_id} has future timestamp ({txn_time})."
            )
            
    return True
 
# Test Cases
clean_batch = [
    {"id": "T1", "amount": 100.0, "timestamp": datetime(2024, 1, 1, tzinfo=timezone.utc)},
    {"id": "T2", "amount": 50.0, "timestamp": datetime(2024, 1, 2, tzinfo=timezone.utc)}
]
print("Clean Batch Validated:", validate_transaction_batch(clean_batch))
 
dirty_batch = [
    {"id": "T3", "amount": -15.0, "timestamp": datetime(2024, 1, 1, tzinfo=timezone.utc)}
]
try:
    validate_transaction_batch(dirty_batch)
except DataQualityValidationError as ex:
    print(f"Caught Expected Exception: {ex}")

def safe_nested_get(data: dict, path: str, default=None):
    """
    Traverses a nested dictionary using a dot-separated key path.
    """
    if not isinstance(data, dict):
        return default
        
    keys = path.split(".")
    current = data
    
    for key in keys:
        if isinstance(current, dict) and key in current:
            current = current[key]
        else:
            return default
            
    return current if current is not None else default
 
# Test Cases
api_payload = {
    "status": "success",
    "data": {
        "user": {
            "id": 402,
            "profile": {
                "name": "Sarah Connor",
                "address": {
                    "city": "Los Angeles",
                    "zip_code": "90001"
                }
            }
        }
    }
}
 
print(safe_nested_get(api_payload, "data.user.profile.address.zip_code")) # 90001
print(safe_nested_get(api_payload, "data.user.profile.phone", "N/A"))     # N/A
print(safe_nested_get(api_payload, "data.order.items.0.price", 0.0))      # 0.0

import time
from functools import wraps
 
def retry(max_attempts: int = 3, backoff_factor: float = 1.5):
    """
    Decorator that retries a function with exponential backoff upon exception.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            delay = 0.1  # Initial sleep time in seconds
            
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as err:
                    attempts += 1
                    if attempts >= max_attempts:
                        raise err
                    time.sleep(delay)
                    delay *= backoff_factor
        return wrapper
    return decorator
 
# Test Case: Function that fails twice before succeeding
call_count = 0
 
@retry(max_attempts=3, backoff_factor=2.0)
def fetch_external_analytics_data():
    global call_count
    call_count += 1
    if call_count < 3:
        raise ConnectionResetError(f"Simulated network timeout (Attempt {call_count})")
    return {"status": "200 OK", "records_synced": 450}
 
result = fetch_external_analytics_data()
print("Success after retries:", result)

import logging
import io
 
def setup_pipeline_logger() -> tuple[logging.Logger, io.StringIO]:
    """
    Configures a structured pipeline logger writing to an in-memory stream.
    """
    log_stream = io.StringIO()
    logger = logging.getLogger("ETLPipeline")
    logger.setLevel(logging.INFO)
    
    # Prevent duplicate handlers if re-executed
    if not logger.handlers:
        handler = logging.StreamHandler(log_stream)
        formatter = logging.Formatter(
            fmt="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        
    return logger, log_stream
 
# Test Run
logger, stream = setup_pipeline_logger()
logger.info("Extract step initiated: querying PostgreSQL warehouse.")
logger.warning("Query latency elevated: took 4.2s to fetch 50,000 records.")
logger.error("Load failure: duplicate primary key detected on customer_id 8942.")
 
print(stream.getvalue().strip())

import pandas as pd
 
def filter_target_transactions(df: pd.DataFrame) -> pd.DataFrame:
    """
    Applies multi-condition boolean masking to an orders DataFrame.
    """
    # Enforce boolean conditions with parentheses and bitwise operators
    mask = (
        df["state"].isin(["CA", "NY"]) &
        (df["total_amount"] > 150.0) &
        (df["status"] == "completed")
    )
    
    return df[mask].sort_values(by="total_amount", ascending=False)
 
# Test DataFrame
orders_df = pd.DataFrame({
    "order_id": [101, 102, 103, 104, 105],
    "customer": ["Elena", "Marcus", "Chloe", "David", "Arthur"],
    "state": ["CA", "TX", "NY", "CA", "NY"],
    "total_amount": [240.50, 420.00, 180.00, 95.00, 310.00],
    "status": ["completed", "completed", "completed", "completed", "refunded"]
})
 
print(filter_target_transactions(orders_df))

import pandas as pd
 
def clean_missing_demographics(df: pd.DataFrame) -> pd.DataFrame:
    """
    Cleans demographic DataFrame using targeted drop and group-median imputation.
    """
    # Step 1: Drop records missing primary key
    df_clean = df.dropna(subset=["customer_id"]).copy()
    
    # Step 2: Impute salary with department-specific median
    dept_medians = df_clean.groupby("department")["salary"].transform("median")
    df_clean["salary"] = df_clean["salary"].fillna(dept_medians)
    
    return df_clean
 
# Test DataFrame
data = {
    "customer_id": [101, 102, None, 104, 105, 106],
    "department": ["Eng", "Eng", "Analytics", "Analytics", "Analytics", "Eng"],
    "salary": [120000.0, None, 95000.0, None, 90000.0, 140000.0]
}
df_raw = pd.DataFrame(data)
 
print(clean_missing_demographics(df_raw))

import pandas as pd
 
def build_sales_pivot_table(df: pd.DataFrame) -> pd.DataFrame:
    """
    Constructs a cross-tabulated sales revenue matrix with row and column totals.
    """
    pivot = pd.pivot_table(
        df,
        values="revenue",
        index="region",
        columns="product_category",
        aggfunc="sum",
        fill_value=0.0,
        margins=True,
        margins_name="Total Revenue"
    )
    return pivot.round(2)
 
# Test DataFrame
sales_data = pd.DataFrame({
    "region": ["North", "North", "South", "South", "East", "North"],
    "product_category": ["Audio", "Displays", "Audio", "Furniture", "Displays", "Audio"],
    "revenue": [450.0, 890.0, 310.0, 750.0, 620.0, 280.0]
})
 
print(build_sales_pivot_table(sales_data))

import pandas as pd
 
def audit_customer_order_relationship(
    customers: pd.DataFrame, 
    orders: pd.DataFrame
) -> pd.DataFrame:
    """
    Performs full outer merge with indicator flag to classify relationship status.
    """
    merged = pd.merge(
        customers,
        orders,
        on="customer_id",
        how="outer",
        indicator=True
    )
    
    # Map the default indicator values to descriptive business terms
    label_map = {
        "both": "Verified Customer Order",
        "left_only": "Customer with Zero Orders",
        "right_only": "Orphaned Order (Missing Customer)"
    }
    merged["audit_status"] = merged["_merge"].map(label_map)
    
    return merged[["customer_id", "customer_name", "order_id", "audit_status"]]
 
# Test DataFrames
customers_data = pd.DataFrame({
    "customer_id": [1, 2, 3],
    "customer_name": ["Alice", "Bob", "Charlie"]
})
 
orders_data = pd.DataFrame({
    "order_id": [501, 502, 503],
    "customer_id": [1, 2, 99]  # Customer 99 does not exist in customers_df
})
 
print(audit_customer_order_relationship(customers_data, orders_data))

import pandas as pd
import numpy as np
import time
 
def assign_tiers_apply(df: pd.DataFrame) -> pd.Series:
    """Row-by-row iteration using .apply (Slow)."""
    def label_tier(row):
        if row["spend"] > 1000:
            return "VIP"
        elif row["spend"] >= 200:
            return "Regular"
        else:
            return "Low"
    return df.apply(label_tier, axis=1)
 
def assign_tiers_vectorized(df: pd.DataFrame) -> pd.Series:
    """Vectorized conditions using np.select (Fast)."""
    conditions = [
        df["spend"] > 1000,
        df["spend"] >= 200
    ]
    choices = ["VIP", "Regular"]
    return pd.Series(np.select(conditions, choices, default="Low"), index=df.index)
 
# Benchmark on 100,000 records
sample_size = 100000
test_df = pd.DataFrame({"spend": np.random.uniform(10, 2000, size=sample_size)})
 
# Benchmark .apply()
start_apply = time.perf_counter()
res_apply = assign_tiers_apply(test_df)
time_apply = time.perf_counter() - start_apply
 
# Benchmark Vectorization
start_vec = time.perf_counter()
res_vec = assign_tiers_vectorized(test_df)
time_vec = time.perf_counter() - start_vec
 
print(f".apply() Execution Time:       {time_apply:.4f} seconds")
print(f"Vectorized Execution Time:     {time_vec:.4f} seconds")
print(f"Speedup Factor:                {time_apply / time_vec:.1f}x faster")
print("Results Identical:", (res_apply == res_vec).all())