Functions and classes

Lesson 3 — Python

Lesson 3 of 20 Beginner to intermediate ~90 min

Learning objectives

  • Define functions with positional, keyword and variadic arguments
  • Use type hints and docstrings effectively
  • Understand scope and closures
  • Write and use decorators
  • Define classes, and know when a dataclass is enough
  • Handle exceptions correctly

Functions

def study_day(event, reference):
    delta = (event - reference).days
    return delta + 1 if delta >= 0 else delta

Arguments

def summarise(data, digits=2, na_rm=True, label=None):
    ...

summarise(df)                             # positional
summarise(df, 3)                          # positional
summarise(df, digits=3)                   # keyword — clearer
summarise(df, na_rm=False, digits=3)      # keyword order does not matter

Variadic arguments:

def combine(*args, **kwargs):
    print(args)      # a tuple of positional arguments
    print(kwargs)    # a dict of keyword arguments

combine(1, 2, 3, a="x", b="y")
# (1, 2, 3)
# {'a': 'x', 'b': 'y'}

# Unpacking at the call site
values = [1, 2, 3]
options = {"digits": 3, "na_rm": False}
summarise(df, *values, **options)

Forcing keyword-only and positional-only arguments:

def f(a, b, *, verbose=False):        # verbose MUST be passed by keyword
    ...
f(1, 2, True)          # TypeError
f(1, 2, verbose=True)  # ok

def g(a, b, /, c):                    # a and b MUST be positional
    ...

The * form is worth using for boolean flags — f(df, True, False) at a call site is unreadable, and forcing f(df, verbose=True, strict=False) fixes it.

Type hints

from datetime import date
from typing import Optional

def study_day(event: date, reference: date) -> int:
    """Return the study day, where day 1 is the reference date."""
    delta = (event - reference).days
    return delta + 1 if delta >= 0 else delta


def classify(age: float | None) -> str:        # 3.10+ union syntax
    ...

def summarise(
    data: "pd.DataFrame",
    columns: list[str],
    by: str | None = None,
    digits: int = 2,
) -> "pd.DataFrame":
    ...

Type hints are not enforced at runtime. They document intent and enable static checking:

pip install mypy
mypy analysis.py
analysis.py:15: error: Argument 1 to "study_day" has incompatible type "str";
expected "date"

In a codebase of any size they pay for themselves, and editors use them for completion and error highlighting.

Docstrings

def derive_trtemfl(
    ae_start: date,
    trt_start: date | None,
    trt_end: date | None,
    window: int = 30,
) -> str:
    """Derive the treatment-emergent adverse event flag.

    An event is treatment emergent if it starts on or after the first dose
    and no later than `window` days after the last dose.

    Args:
        ae_start: Adverse event start date.
        trt_start: First dose date. `None` means the subject was not treated.
        trt_end: Last dose date. `None` means treatment is ongoing.
        window: Days after last dose to include. Defaults to 30.

    Returns:
        "Y" if treatment emergent, otherwise "N".

    Raises:
        ValueError: If `window` is negative.

    Examples:
        >>> derive_trtemfl(date(2026, 4, 1), date(2026, 3, 15), date(2026, 6, 15))
        'Y'
    """
    if window < 0:
        raise ValueError(f"window must be non-negative, got {window}")
    if trt_start is None:
        return "N"
    if ae_start < trt_start:
        return "N"
    if trt_end is not None and (ae_start - trt_end).days > window:
        return "N"
    return "Y"

This is Google style; NumPy style is also common. Pick one per project.

The >>> examples in the docstring are runnable:

python -m doctest analysis.py -v

Lambdas

square = lambda x: x**2                # legal but discouraged
sorted(people, key=lambda p: p["age"]) # this is the right use

# Prefer a def for anything named
def square(x):
    return x**2

Lambdas are for short throwaway callables passed to sorted, map, filter, or a pandas apply. If it needs a name, use def.

Scope and closures

x = 10

def f():
    x = 20          # a NEW local x
    print(x)        # 20

