Functions

Functions are the building blocks of reusable code. Master them to write clean, maintainable automation.

Basic Functions

Definition

# Basic function
def check_host(hostname: str) -> bool:
    """Check if host is reachable."""
    # Implementation
    return True

# Call it
result = check_host("ise-01")

# Function with no return (returns None)
def log_event(message: str) -> None:
    print(f"[INFO] {message}")

Return Values

# Single return
def get_status(host: str) -> str:
    return "active"

# Multiple returns (tuple)
def get_host_info(hostname: str) -> tuple[str, int, bool]:
    ip = "10.50.1.20"
    port = 443
    is_active = True
    return ip, port, is_active

# Unpack
ip, port, active = get_host_info("ise-01")

# Early return
def validate_ip(ip: str) -> bool:
    parts = ip.split(".")
    if len(parts) != 4:
        return False
    for part in parts:
        if not part.isdigit():
            return False
        if not 0 <= int(part) <= 255:
            return False
    return True

Arguments

Positional and Keyword

def connect(host: str, port: int, timeout: int = 30) -> None:
    """Connect to host."""
    print(f"Connecting to {host}:{port} (timeout={timeout})")

# Positional
connect("ise-01", 443)

# Keyword
connect(host="ise-01", port=443)

# Mixed (positional first)
connect("ise-01", 443, timeout=60)

# Keyword order doesn't matter
connect(timeout=60, host="ise-01", port=443)

Default Values

def create_endpoint(
    mac: str,
    group: str = "Registered_Devices",
    description: str = "",
    static_profile: bool = False
) -> dict:
    return {
        "mac": mac,
        "group": group,
        "description": description,
        "staticProfile": static_profile
    }

# Use defaults
ep1 = create_endpoint("00:11:22:33:44:55")

# Override some
ep2 = create_endpoint("00:11:22:33:44:56", group="Employees")

# Override all
ep3 = create_endpoint(
    mac="00:11:22:33:44:57",
    group="Guests",
    description="Test device",
    static_profile=True
)

*args (Variable Positional)

# Accept any number of positional arguments
def check_hosts(*hosts: str) -> dict[str, bool]:
    """Check multiple hosts."""
    results = {}
    for host in hosts:
        results[host] = ping(host)
    return results

# Call with any number
check_hosts("ise-01")
check_hosts("ise-01", "ise-02", "ise-03")

# Unpack list into args
host_list = ["ise-01", "ise-02", "ise-03"]
check_hosts(*host_list)  # Same as check_hosts("ise-01", "ise-02", "ise-03")

**kwargs (Variable Keyword)

# Accept any number of keyword arguments
def create_config(**options) -> dict:
    """Create config from keyword arguments."""
    config = {
        "hostname": options.get("hostname", "default"),
        "port": options.get("port", 443),
        "timeout": options.get("timeout", 30)
    }
    # Add any extra options
    for key, value in options.items():
        if key not in config:
            config[key] = value
    return config

# Call with any keywords
config = create_config(hostname="ise-01", port=443, verify_ssl=True)

# Unpack dict into kwargs
settings = {"hostname": "ise-01", "port": 443, "timeout": 60}
config = create_config(**settings)

Combined

# Order: positional, *args, keyword-only, **kwargs
def api_call(
    method: str,          # Required positional
    endpoint: str,        # Required positional
    *path_parts,          # Variable positional
    timeout: int = 30,    # Keyword with default
    **headers             # Variable keyword
) -> dict:
    path = "/".join(path_parts)
    url = f"{endpoint}/{path}"
    print(f"{method} {url} (timeout={timeout})")
    print(f"Headers: {headers}")
    return {"url": url}

# Usage
api_call(
    "GET",
    "https://ise-01:443/ers/config",
    "endpoint",
    "00:11:22:33:44:55",
    timeout=60,
    Accept="application/json",
    Authorization="Bearer token123"
)

Keyword-Only Arguments

# Arguments after * are keyword-only
def delete_endpoint(mac: str, *, force: bool = False, dry_run: bool = False) -> bool:
    """Delete endpoint. force and dry_run must be keyword arguments."""
    if dry_run:
        print(f"Would delete {mac}")
        return False
    if force:
        # Actually delete
        return True
    return False

# Correct
delete_endpoint("00:11:22:33:44:55", force=True)

# Error: force is keyword-only
# delete_endpoint("00:11:22:33:44:55", True)  # TypeError

Positional-Only Arguments (Python 3.8+)

# Arguments before / are positional-only
def connect(host, port, /, timeout=30):
    """host and port must be positional."""
    print(f"Connecting to {host}:{port}")

# Correct
connect("ise-01", 443)
connect("ise-01", 443, timeout=60)

