Python Functions & Builtin Functions — Middle Level¶

def add(a: int, b: int) -> int:
    return a + b

def subtract(a: int, b: int) -> int:
    return a - b

# Functions stored in a dictionary
operations: dict[str, callable] = {
    "+": add,
    "-": subtract,
}

# Dispatch pattern
def calculate(op: str, a: int, b: int) -> int:
    func = operations.get(op)
    if func is None:
        raise ValueError(f"Unknown operation: {op}")
    return func(a, b)

print(calculate("+", 10, 3))  # 13
print(calculate("-", 10, 3))  # 7

def make_counter(start: int = 0):
    """Create a counter function with an internal state."""
    count = start

    def increment(step: int = 1) -> int:
        nonlocal count  # modify the enclosing variable
        count += step
        return count

    return increment

counter = make_counter(10)
print(counter())      # 11
print(counter())      # 12
print(counter(5))     # 17

# Each call to make_counter creates a separate closure
counter2 = make_counter(0)
print(counter2())     # 1
print(counter())      # 18 — independent state

x = "global"

def outer():
    x = "enclosing"

    def inner():
        x = "local"
        print(f"inner sees: {x}")       # local

    inner()
    print(f"outer sees: {x}")           # enclosing

outer()
print(f"module sees: {x}")              # global
print(f"builtin 'len' is: {len}")       # <built-in function len>

# global — modifies a module-level variable from inside a function
total_requests = 0

def handle_request():
    global total_requests
    total_requests += 1
    return total_requests

print(handle_request())  # 1
print(handle_request())  # 2

# nonlocal — modifies a variable in the nearest enclosing scope
def make_accumulator():
    total = 0

    def add(value: float) -> float:
        nonlocal total
        total += value
        return total

    return add

acc = make_accumulator()
print(acc(10))   # 10
print(acc(20))   # 30
print(acc(5))    # 35

# Keyword-only: everything after * must be keyword
def fetch_data(url: str, *, timeout: int = 30, retries: int = 3) -> dict:
    """timeout and retries can only be passed as keyword arguments."""
    print(f"Fetching {url} with timeout={timeout}, retries={retries}")
    return {}

fetch_data("https://api.example.com", timeout=10)
# fetch_data("https://api.example.com", 10)  # TypeError!

# Positional-only (Python 3.8+): everything before / must be positional
def pow(base: float, exp: float, /) -> float:
    """base and exp can only be passed positionally."""
    return base ** exp

print(pow(2, 10))      # 1024
# print(pow(base=2, exp=10))  # TypeError!

# Combined: positional-only, normal, keyword-only
def mixed(a: int, b: int, /, c: int = 0, *, d: int = 0) -> int:
    return a + b + c + d

print(mixed(1, 2, c=3, d=4))  # 10

from typing import Any

# sorted() with complex keys
users = [
    {"name": "Alice", "age": 30, "score": 95},
    {"name": "Bob", "age": 25, "score": 87},
    {"name": "Charlie", "age": 35, "score": 92},
]

# Sort by age, then by score descending
by_age = sorted(users, key=lambda u: u["age"])
by_score_desc = sorted(users, key=lambda u: -u["score"])
# Multi-key sort
by_age_then_score = sorted(users, key=lambda u: (u["age"], -u["score"]))

print([u["name"] for u in by_age_then_score])
# ['Bob', 'Alice', 'Charlie']

# zip() with itertools.zip_longest
from itertools import zip_longest

keys = ["a", "b", "c", "d"]
values = [1, 2]

paired = dict(zip_longest(keys, values, fillvalue=0))
print(paired)  # {'a': 1, 'b': 2, 'c': 0, 'd': 0}

# enumerate with custom start
lines = ["first", "second", "third"]
for line_num, text in enumerate(lines, start=1):
    print(f"Line {line_num}: {text}")

# map() with multiple iterables
a = [1, 2, 3]
b = [10, 20, 30]
sums = list(map(lambda x, y: x + y, a, b))
print(sums)  # [11, 22, 33]

# isinstance() with tuple of types
def process(value: Any) -> str:
    if isinstance(value, (int, float)):
        return f"number: {value}"
    elif isinstance(value, str):
        return f"string: {value}"
    return f"other: {type(value).__name__}"

print(process(42))       # number: 42
print(process("hello"))  # string: hello
print(process([1, 2]))   # other: list

from functools import partial

def multiply(a: float, b: float) -> float:
    return a * b

double = partial(multiply, b=2)
triple = partial(multiply, b=3)

