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.