# Error: host is positional-only
# connect(host="ise-01", port=443)  # TypeError

Lambda Functions

# Anonymous single-expression functions
# lambda arguments: expression

# Sort hosts by number
hosts = ["ise-03", "ise-01", "ise-02"]
sorted_hosts = sorted(hosts, key=lambda h: int(h.split("-")[1]))
# ["ise-01", "ise-02", "ise-03"]

# Filter active endpoints
endpoints = [
    {"mac": "00:11:22:33:44:55", "status": "active"},
    {"mac": "00:11:22:33:44:56", "status": "inactive"},
]
active = list(filter(lambda e: e["status"] == "active", endpoints))

# Map to extract MACs
macs = list(map(lambda e: e["mac"], endpoints))

# Better: use comprehensions
active = [e for e in endpoints if e["status"] == "active"]
macs = [e["mac"] for e in endpoints]

Closures

# Inner function that captures outer variables
def make_logger(prefix: str):
    """Return a logger function with prefix."""
    def logger(message: str) -> None:
        print(f"[{prefix}] {message}")
    return logger

# Create specialized loggers
info_log = make_logger("INFO")
error_log = make_logger("ERROR")

info_log("Connected to ISE")   # [INFO] Connected to ISE
error_log("Auth failed")       # [ERROR] Auth failed

# Practical: rate limiter
def make_rate_limiter(max_calls: int, window_seconds: float):
    """Return a function that rate limits calls."""
    import time
    calls = []

    def can_call() -> bool:
        nonlocal calls
        now = time.time()
        # Remove old calls outside window
        calls = [t for t in calls if now - t < window_seconds]
        if len(calls) < max_calls:
            calls.append(now)
            return True
        return False

    return can_call

# Allow 5 calls per second
limiter = make_rate_limiter(5, 1.0)
for i in range(10):
    if limiter():
        print(f"Call {i} allowed")
    else:
        print(f"Call {i} rate limited")

Decorators

Basic Decorator

import functools
import time

def timer(func):
    """Measure execution time."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def fetch_endpoints(host: str) -> list:
    """Fetch all endpoints from ISE."""
    time.sleep(0.5)  # Simulate API call
    return ["endpoint1", "endpoint2"]

# Usage
endpoints = fetch_endpoints("ise-01")
# fetch_endpoints took 0.5003s

Decorator with Arguments

def retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry function on failure."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    print(f"Attempt {attempt + 1} failed: {e}")
                    if attempt < max_attempts - 1:
                        time.sleep(delay)
            raise last_error
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0)
def connect_to_ise(host: str) -> bool:
    """Connect to ISE with retry."""
    # May raise exception
    return True

Practical Decorators

# Authentication check
def require_auth(func):
    """Ensure user is authenticated."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if not is_authenticated():
            raise PermissionError("Authentication required")
        return func(*args, **kwargs)
    return wrapper

# Caching
def cache(ttl_seconds: int = 300):
    """Cache function results."""
    def decorator(func):
        cache_data = {}

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            now = time.time()

            if key in cache_data:
                result, timestamp = cache_data[key]
                if now - timestamp < ttl_seconds:
                    return result

            result = func(*args, **kwargs)
            cache_data[key] = (result, now)
            return result
        return wrapper
    return decorator

@cache(ttl_seconds=60)
def get_ise_nodes(deployment: str) -> list:
    """Get ISE nodes (cached for 60s)."""
    # Expensive API call
    return ["ise-01", "ise-02"]

# Validation
def validate_mac(func):
    """Validate MAC address argument."""
    import re

    @functools.wraps(func)
    def wrapper(mac: str, *args, **kwargs):
        pattern = r"^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$"
        if not re.match(pattern, mac):
            raise ValueError(f"Invalid MAC: {mac}")
        return func(mac, *args, **kwargs)
    return wrapper

@validate_mac
def lookup_endpoint(mac: str) -> dict:
    """Look up endpoint by MAC."""
    return {"mac": mac, "status": "found"}

Class Decorators

# Decorator that modifies a class
def singleton(cls):
    """Ensure only one instance exists."""
    instances = {}

    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance

@singleton
class ISEConnection:
    def __init__(self, host: str):
        self.host = host
        print(f"Connecting to {host}")

# Both get same instance
conn1 = ISEConnection("ise-01")  # Prints: Connecting to ise-01
conn2 = ISEConnection("ise-02")  # No print - returns existing
print(conn1 is conn2)  # True

Type Hints

Basic Types

from typing import Optional, Union

def process_host(
    hostname: str,
    port: int,
    timeout: float,
    enabled: bool
) -> str:
    return f"{hostname}:{port}"