print(double(5))   # 10
print(triple(5))   # 15

import time
import logging
from typing import TypeVar, Callable, Any

logger = logging.getLogger(__name__)
T = TypeVar("T")


def retry(
    func: Callable[..., T],
    *args: Any,
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[type[Exception], ...] = (Exception,),
    **kwargs: Any,
) -> T:
    """Retry a function with exponential backoff.

    Args:
        func: The function to call.
        *args: Positional arguments for func.
        max_attempts: Maximum number of attempts.
        delay: Initial delay between retries in seconds.
        backoff: Multiplier for delay after each attempt.
        exceptions: Tuple of exception types to catch.
        **kwargs: Keyword arguments for func.

    Returns:
        The return value of func.

    Raises:
        The last exception if all attempts fail.
    """
    last_exception: Exception | None = None
    current_delay = delay

    for attempt in range(1, max_attempts + 1):
        try:
            result = func(*args, **kwargs)
            if attempt > 1:
                logger.info(
                    "Succeeded on attempt %d/%d", attempt, max_attempts
                )
            return result
        except exceptions as e:
            last_exception = e
            logger.warning(
                "Attempt %d/%d failed: %s. Retrying in %.1fs",
                attempt, max_attempts, e, current_delay,
            )
            if attempt < max_attempts:
                time.sleep(current_delay)
                current_delay *= backoff

    raise last_exception  # type: ignore[misc]


# Usage
def fetch_data(url: str) -> dict:
    """Simulate an unreliable network call."""
    import random
    if random.random() < 0.7:
        raise ConnectionError(f"Failed to connect to {url}")
    return {"status": "ok", "url": url}


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    result = retry(
        fetch_data,
        "https://api.example.com/data",
        max_attempts=5,
        delay=0.5,
        exceptions=(ConnectionError,),
    )
    print(result)

from typing import NamedTuple


class Employee(NamedTuple):
    name: str
    department: str
    salary: float
    years: int


def build_report(employees: list[Employee]) -> dict:
    """Build a salary report using builtin functions.

    Demonstrates: sorted, filter, map, min, max, sum, len,
    zip, enumerate, any, all, isinstance, round.
    """
    # Validate input
    if not all(isinstance(e, Employee) for e in employees):
        raise TypeError("All items must be Employee instances")

    # Filter senior employees (5+ years)
    seniors = list(filter(lambda e: e.years >= 5, employees))

    # Sort by salary descending
    by_salary = sorted(employees, key=lambda e: e.salary, reverse=True)

    # Calculate stats
    salaries = list(map(lambda e: e.salary, employees))
    total_salary = sum(salaries)
    avg_salary = round(total_salary / len(salaries), 2) if salaries else 0

    # Department breakdown using zip patterns
    departments = sorted(set(map(lambda e: e.department, employees)))
    dept_counts = []
    dept_avgs = []
    for dept in departments:
        dept_employees = [e for e in employees if e.department == dept]
        dept_counts.append(len(dept_employees))
        dept_salaries = [e.salary for e in dept_employees]
        dept_avgs.append(round(sum(dept_salaries) / len(dept_salaries), 2))

    dept_summary = dict(zip(departments, zip(dept_counts, dept_avgs)))

    # Top 3 with rank
    top_3 = [
        (rank, emp.name, emp.salary)
        for rank, emp in enumerate(by_salary[:3], start=1)
    ]

    return {
        "total_employees": len(employees),
        "total_salary": total_salary,
        "avg_salary": avg_salary,
        "min_salary": min(salaries),
        "max_salary": max(salaries),
        "salary_range": abs(max(salaries) - min(salaries)),
        "senior_count": len(seniors),
        "any_above_100k": any(s > 100_000 for s in salaries),
        "all_above_30k": all(s > 30_000 for s in salaries),
        "top_3": top_3,
        "dept_summary": dept_summary,
    }


if __name__ == "__main__":
    team = [
        Employee("Alice", "Engineering", 120_000, 8),
        Employee("Bob", "Engineering", 95_000, 3),
        Employee("Charlie", "Marketing", 85_000, 5),
        Employee("Diana", "Marketing", 78_000, 2),
        Employee("Eve", "Engineering", 110_000, 6),
        Employee("Frank", "Sales", 72_000, 4),
    ]

    report = build_report(team)
    for key, value in report.items():
        print(f"  {key}: {value}")

from typing import Callable