f()
print(x)            # 10 — unchanged
def g():
    global x
    x = 20          # modifies the module-level x

def outer():
    count = 0
    def inner():
        nonlocal count      # modifies the ENCLOSING scope
        count += 1
        return count
    return inner

counter = outer()
counter()   # 1
counter()   # 2

inner is a closure — it keeps a reference to count from the enclosing scope. This is how decorators work.

Avoid global. nonlocal is occasionally the right tool inside a closure but is rare in data code.

Decorators

A function that wraps another function.

import functools
import time


def timed(func):
    """Log how long a function takes."""
    @functools.wraps(func)              # preserves __name__ and __doc__
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.3f}s")
        return result
    return wrapper


@timed
def derive_adsl(dm, ex, ds):
    ...
    return adsl

adsl = derive_adsl(dm, ex, ds)
# derive_adsl took 2.417s

@timed is syntactic sugar for derive_adsl = timed(derive_adsl).

Useful built-in decorators:

from functools import cache, lru_cache

@cache                          # memoise; unbounded
def expensive(n):
    ...

@lru_cache(maxsize=128)         # memoise, bounded
def lookup(key):
    ...

class Circle:
    @property                   # access as an attribute, computed on demand
    def area(self):
        return 3.14159 * self.r ** 2

    @staticmethod               # no self
    def from_diameter(d):
        return Circle(d / 2)

    @classmethod                # receives the class, not the instance
    def unit(cls):
        return cls(1)

A decorator worth writing for data work:

def validates(**checks):
    """Assert conditions on a DataFrame returned by the wrapped function."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            for name, check in checks.items():
                if not check(result):
                    raise ValueError(f"{func.__name__}: check '{name}' failed")
            return result
        return wrapper
    return decorator


@validates(
    unique_subjects=lambda df: not df["USUBJID"].duplicated().any(),
    has_saffl=lambda df: "SAFFL" in df.columns,
    non_empty=lambda df: len(df) > 0,
)
def derive_adsl(dm, ex):
    ...

The postcondition checks from R’s error handling lesson, expressed as a decorator.

Classes

class Subject:
    """A trial subject."""

    def __init__(self, usubjid: str, age: int, arm: str):
        self.usubjid = usubjid
        self.age = age
        self.arm = arm
        self._events = []              # leading _ means "internal"

    def add_event(self, term: str, severity: str) -> None:
        self._events.append({"term": term, "severity": severity})

    @property
    def n_events(self) -> int:
        return len(self._events)

    @property
    def age_group(self) -> str:
        if self.age < 18:
            return "<18"
        return "18-64" if self.age < 65 else ">=65"

    def __repr__(self) -> str:
        return f"Subject({self.usubjid!r}, age={self.age}, arm={self.arm!r})"

    def __eq__(self, other) -> bool:
        if not isinstance(other, Subject):
            return NotImplemented
        return self.usubjid == other.usubjid


s = Subject("001", 45, "Placebo")
s.add_event("Headache", "MILD")
s.n_events         # 1 — no parentheses, it is a property
s.age_group        # "18-64"
s                  # Subject('001', age=45, arm='Placebo')

self is explicit and is always the first parameter of an instance method.

Dunder methods

__init__(self, ...)         # constructor
__repr__(self)              # unambiguous representation, for developers
__str__(self)               # readable representation, for users
__len__(self)               # len(obj)
__getitem__(self, key)      # obj[key]
__iter__(self)              # for x in obj
__eq__(self, other)         # ==
__lt__(self, other)         # <, enables sorting
__enter__/__exit__          # with obj:
__call__(self, ...)         # obj()

Implementing __repr__ is worth doing on every class — the default is <__main__.Subject object at 0x7f8b1c>, which tells you nothing during debugging.

Dataclasses

For classes that mostly hold data, @dataclass removes the boilerplate:

from dataclasses import dataclass, field
from datetime import date


@dataclass
class Subject:
    usubjid: str
    age: int
    arm: str
    trt_start: date | None = None
    events: list = field(default_factory=list)     # NOT events: list = []

    @property
    def age_group(self) -> str:
        return "<18" if self.age < 18 else ("18-64" if self.age < 65 else ">=65")


s = Subject("001", 45, "Placebo")
s                    # Subject(usubjid='001', age=45, arm='Placebo', ...)
s == Subject("001", 45, "Placebo")     # True — __eq__ generated

@dataclass generates __init__, __repr__ and __eq__. Note field(default_factory=list) — the mutable default problem from lesson 2 applies here, and @dataclass will refuse a bare = [].

Variants:

@dataclass(frozen=True)      # immutable, hashable
@dataclass(order=True)       # generates <, <=, >, >=
@dataclass(slots=True)       # lower memory, no dynamic attributes

Inheritance

class BaseDerivation:
    def __init__(self, spec):
        self.spec = spec

    def run(self, data):
        data = self.validate(data)
        data = self.derive(data)
        return self.finalise(data)

    def validate(self, data):
        return data

    def derive(self, data):
        raise NotImplementedError("Subclasses must implement derive()")

    def finalise(self, data):
        return data[self.spec["columns"]]


class ADSLDerivation(BaseDerivation):
    def derive(self, data):
        return data.assign(AGEGR1=...)

Use inheritance sparingly. Most data code is better served by functions; a class hierarchy is justified when you genuinely have several variants sharing a workflow, as above.

Exceptions

try:
    df = pd.read_csv(path)
except FileNotFoundError:
    print(f"Not found: {path}")
    df = pd.DataFrame()
except pd.errors.ParserError as e:
    print(f"Could not parse: {e}")
    raise
except Exception as e:
    print(f"Unexpected: {type(e).__name__}: {e}")
    raise
else:
    print(f"Loaded {len(df)} rows")     # runs only if no exception
finally:
    print("Attempt complete")           # always runs

Raising:

raise ValueError(f"Expected a positive value, got {x}")
raise ValueError("Invalid input") from original_error     # chains the cause

Custom exceptions:

class ValidationError(Exception):
    """A dataset failed a validation check."""


class MissingVariableError(ValidationError):
    def __init__(self, dataset: str, variables: list[str]):
        self.dataset = dataset
        self.variables = variables
        super().__init__(
            f"{dataset} is missing required variable(s): {', '.join(variables)}"
        )


try:
    validate(adsl)
except MissingVariableError as e:
    logger.error("Validation failed for %s: %s", e.dataset, e.variables)
WarningNever use a bare except: 
try:
    ...
except:                 # catches EVERYTHING, including KeyboardInterrupt
    pass

This swallows KeyboardInterrupt and SystemExit, so you cannot stop the program with Ctrl-C. And except: pass hides every bug.

except Exception as e:  # catches errors but not interrupts
    logger.exception("Failed")
    raise

Context managers

with open("data.csv") as f:
    content = f.read()
# file is closed automatically, even if an exception was raised

Write your own:

from contextlib import contextmanager
import time


@contextmanager
def timer(label):
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - start:.3f}s")


with timer("ADSL derivation"):
    adsl = derive_adsl(dm, ex, ds)
# ADSL derivation: 2.417s

The try/finally around the yield is what guarantees cleanup on an exception — the equivalent of on.exit() in R.

R and Python side by side

Concept R Python
Define f <- function(x) x def f(x): return x
Default function(x, y = 1) def f(x, y=1):
Variadic ... *args, **kwargs
Anonymous \(x) x + 1 lambda x: x + 1
Return Last expression Explicit return
Docs roxygen #' Docstring """..."""
Types Not standard Type hints
Objects S3, S4, R6 Classes
Method call print(obj) obj.method()
Cleanup on.exit() finally: / context manager
Errors stop(), tryCatch() raise, try/except
Memoise memoise::memoise() @functools.cache

The biggest stylistic difference: Python is method-oriented (df.groupby().mean()) where R is function-oriented (summarise(group_by(df))). That shapes how the two languages read.

Common mistakes

Mistake Consequence Fix
Mutable default argument State persists between calls Default None
Forgetting return Function returns None Explicit return
Forgetting self TypeError on call First parameter is always self
Bare except: Cannot interrupt; bugs hidden except Exception as e:
except: pass Silent failure Log and re-raise
Decorator without functools.wraps Loses __name__ and docstring Add it
Deep inheritance Hard to follow Prefer functions and composition
No __repr__ Useless debugging output Implement it, or use @dataclass

Exercise 3.1 — A validated derivation function

Write derive_bmi(weight_kg, height_cm) with type hints, a docstring, input validation raising informative errors, and handling of None values. Then write a @dataclass holding a subject’s measurements with BMI as a computed property.

Show solution 
from dataclasses import dataclass
from datetime import date
import math


def derive_bmi(weight_kg: float | None, height_cm: float | None) -> float | None:
    """Compute body mass index from weight and height.

    Args:
        weight_kg: Weight in kilograms. `None` or NaN returns `None`.
        height_cm: Height in centimetres. Must be positive.

    Returns:
        BMI in kg/m², or `None` if either input is missing.

    Raises:
        TypeError: If an input is neither a number nor `None`.
        ValueError: If height is not positive, or either value is implausible.

    Examples:
        >>> round(derive_bmi(70, 175), 2)
        22.86
        >>> derive_bmi(70, None) is None
        True
    """
    def _is_missing(x):
        return x is None or (isinstance(x, float) and math.isnan(x))

    if _is_missing(weight_kg) or _is_missing(height_cm):
        return None

    for name, value in (("weight_kg", weight_kg), ("height_cm", height_cm)):
        if not isinstance(value, (int, float)) or isinstance(value, bool):
            raise TypeError(
                f"{name} must be a number, got {type(value).__name__}: {value!r}"
            )

    if height_cm <= 0:
        raise ValueError(
            f"height_cm must be positive, got {height_cm}. "
            "Check the units — are these metres recorded as centimetres?"
        )
    if weight_kg <= 0:
        raise ValueError(f"weight_kg must be positive, got {weight_kg}")

    if not 50 <= height_cm <= 260:
        raise ValueError(
            f"height_cm of {height_cm} is outside the plausible range 50-260. "
            "Check the units."
        )

    return weight_kg / (height_cm / 100) ** 2

Note isinstance(value, bool) in the type check: bool is a subclass of int in Python, so isinstance(True, int) is True and derive_bmi(True, 175) would otherwise compute a BMI of 0.03 rather than erroring.

@dataclass
class SubjectMeasurements:
    """Baseline measurements for a trial subject."""

    usubjid: str
    weight_kg: float | None = None
    height_cm: float | None = None
    measured_on: date | None = None

    @property
    def bmi(self) -> float | None:
        """BMI in kg/m², or None if inputs are missing."""
        return derive_bmi(self.weight_kg, self.height_cm)

    @property
    def bmi_category(self) -> str:
        """WHO BMI category."""
        b = self.bmi
        if b is None:
            return "Missing"
        if b < 18.5:
            return "Underweight"
        if b < 25:
            return "Normal"
        if b < 30:
            return "Overweight"
        return "Obese"

    def __str__(self) -> str:
        b = self.bmi
        bmi_str = f"{b:.1f}" if b is not None else "-"
        return f"{self.usubjid}: BMI {bmi_str} ({self.bmi_category})"


s = SubjectMeasurements("001", weight_kg=70, height_cm=175)
print(s)
#> 001: BMI 22.9 (Normal)

print(SubjectMeasurements("002"))
#> 002: BMI - (Missing)

SubjectMeasurements("003", weight_kg=70, height_cm=0).bmi
#> ValueError: height_cm must be positive, got 0. Check the units — ...

Design points:

  • BMI is a property, not a stored field. It cannot then go stale when weight is updated, and there is no possibility of the stored value disagreeing with the inputs.
  • The plausible-range check catches the single most common real error in this derivation: height recorded in metres (1.75) rather than centimetres, which produces a BMI of 22,857 rather than 22.9. The error message names the likely cause.
  • Missing returns None, invalid raises. Missing data is normal and must propagate; an impossible value is a data error and must stop.

Exercise 3.2 — A retry decorator

Write a decorator that retries a function on failure with exponential backoff, logging each attempt, and re-raising if all attempts fail. Make the number of attempts and the exception types configurable.

Show solution 
import functools
import logging
import random
import time
from typing import Callable, TypeVar

logger = logging.getLogger(__name__)

T = TypeVar("T")


def retry(
    attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    jitter: bool = True,
    exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable:
    """Retry a function on failure with exponential backoff.

    Args:
        attempts: Total number of attempts, including the first.
        delay: Seconds to wait before the first retry.
        backoff: Multiplier applied to the delay after each failure.
        jitter: Add random jitter to avoid synchronised retries.
        exceptions: Exception types that trigger a retry. Anything else
            propagates immediately.

    Raises:
        ValueError: If `attempts` is less than 1.
    """
    if attempts < 1:
        raise ValueError(f"attempts must be at least 1, got {attempts}")

    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> T:
            current_delay = delay
            last_exception = None

            for attempt in range(1, attempts + 1):
                try:
                    return func(*args, **kwargs)

                except exceptions as e:
                    last_exception = e

                    if attempt == attempts:
                        logger.error(
                            "%s failed after %d attempt(s): %s",
                            func.__name__, attempts, e,
                        )
                        raise

                    wait = current_delay
                    if jitter:
                        wait *= 1 + random.random() * 0.25

                    logger.warning(
                        "%s attempt %d/%d failed (%s: %s); retrying in %.1fs",
                        func.__name__, attempt, attempts,
                        type(e).__name__, e, wait,
                    )
                    time.sleep(wait)
                    current_delay *= backoff

            raise last_exception          # unreachable, but satisfies type checkers

        return wrapper
    return decorator

Used:

import requests

@retry(attempts=4, delay=2.0, exceptions=(requests.RequestException,))
def fetch_terminology(version: str) -> dict:
    r = requests.get(f"https://api.example.com/ct/{version}", timeout=30)
    r.raise_for_status()
    return r.json()


@retry(attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
def query_database(sql: str):
    ...
WARNING  fetch_terminology attempt 1/4 failed (ConnectionError: ...); retrying in 2.3s
WARNING  fetch_terminology attempt 2/4 failed (ConnectionError: ...); retrying in 4.6s
INFO     Loaded 1,284 codelist terms

Design decisions worth explaining:

  • exceptions defaults to Exception but should be narrowed. Retrying a ValueError from your own validation logic is pointless — it will fail identically every time, and you have just made a fast failure slow. Retry is for transient failures: network, timeout, lock contention.

  • Jitter. If ten parallel jobs all fail at the same moment and all retry after exactly 2 seconds, they collide again. Randomising the wait spreads them out. This matters when the thing you are retrying is the thing that failed because it was overloaded.

  • functools.wraps preserves __name__, __doc__ and the signature. Without it, fetch_terminology.__name__ is "wrapper", which makes the log messages above useless and breaks introspection.

  • Re-raise on the final attempt rather than returning None. A caller that gets None from a failed fetch will produce a confusing error somewhere else; a caller that gets the original exception knows exactly what happened.

  • Logging, not printing. logger.warning respects the application’s log configuration and can be routed to a file, suppressed, or escalated. print() cannot.

What not to do

@retry(attempts=10, delay=0.1)
def derive_adsl(dm, ex):
    ...
Retrying a deterministic computation ten times is ten identical failures. Reserve this decorator for I/O.

Recap

  • Keyword arguments for clarity; * to force keyword-only for flags
  • Type hints document intent and enable mypy; they are not enforced at runtime
  • Never use a mutable default argument
  • Decorators wrap functions; always use functools.wraps
  • @dataclass removes boilerplate for data-holding classes
  • Implement __repr__ or use @dataclass — the default is useless
  • Never a bare except:; catch Exception, log, and re-raise
  • Context managers (with) are Python’s on.exit()

Next: NumPy.

Back to top