{ "cells": [ { "cell_type": "markdown", "id": "title", "metadata": {}, "source": [ "# Seminar 4\n", "\n", "Decorators and how to define your own\n", "\n", "Useful standard decorators\n", "\n", "Decorators with parameters" ] }, { "cell_type": "markdown", "id": "first-class-heading", "metadata": {}, "source": [ "## Functions are objects" ] }, { "cell_type": "markdown", "id": "first-class-text", "metadata": {}, "source": [ "Python functions are **first-class objects**. They can be assigned to variables, passed as arguments, and returned from other functions. A function that accepts or returns another function is called a **higher-order function**.\n", "\n", "This is the foundation of decorators." ] }, { "cell_type": "code", "id": "first-class-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:39:46.184451Z", "start_time": "2026-10-01T12:39:46.177959Z" } }, "source": [ "def apply_twice(function, value):\n", " return function(function(value))\n", "\n", "\n", "def double(number):\n", " return number * 2\n", "\n", "\n", "operation = double\n", "print(operation(5))\n", "print(apply_twice(double, 5))" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "10\n", "20\n" ] } ], "execution_count": 1 }, { "cell_type": "markdown", "id": "closure-text", "metadata": {}, "source": [ "A nested function can remember values from the enclosing function even after that function has finished. This is called a **closure**. Decorators use closures to remember the original function and, later, decorator parameters or state." ] }, { "cell_type": "code", "id": "closure-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:39:51.278788Z", "start_time": "2026-10-01T12:39:51.275758Z" } }, "source": [ "def make_multiplier(factor):\n", " def multiply(number):\n", " return number * factor\n", "\n", " return multiply\n", "\n", "\n", "triple = make_multiplier(3)\n", "print(triple(10))" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "30\n" ] } ], "execution_count": 2 }, { "cell_type": "markdown", "id": "decorator-heading", "metadata": {}, "source": [ "## What is a decorator?" ] }, { "cell_type": "markdown", "id": "decorator-definition", "metadata": {}, "source": [ "A function decorator receives a function and returns the function that should replace it. The replacement is often a new wrapper function." ] }, { "cell_type": "code", "id": "manual-decoration", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:39:58.484810Z", "start_time": "2026-10-01T12:39:58.481381Z" } }, "source": [ "def announce(function):\n", " def wrapper():\n", " print(f\"Calling {function.__name__}\")\n", " result = function()\n", " print(\"Call finished\")\n", " return result\n", "\n", " return wrapper\n", "\n", "\n", "def greet():\n", " return \"Hello!\"\n", "\n", "\n", "greet = announce(greet)\n", "print(greet())" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Calling greet\n", "Call finished\n", "Hello!\n" ] } ], "execution_count": 3 }, { "cell_type": "markdown", "id": "at-syntax-text", "metadata": {}, "source": [ "The `@` syntax performs the same assignment immediately after the decorated function is created:\n", "\n", "```python\n", "@announce\n", "def greet():\n", " return \"Hello!\"\n", "\n", "# Equivalent to: greet = announce(greet)\n", "```\n", "\n", "The decorator runs when Python executes the `def` statement. The wrapper runs later, each time the decorated function is called." ] }, { "cell_type": "markdown", "id": "general-wrapper-heading", "metadata": {}, "source": [ "### A reusable wrapper" ] }, { "cell_type": "markdown", "id": "general-wrapper-text", "metadata": {}, "source": [ "A general-purpose decorator must forward arbitrary positional and keyword arguments and return the original result. `functools.wraps` copies important metadata and provides `__wrapped__`, which points to the original function." ] }, { "cell_type": "code", "id": "trace-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:41:10.494146Z", "start_time": "2026-10-01T12:41:10.489981Z" } }, "source": [ "from functools import wraps\n", "\n", "\n", "def trace(function):\n", " @wraps(function)\n", " def wrapper(*args, **kwargs):\n", " print(f\"Calling {function.__name__} with {args=}, {kwargs=}\")\n", " result = function(*args, **kwargs)\n", " print(f\"{function.__name__} returned {result!r}\")\n", " return result\n", "\n", " return wrapper\n", "\n", "\n", "@trace\n", "def power(base, exponent=2):\n", " \"\"\"Raise base to the selected exponent.\"\"\"\n", " return base ** exponent\n", "\n", "\n", "print(power(3, exponent=3))\n", "print(power.__name__)\n", "print(power.__doc__)" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Calling power with args=(3,), kwargs={'exponent': 3}\n", "power returned 27\n", "27\n", "power\n", "Raise base to the selected exponent.\n" ] } ], "execution_count": 4 }, { "cell_type": "markdown", "id": "wrapper-checklist", "metadata": {}, "source": [ "A well-behaved wrapper normally does four things:\n", "\n", "1. accepts `*args` and `**kwargs`;\n", "2. calls the original function with those arguments;\n", "3. returns the original result;\n", "4. uses `@wraps(function)`.\n", "\n", "If a wrapper observes exceptions, it should not silently suppress them unless that behavior is the explicit purpose of the decorator." ] }, { "cell_type": "markdown", "id": "exercise-one", "metadata": {}, "source": [ "## Exercise 1: Function monitor\n", "\n", "Implement a `@monitor` decorator that collects statistics about a function.\n", "\n", "The decorator must:\n", "\n", "- work with arbitrary positional and keyword arguments;\n", "- preserve the original function's metadata;\n", "- return the original result;\n", "- count all calls, successful calls, and failed calls separately;\n", "- measure the total time spent inside the original function;\n", "- re-raise exceptions without replacing or suppressing them;\n", "- expose a mutable statistics dictionary as `wrapped.stats`.\n", "\n", "The time must be recorded for both successful and failed calls. Use `time.perf_counter()` for measurement." ] }, { "cell_type": "code", "id": "exercise-one-starter", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:45:56.462470Z", "start_time": "2026-10-01T12:45:56.459271Z" } }, "source": [ "from functools import wraps\n", "from time import perf_counter\n", "\n", "\n", "def monitor(function):\n", " stats = {\"calls\": 0, \"successes\": 0, \"failures\": 0, \"total_time\": 0.0}\n", "\n", " @wraps(function)\n", " def wrapper(*args, **kwargs):\n", " # Add monitoring here.\n", " return function(*args, **kwargs)\n", "\n", " wrapper.stats = stats\n", " return wrapper\n", "\n", "\n", "@monitor\n", "def divide(dividend, divisor=1):\n", " \"\"\"Divide one number by another.\"\"\"\n", " return dividend / divisor\n", "\n", "\n", "# After implementing monitor, these checks should pass.\n", "# assert divide(10, divisor=2) == 5\n", "# try:\n", "# divide(10, 0)\n", "# except ZeroDivisionError:\n", "# pass\n", "# assert divide.__name__ == \"divide\"\n", "# assert divide.stats[\"calls\"] == 2\n", "# assert divide.stats[\"successes\"] == 1\n", "# assert divide.stats[\"failures\"] == 1\n", "# assert divide.stats[\"total_time\"] >= 0" ], "outputs": [], "execution_count": 5 }, { "cell_type": "markdown", "id": "exercise-one-solution-heading", "metadata": {}, "source": [ "### Solution" ] }, { "cell_type": "code", "id": "exercise-one-solution", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:46:06.726955Z", "start_time": "2026-10-01T12:46:06.719120Z" } }, "source": [ "from functools import wraps\n", "from time import perf_counter\n", "\n", "\n", "def monitor(function):\n", " stats = {\"calls\": 0, \"successes\": 0, \"failures\": 0, \"total_time\": 0.0}\n", "\n", " @wraps(function)\n", " def wrapper(*args, **kwargs):\n", " stats[\"calls\"] += 1\n", " started_at = perf_counter()\n", " try:\n", " result = function(*args, **kwargs)\n", " except Exception:\n", " stats[\"failures\"] += 1\n", " raise\n", " else:\n", " stats[\"successes\"] += 1\n", " return result\n", " finally:\n", " stats[\"total_time\"] += perf_counter() - started_at\n", "\n", " wrapper.stats = stats\n", " return wrapper\n", "\n", "\n", "@monitor\n", "def divide(dividend, divisor=1):\n", " \"\"\"Divide one number by another.\"\"\"\n", " return dividend / divisor\n", "\n", "\n", "assert divide(10, divisor=2) == 5\n", "try:\n", " divide(10, 0)\n", "except ZeroDivisionError:\n", " pass\n", "assert divide.__name__ == \"divide\"\n", "assert divide.stats[\"calls\"] == 2\n", "assert divide.stats[\"successes\"] == 1\n", "assert divide.stats[\"failures\"] == 1\n", "assert divide.stats[\"total_time\"] >= 0\n", "divide.stats" ], "outputs": [ { "data": { "text/plain": [ "{'calls': 2,\n", " 'successes': 1,\n", " 'failures': 1,\n", " 'total_time': 2.6949996936309617e-06}" ] }, "execution_count": 6, "metadata": {}, "output_type": "execute_result" } ], "execution_count": 6 }, { "cell_type": "markdown", "id": "parameterized-heading", "metadata": {}, "source": [ "## Decorators with parameters" ] }, { "cell_type": "markdown", "id": "parameterized-text", "metadata": {}, "source": [ "In `@repeat(times=3)`, Python first calls `repeat(times=3)`. That call must return the actual decorator, which then receives the decorated function. The wrapper receives the arguments used when the decorated function is called.\n", "\n", "This produces three function levels:\n", "\n", "1. the **decorator factory** remembers decorator arguments;\n", "2. the **decorator** receives the original function;\n", "3. the **wrapper** receives call arguments." ] }, { "cell_type": "code", "execution_count": null, "id": "repeat-example", "metadata": {}, "outputs": [], "source": [ "from functools import wraps\n", "\n", "\n", "def repeat(*, times):\n", " if not isinstance(times, int) or isinstance(times, bool) or times < 1:\n", " raise ValueError(\"times must be a positive integer\")\n", "\n", " def decorator(function):\n", " @wraps(function)\n", " def wrapper(*args, **kwargs):\n", " result = None\n", " for _ in range(times):\n", " result = function(*args, **kwargs)\n", " return result\n", "\n", " return wrapper\n", "\n", " return decorator\n", "\n", "\n", "@repeat(times=3)\n", "def announce(message):\n", " print(message)\n", " return len(message)\n", "\n", "\n", "length = announce(\"Decorators!\")\n", "print(f\"{length=}\")" ] }, { "cell_type": "markdown", "id": "validation-text", "metadata": {}, "source": [ "Parameters such as `times` are usually validated in the factory. Invalid configuration is then reported immediately when the decorated function is defined, instead of much later when it happens to be called." ] }, { "cell_type": "markdown", "id": "exercise-two", "metadata": {}, "source": [ "## Exercise 2: Configurable retry policy\n", "\n", "Implement this parameterized decorator:\n", "\n", "```python\n", "@retry(attempts=4, exceptions=(ConnectionError, TimeoutError))\n", "def fetch_data():\n", " ...\n", "```\n", "\n", "Requirements:\n", "\n", "- `attempts` is the total number of calls, not the number of retries after the first call;\n", "- reject a non-integer, a boolean, or a value below one when the function is decorated;\n", "- retry only exceptions matched by `exceptions`;\n", "- return immediately after a successful call;\n", "- re-raise the final exception with its original traceback;\n", "- preserve metadata and support arbitrary arguments.\n", "\n", "Do not add `sleep`: deterministic, fast retries are easier to test in this seminar." ] }, { "cell_type": "code", "id": "exercise-two-starter", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:49:28.176296Z", "start_time": "2026-10-01T12:49:28.172524Z" } }, "source": [ "from functools import wraps\n", "\n", "\n", "def retry(*, attempts, exceptions=(Exception,)):\n", " # Validate attempts here.\n", " def decorator(function):\n", " @wraps(function)\n", " def wrapper(*args, **kwargs):\n", " # Add the retry loop here.\n", " return function(*args, **kwargs)\n", "\n", " return wrapper\n", "\n", " return decorator\n", "\n", "\n", "outcomes = iter([ConnectionError(\"service unavailable\"), TimeoutError(\"request timed out\"), {\"status\": \"ok\"}])\n", "\n", "\n", "@retry(attempts=4, exceptions=(ConnectionError, TimeoutError))\n", "def fetch_data(endpoint):\n", " outcome = next(outcomes)\n", " if isinstance(outcome, Exception):\n", " raise outcome\n", " return endpoint, outcome\n", "\n", "\n", "# assert fetch_data(\"/health\") == (\"/health\", {\"status\": \"ok\"})\n", "# assert fetch_data.__name__ == \"fetch_data\"" ], "outputs": [], "execution_count": 10 }, { "cell_type": "markdown", "id": "exercise-two-solution-heading", "metadata": {}, "source": [ "### Solution" ] }, { "cell_type": "code", "id": "exercise-two-solution", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:50:32.980121Z", "start_time": "2026-10-01T12:50:32.975837Z" } }, "source": [ "from functools import wraps\n", "\n", "\n", "def retry(*, attempts, exceptions=(Exception,)):\n", " if not isinstance(attempts, int) or isinstance(attempts, bool) or attempts < 1:\n", " raise ValueError(\"attempts must be a positive integer\")\n", "\n", " def decorator(function):\n", " @wraps(function)\n", " def wrapper(*args, **kwargs):\n", " for attempt_number in range(1, attempts + 1):\n", " try:\n", " return function(*args, **kwargs)\n", " except exceptions:\n", " if attempt_number == attempts:\n", " raise\n", "\n", " return wrapper\n", "\n", " return decorator\n", "\n", "\n", "outcomes = iter([ConnectionError(\"service unavailable\"), TimeoutError(\"request timed out\"), {\"status\": \"ok\"}])\n", "\n", "\n", "@retry(attempts=4, exceptions=(ConnectionError, TimeoutError))\n", "def fetch_data(endpoint):\n", " outcome = next(outcomes)\n", " if isinstance(outcome, Exception):\n", " raise outcome\n", " return endpoint, outcome\n", "\n", "\n", "assert fetch_data(\"/health\") == (\"/health\", {\"status\": \"ok\"})\n", "assert fetch_data.__name__ == \"fetch_data\"" ], "outputs": [], "execution_count": 14 }, { "cell_type": "markdown", "id": "stacking-heading", "metadata": {}, "source": [ "## Combining decorators" ] }, { "cell_type": "markdown", "id": "stacking-text", "metadata": {}, "source": [ "Stacked decorators are applied from bottom to top:\n", "\n", "```python\n", "@outer\n", "@inner\n", "def process():\n", " ...\n", "\n", "# Equivalent to: process = outer(inner(process))\n", "```\n", "\n", "The order can change the meaning. With `@monitor` outside `@retry`, one logical request is counted once. With `@monitor` inside `@retry`, every attempt is counted." ] }, { "cell_type": "code", "id": "stacking-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:51:16.502107Z", "start_time": "2026-10-01T12:51:16.496790Z" } }, "source": [ "def make_unstable_function(failures_before_success):\n", " calls = 0\n", "\n", " def unstable():\n", " nonlocal calls\n", " calls += 1\n", " print(f\"Inner calls: {calls}\")\n", " if calls <= failures_before_success:\n", " raise ConnectionError(\"temporary failure\")\n", " return \"ready\"\n", "\n", " return unstable\n", "\n", "\n", "monitored_request = monitor(\n", " retry(attempts=3, exceptions=(ConnectionError,))(\n", " make_unstable_function(failures_before_success=2)\n", " )\n", ")\n", "\n", "assert monitored_request() == \"ready\"\n", "assert monitored_request.stats[\"calls\"] == 1\n", "monitored_request.stats" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inner calls: 1\n", "Inner calls: 2\n", "Inner calls: 3\n" ] }, { "data": { "text/plain": [ "{'calls': 1,\n", " 'successes': 1,\n", " 'failures': 0,\n", " 'total_time': 5.0286999794479925e-05}" ] }, "execution_count": 18, "metadata": {}, "output_type": "execute_result" } ], "execution_count": 18 }, { "cell_type": "markdown", "id": "standard-heading", "metadata": {}, "source": [ "## Useful standard decorators" ] }, { "cell_type": "markdown", "id": "cache-heading", "metadata": {}, "source": [ "### `functools.lru_cache` and `functools.cache`" ] }, { "cell_type": "markdown", "id": "cache-text", "metadata": {}, "source": [ "`@lru_cache(maxsize=...)` remembers results by arguments. A repeated call can return the stored result without executing the function body again. When the size limit is reached, the least recently used entries are removed.\n", "\n", "Cached arguments must be hashable because they are used as dictionary keys. Caching is safest for **pure functions**: the same arguments should always produce the same result without important side effects." ] }, { "cell_type": "code", "id": "lru-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:51:49.278069Z", "start_time": "2026-10-01T12:51:49.274735Z" } }, "source": [ "from functools import lru_cache\n", "\n", "\n", "@lru_cache(maxsize=None)\n", "def count_grid_paths(rows, columns):\n", " if rows == 0 or columns == 0:\n", " return 1\n", " return count_grid_paths(rows - 1, columns) + count_grid_paths(rows, columns - 1)\n", "\n", "\n", "print(count_grid_paths(20, 20))\n", "print(count_grid_paths.cache_info())\n", "count_grid_paths.cache_clear()" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "137846528820\n", "CacheInfo(hits=361, misses=440, maxsize=None, currsize=440)\n" ] } ], "execution_count": 19 }, { "cell_type": "markdown", "id": "cache-details", "metadata": {}, "source": [ "Useful cache operations:\n", "\n", "- `function.cache_info()` reports hits, misses, size, and capacity;\n", "- `function.cache_clear()` removes all stored results;\n", "- `function.cache_parameters()` reports the configuration;\n", "- `@functools.cache` is equivalent to an unbounded cache and is appropriate only when unrestricted growth is acceptable.\n", "\n", "Avoid caching functions that read changing files, request live data, generate randomness, or depend on hidden global state. Also remember that `f(1)` and `f(value=1)` can occupy different cache entries." ] }, { "cell_type": "code", "id": "unhashable-cache-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:52:25.758907Z", "start_time": "2026-10-01T12:52:25.755977Z" } }, "source": [ "from functools import cache\n", "\n", "\n", "@cache\n", "def total(values):\n", " return sum(values)\n", "\n", "\n", "try:\n", " total([1, 2, 3])\n", "except TypeError as error:\n", " print(error)\n", "\n", "print(total((1, 2, 3))) # A tuple of numbers is hashable." ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "unhashable type: 'list'\n", "6\n" ] } ], "execution_count": 20 }, { "cell_type": "markdown", "id": "static-class-heading", "metadata": {}, "source": [ "### `staticmethod` and `classmethod`" ] }, { "cell_type": "markdown", "id": "static-class-text", "metadata": {}, "source": [ "These decorators control how a function stored on a class is bound:\n", "\n", "- a normal method receives an instance as its first argument;\n", "- a `@classmethod` receives the class as `cls`;\n", "- a `@staticmethod` receives neither automatically.\n", "\n", "We keep the example small because class design is a separate topic." ] }, { "cell_type": "code", "execution_count": null, "id": "static-class-example", "metadata": {}, "outputs": [], "source": [ "class CsvTools:\n", " separator = \",\"\n", "\n", " @staticmethod\n", " def clean(value):\n", " return str(value).strip()\n", "\n", " @classmethod\n", " def make_row(cls, values):\n", " cleaned_values = map(cls.clean, values)\n", " return cls.separator.join(cleaned_values)\n", "\n", "\n", "print(CsvTools.clean(\" Bremen \"))\n", "print(CsvTools.make_row([\"Ada\", 42, \"Berlin\"]))" ] }, { "cell_type": "markdown", "id": "singledispatch-heading", "metadata": {}, "source": [ "### `functools.singledispatch`" ] }, { "cell_type": "markdown", "id": "singledispatch-text", "metadata": {}, "source": [ "`@singledispatch` creates a generic function whose implementation is selected from the type of its first argument. New implementations can be registered without adding a long `if`/`elif` chain to the original function." ] }, { "cell_type": "code", "id": "singledispatch-example", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:55:05.989858Z", "start_time": "2026-10-01T12:55:05.986436Z" } }, "source": [ "from functools import singledispatch\n", "\n", "\n", "@singledispatch\n", "def describe(value):\n", " return f\"object of type {type(value).__name__}\"\n", "\n", "\n", "@describe.register\n", "def _(value: int):\n", " return f\"integer with absolute value {abs(value)}\"\n", "\n", "\n", "@describe.register\n", "def _(value: list):\n", " return f\"list containing {len(value)} items\"\n", "\n", "\n", "print(describe(-7))\n", "print(describe([1, 2, 3]))\n", "print(describe({\"name\": \"Ada\"}))" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "integer with absolute value 7\n", "list containing 3 items\n", "object of type dict\n" ] } ], "execution_count": 24 }, { "cell_type": "markdown", "id": "singledispatch-note", "metadata": {}, "source": [ "Only the first argument controls dispatch. The other arguments are passed to the selected implementation normally. The base implementation is also the fallback for unregistered types.\n", "\n", "Registered functions are often named `_` because registration keeps a reference to each implementation; the module-level name itself is not needed afterwards. Use descriptive names when you want to inspect or decorate a particular implementation later." ] }, { "cell_type": "markdown", "id": "exercise-three", "metadata": {}, "source": [ "## Exercise 3: Extensible data formatter\n", "\n", "Build `format_value(value)`, which converts nested Python data into a compact, JSON-like string.\n", "\n", "Requirements:\n", "\n", "1. Use `@singledispatch` rather than a type-checking `if`/`elif` chain.\n", "2. Support `str`, `int`, `float`, `bool`, `None`, `list`, and `dict`.\n", "3. Lists and dictionaries must recursively call `format_value`, so arbitrary nesting works.\n", "4. Raise `TypeError` for unsupported values.\n", "5. Apply `@monitor` to the generic function so recursive calls are counted.\n", "6. Add `@lru_cache` to the string formatter, whose arguments are safe cache keys. Do not cache the generic function because lists and dictionaries are unhashable.\n", "\n", "For simplicity, dictionary keys may be converted to strings. Preserve dictionary insertion order." ] }, { "cell_type": "code", "id": "exercise-three-starter", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:56:22.867977Z", "start_time": "2026-10-01T12:56:22.865357Z" } }, "source": [ "from functools import lru_cache, singledispatch\n", "\n", "\n", "@monitor\n", "@singledispatch\n", "def format_value(value):\n", " # Raise TypeError for unsupported types.\n", " pass\n", "\n", "\n", "# Register the required implementations here.\n", "\n", "\n", "record = {\"name\": \"Ada\", \"scores\": [10, 20.5], \"active\": True, \"note\": None}\n", "\n", "# expected = (\n", "# '{\"name\": \"Ada\", \"scores\": [10, 20.5], '\n", "# '\"active\": true, \"note\": null}'\n", "# )\n", "# assert format_value(record) == expected\n", "# assert format_value.stats[\"calls\"] > 1" ], "outputs": [], "execution_count": 25 }, { "cell_type": "markdown", "id": "exercise-three-solution-heading", "metadata": {}, "source": [ "### Solution" ] }, { "cell_type": "code", "id": "exercise-three-solution", "metadata": { "ExecuteTime": { "end_time": "2026-10-01T12:57:17.383161Z", "start_time": "2026-10-01T12:57:17.377828Z" } }, "source": [ "from functools import lru_cache, singledispatch\n", "\n", "\n", "@monitor\n", "@singledispatch\n", "def format_value(value):\n", " raise TypeError(f\"unsupported type: {type(value).__name__}\")\n", "\n", "\n", "@format_value.register\n", "@lru_cache(maxsize=128)\n", "def format_string(value: str):\n", " escaped = value.replace(\"\\\\\", \"\\\\\\\\\").replace('\\\"', '\\\\\"')\n", " return f'\"{escaped}\"'\n", "\n", "\n", "@format_value.register\n", "def format_bool(value: bool):\n", " return \"true\" if value else \"false\"\n", "\n", "\n", "@format_value.register\n", "def format_int(value: int):\n", " return str(value)\n", "\n", "\n", "@format_value.register\n", "def format_float(value: float):\n", " return str(value)\n", "\n", "\n", "@format_value.register(type(None))\n", "def format_none(value):\n", " return \"null\"\n", "\n", "\n", "@format_value.register\n", "def format_list(value: list):\n", " items = (format_value(item) for item in value)\n", " return \"[\" + \", \".join(items) + \"]\"\n", "\n", "\n", "@format_value.register\n", "def format_dict(value: dict):\n", " items = (f\"{format_value(str(key))}: {format_value(item)}\" for key, item in value.items())\n", " return \"{\" + \", \".join(items) + \"}\"\n", "\n", "\n", "record = {\"name\": \"Ada\", \"scores\": [10, 20.5], \"active\": True, \"note\": None}\n", "expected = ('{\"name\": \"Ada\", \"scores\": [10, 20.5], ' '\"active\": true, \"note\": null}')\n", "\n", "assert format_value(record) == expected\n", "assert format_value.stats[\"calls\"] > 1\n", "print(format_value(record))\n", "print(format_value.stats)\n", "print(format_string.cache_info())" ], "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "{\"name\": \"Ada\", \"scores\": [10, 20.5], \"active\": true, \"note\": null}\n", "{'calls': 22, 'successes': 22, 'failures': 0, 'total_time': 9.158899865724379e-05}\n", "CacheInfo(hits=5, misses=5, maxsize=128, currsize=5)\n" ] } ], "execution_count": 26 }, { "cell_type": "markdown", "id": "design-notes-heading", "metadata": {}, "source": [ "## Design notes and common mistakes" ] }, { "cell_type": "markdown", "id": "design-notes", "metadata": {}, "source": [ "- **Do not forget the return value.** A wrapper that omits `return` changes every result to `None`.\n", "- **Use `@wraps`.** Documentation tools, debuggers, and other decorators rely on correct metadata.\n", "- **Validate decorator configuration early.** Factory arguments are known when the function is defined.\n", "- **Be explicit about exceptions.** Logging and monitoring should normally re-raise; retrying should catch only expected failures.\n", "- **Consider decorator order.** `monitor(retry(function))` and `retry(monitor(function))` collect different statistics.\n", "- **Cache values, not side effects.** Hashability is required, but purity is the more important design condition.\n", "- **Prefer functions for function decorators.** Class-based decorators exist, but closures are sufficient for the patterns in this seminar." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10" } }, "nbformat": 4, "nbformat_minor": 5 }