def make_validator(
    *,
    min_length: int = 0,
    max_length: int = 255,
    required: bool = True,
    pattern: str | None = None,
) -> Callable[[str], tuple[bool, str]]:
    """Create a string validator using closure over configuration.

    Args:
        min_length: Minimum allowed length.
        max_length: Maximum allowed length.
        required: Whether empty string is rejected.
        pattern: Optional regex pattern to match.

    Returns:
        A validator function that returns (is_valid, error_message).
    """
    import re
    compiled_pattern = re.compile(pattern) if pattern else None

    def validate(value: str) -> tuple[bool, str]:
        if not isinstance(value, str):
            return False, f"Expected str, got {type(value).__name__}"
        if required and len(value) == 0:
            return False, "Value is required"
        if len(value) < min_length:
            return False, f"Too short (min {min_length})"
        if len(value) > max_length:
            return False, f"Too long (max {max_length})"
        if compiled_pattern and not compiled_pattern.match(value):
            return False, f"Does not match pattern: {pattern}"
        return True, ""

    return validate


# Create specialized validators
validate_username = make_validator(
    min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$"
)
validate_email = make_validator(
    min_length=5, max_length=254, pattern=r"^[^@]+@[^@]+\.[^@]+$"
)

# Test them
test_cases = [
    ("username", validate_username, "alice_123"),
    ("username", validate_username, "ab"),
    ("username", validate_username, "bad name!"),
    ("email", validate_email, "alice@example.com"),
    ("email", validate_email, "not-an-email"),
]

for field, validator, value in test_cases:
    is_valid, error = validator(value)
    status = "PASS" if is_valid else f"FAIL: {error}"
    print(f"  {field}({value!r}) -> {status}")

# Bad — cryptic function signature
def proc(d: bytes, f: bool) -> bytes: ...

# Good — self-documenting
def compress_payload(input_data: bytes, include_checksum: bool) -> bytes: ...

# Bad — repeated validation logic
def create_user(name: str, email: str) -> None:
    if not name: raise ValueError("name required")
    if not email: raise ValueError("email required")
    ...

def update_user(name: str, email: str) -> None:
    if not name: raise ValueError("name required")
    if not email: raise ValueError("email required")
    ...

# Good — extract into reusable function
def require_non_empty(**fields: str) -> None:
    for name, value in fields.items():
        if not value:
            raise ValueError(f"{name} is required")

def create_user(name: str, email: str) -> None:
    require_non_empty(name=name, email=email)
    ...

class ValidationError(Exception):
    def __init__(self, field: str, message: str):
        self.field = field
        self.message = message
        super().__init__(f"{field}: {message}")


class RequiredFieldError(ValidationError):
    def __init__(self, field: str):
        super().__init__(field, "This field is required")


def validate_input(
    *,
    name: str,
    age: int,
    email: str,
) -> dict:
    """Validate input with descriptive errors."""
    errors: list[ValidationError] = []

    if not name.strip():
        errors.append(RequiredFieldError("name"))
    if age < 0 or age > 150:
        errors.append(ValidationError("age", f"Invalid age: {age}"))
    if "@" not in email:
        errors.append(ValidationError("email", "Invalid email format"))

    if errors:
        # Raise the first error; in production, collect all
        raise errors[0]

    return {"name": name.strip(), "age": age, "email": email.lower()}

from typing import TypeVar, Iterable

T = TypeVar("T")


def safe_min(iterable: Iterable[T], *, default: T) -> T:
    """Like min() but with a default for empty iterables."""
    try:
        return min(iterable)
    except ValueError:
        return default


print(safe_min([], default=0))       # 0
print(safe_min([3, 1, 4], default=0))  # 1

# Insecure — arbitrary code execution
def apply_formula(formula: str, value: float) -> float:
    return eval(formula)  # NEVER do this with user input!

# Secure — use a whitelist approach
import operator

SAFE_OPS: dict[str, callable] = {
    "+": operator.add,
    "-": operator.sub,
    "*": operator.mul,
    "/": operator.truediv,
}

def apply_operation(op: str, a: float, b: float) -> float:
    func = SAFE_OPS.get(op)
    if func is None:
        raise ValueError(f"Unsupported operation: {op}")
    return func(a, b)

# Risky — user-controlled kwargs passed to database query
def search_users(**filters):
    # If filters come from user input, they could include
    # admin=True or password__contains="..."
    query = db.users.filter(**filters)  # Dangerous!

# Safe — whitelist allowed filters
ALLOWED_FILTERS = {"name", "age", "city"}