# Optional (can be None)
def get_endpoint(mac: str) -> Optional[dict]:
    """Return endpoint or None if not found."""
    return None

# Union (multiple types)
def get_id(resource: Union[str, int]) -> str:
    """Accept string or int ID."""
    return str(resource)

# Python 3.10+ union syntax
def get_id_v2(resource: str | int) -> str:
    return str(resource)

Collection Types

from typing import List, Dict, Set, Tuple

# List of strings
def get_hosts() -> List[str]:
    return ["ise-01", "ise-02"]

# Dict with string keys and int values
def get_port_map() -> Dict[str, int]:
    return {"http": 80, "https": 443}

# Set of integers
def get_vlans() -> Set[int]:
    return {10, 20, 30}

# Tuple with specific types
def get_endpoint_info() -> Tuple[str, str, int]:
    return ("ise-01", "10.50.1.20", 443)

# Python 3.9+ built-in generics
def get_hosts_v2() -> list[str]:
    return ["ise-01", "ise-02"]

def get_config() -> dict[str, str | int]:
    return {"host": "ise-01", "port": 443}

Callable Types

from typing import Callable

# Function that takes a callback
def process_endpoints(
    endpoints: list[dict],
    callback: Callable[[dict], bool]
) -> list[dict]:
    """Process endpoints, filtering with callback."""
    return [e for e in endpoints if callback(e)]

# Usage
def is_active(endpoint: dict) -> bool:
    return endpoint.get("status") == "active"

active = process_endpoints(endpoints, is_active)

# Type alias for complex types
EndpointCallback = Callable[[dict], bool]
EndpointList = list[dict[str, str]]

TypedDict

from typing import TypedDict

class Endpoint(TypedDict):
    mac: str
    ip: str
    status: str
    vlan: int

def create_endpoint(mac: str, ip: str) -> Endpoint:
    return {
        "mac": mac,
        "ip": ip,
        "status": "active",
        "vlan": 10
    }

# Type checker knows the structure
endpoint = create_endpoint("00:11:22:33:44:55", "10.50.1.100")
print(endpoint["mac"])  # OK
# print(endpoint["invalid"])  # Type error

Practical Patterns

Config Factory

def create_ise_client(
    host: str,
    username: str,
    password: str,
    *,
    verify_ssl: bool = True,
    timeout: int = 30,
    retries: int = 3
) -> "ISEClient":
    """Factory function for ISE client with sensible defaults."""
    return ISEClient(
        host=host,
        auth=(username, password),
        verify=verify_ssl,
        timeout=timeout,
        max_retries=retries
    )

# Clean instantiation
client = create_ise_client(
    "ise-01.inside.domusdigitalis.dev",
    "admin",
    "password",
    verify_ssl=False  # Dev only
)

Batch Processor

from typing import Callable, TypeVar, Iterator

T = TypeVar("T")
R = TypeVar("R")

def batch_process(
    items: list[T],
    processor: Callable[[T], R],
    batch_size: int = 100,
    on_error: Callable[[T, Exception], None] | None = None
) -> Iterator[R]:
    """Process items in batches with error handling."""
    for i, item in enumerate(items):
        try:
            yield processor(item)
        except Exception as e:
            if on_error:
                on_error(item, e)
            else:
                raise

        if (i + 1) % batch_size == 0:
            print(f"Processed {i + 1} items")

# Usage
def process_endpoint(mac: str) -> dict:
    return {"mac": mac, "processed": True}

def handle_error(mac: str, error: Exception) -> None:
    print(f"Failed to process {mac}: {error}")

results = list(batch_process(
    mac_addresses,
    process_endpoint,
    batch_size=50,
    on_error=handle_error
))

Command Pattern

from typing import Protocol

class Command(Protocol):
    def execute(self) -> None: ...
    def undo(self) -> None: ...

def create_endpoint_command(mac: str, group: str):
    """Create a command to add endpoint."""
    endpoint_id = None

    def execute() -> None:
        nonlocal endpoint_id
        endpoint_id = api.create_endpoint(mac, group)
        print(f"Created endpoint {endpoint_id}")

    def undo() -> None:
        if endpoint_id:
            api.delete_endpoint(endpoint_id)
            print(f"Deleted endpoint {endpoint_id}")

    return type("CreateEndpoint", (), {
        "execute": staticmethod(execute),
        "undo": staticmethod(undo)
    })()

# Usage with command history
history = []
cmd = create_endpoint_command("00:11:22:33:44:55", "Employees")
cmd.execute()
history.append(cmd)

# Undo
history[-1].undo()

Next Module

Object-Oriented Programming - Classes, inheritance, magic methods, and dataclasses.