def search_users(**filters):
    safe_filters = {
        k: v for k, v in filters.items()
        if k in ALLOWED_FILTERS
    }
    query = db.users.filter(**safe_filters)

import functools
import time


@functools.lru_cache(maxsize=128)
def expensive_computation(n: int) -> int:
    """Simulate expensive work — cached after first call."""
    time.sleep(0.1)  # simulate delay
    return sum(range(n))


start = time.perf_counter()
result1 = expensive_computation(1_000_000)
first_call = time.perf_counter() - start

start = time.perf_counter()
result2 = expensive_computation(1_000_000)
cached_call = time.perf_counter() - start

print(f"First call: {first_call:.4f}s")
print(f"Cached call: {cached_call:.6f}s")
print(f"Cache info: {expensive_computation.cache_info()}")

import operator
from functools import reduce

numbers = [1, 2, 3, 4, 5]

# Slower — lambda creates a new function object
total = reduce(lambda a, b: a + b, numbers)

# Faster — operator.add is a C-level function
total = reduce(operator.add, numbers)

# Even faster — just use sum()
total = sum(numbers)

# Bad — creates intermediate list
result = sum(list(map(lambda x: x ** 2, range(1_000_000))))

# Good — generator expression, no intermediate list
result = sum(x ** 2 for x in range(1_000_000))

import inspect

def example_func(a: int, b: str = "hello", *args, **kwargs) -> None:
    pass

sig = inspect.signature(example_func)
print(f"Parameters: {sig}")
for name, param in sig.parameters.items():
    print(f"  {name}: kind={param.kind.name}, default={param.default}")

def make_multiplier(factor):
    def multiply(x):
        return x * factor
    return multiply

double = make_multiplier(2)

# Inspect closure variables
print(double.__closure__[0].cell_contents)  # 2
print(double.__code__.co_freevars)           # ('factor',)

# Bug — all functions capture the same variable
funcs = [lambda: i for i in range(5)]
print([f() for f in funcs])  # [4, 4, 4, 4, 4] — all 4!

# Fix — capture current value with default argument
funcs = [lambda i=i: i for i in range(5)]
print([f() for f in funcs])  # [0, 1, 2, 3, 4]

# sorted() is stable — equal elements preserve original order
data = [("Alice", 30), ("Bob", 25), ("Charlie", 30)]
# Sort by age — Alice and Charlie both age 30, Alice stays first
result = sorted(data, key=lambda x: x[1])
print(result)
# [('Bob', 25), ('Alice', 30), ('Charlie', 30)]

# zip() silently drops extra elements
a = [1, 2, 3, 4, 5]
b = ["a", "b", "c"]

print(list(zip(a, b)))  # [(1, 'a'), (2, 'b'), (3, 'c')]
# 4 and 5 are silently lost!

# In Python 3.10+ use strict mode
# list(zip(a, b, strict=True))  # ValueError!

# Or use itertools.zip_longest
from itertools import zip_longest
print(list(zip_longest(a, b, fillvalue="?")))
# [(1, 'a'), (2, 'b'), (3, 'c'), (4, '?'), (5, '?')]

# Wrong
for i, value in enumerate(["a", "b"]):
    # i is the index, value is the element
    pass

# Common bug — forgetting to unpack
for pair in enumerate(["a", "b"]):
    print(pair)  # (0, 'a') — it is a tuple!

# Less readable
result = list(map(lambda x: x.strip().lower(), raw_strings))

# More Pythonic
result = [s.strip().lower() for s in raw_strings]

import functools

# Bad — loses function metadata
def my_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# Good — preserves __name__, __doc__, etc.
def my_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

def add(a, b, c):
    return a + b + c

args = [1, 2, 3]
print(add(*args))      # 6 — unpacks list into positional args

kwargs = {"a": 1, "b": 2, "c": 3}
print(add(**kwargs))    # 6 — unpacks dict into keyword args

# Combining
print(add(*[1, 2], **{"c": 3}))  # 6

# Parameters are evaluated left to right
def show(a=print("a"), b=print("b")):
    pass

# When the function is DEFINED (not called), this prints:
# a
# b

def outer():
    x = 10
    def inner():
        nonlocal x
        x += 5
        return x
    return inner()

print(outer())

funcs = [lambda x: x + i for i in range(3)]
print([f(10) for f in funcs])

matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
result = list(zip(*matrix))

def f(a, b=[], /):
    b.append(a)
    return b

print(f(1))
print(f(2))