Pythonデコレータ — 仕組みと使いどき

def greet(name: str) -> str:
    return f"Hello, {name}"

# Assign to a variable — no () means we're not calling it, just referencing it
say_hello = greet
print(say_hello("Alice"))   # 'Hello, Alice'

# Pass a function as an argument
def run_twice(fn, value):
    return fn(value), fn(value)

run_twice(greet, "Bob")     # ('Hello, Bob', 'Hello, Bob')

# Return a function from another function — this is a "factory"
def make_prefixer(prefix: str):
    def prefixed_greet(name: str) -> str:
        return f"{prefix}, {name}"
    return prefixed_greet

morning_hello = make_prefixer("Good morning")
morning_hello("Carol")      # 'Good morning, Carol'

import time
import functools

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

@timer
def fetch_user_records(db, user_id: int):
    """Fetch all records for a given user from the database."""
    return db.query("SELECT * FROM records WHERE user_id = ?", user_id)

# Calling fetch_user_records is now: timer(fetch_user_records)(db, 42)
# Which is exactly what @timer desugars to

import functools

# Without @functools.wraps — broken
def bad_timer(fn):
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper

@bad_timer
def process_batch(batch_id: int):
    """Process a single batch job."""
    pass

print(process_batch.__name__)   # 'wrapper'   ← wrong
print(process_batch.__doc__)    # None         ← docstring gone

# With @functools.wraps — correct
def good_timer(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper

@good_timer
def process_batch(batch_id: int):
    """Process a single batch job."""
    pass

print(process_batch.__name__)   # 'process_batch'  ← correct
print(process_batch.__doc__)    # 'Process a single batch job.'  ← preserved

import functools
import time

def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
    """
    Retry a function on exception with exponential backoff.

    Usage:
        @retry(max_attempts=5, delay=0.5, backoff=2.0)
        def call_external_api(endpoint: str) -> dict:
            ...
    """
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            current_delay = delay
            last_exception = None

            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception as exc:
                    last_exception = exc
                    if attempt < max_attempts:
                        print(
                            f"{fn.__name__}: attempt {attempt}/{max_attempts} failed "
                            f"({exc!r}), retrying in {current_delay:.1f}s..."
                        )
                        time.sleep(current_delay)
                        current_delay *= backoff
                    else:
                        print(f"{fn.__name__}: all {max_attempts} attempts failed.")

            raise last_exception

        return wrapper
    return decorator

@retry(max_attempts=4, delay=0.5, backoff=2.0)
def fetch_price_data(ticker: str) -> dict:
    """Fetch stock price data from external API."""
    response = requests.get(f"https://api.example.com/prices/{ticker}", timeout=5)
    response.raise_for_status()
    return response.json()

import functools
import time

class RateLimiter:
    """
    Decorator that limits how often a function can be called.
    Raises RuntimeError if the function is called within `min_interval` seconds
    of the previous call.
    """

    def __init__(self, min_interval: float):
        self.min_interval = min_interval
        self._last_called: float = 0.0

    def __call__(self, fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            since_last = now - self._last_called
            if since_last < self.min_interval:
                wait = self.min_interval - since_last
                raise RuntimeError(
                    f"{fn.__name__} called too soon — "
                    f"wait {wait:.2f}s before calling again."
                )
            self._last_called = now
            return fn(*args, **kwargs)
        return wrapper

# Usage — one call per 2 seconds
@RateLimiter(min_interval=2.0)
def send_sms_alert(phone: str, message: str) -> None:
    """Send an SMS alert via the gateway API."""
    sms_gateway.send(phone, message)

from functools import lru_cache, cache
import requests

# @cache — unbounded memoisation (Python 3.9+)
# Caches every unique set of arguments forever.
# Good for pure functions with a small domain of inputs.
@cache
def get_country_name(country_code: str) -> str:
    """Look up a country name from ISO 3166 code. Cached after first call."""
    response = requests.get(f"https://restcountries.com/v3.1/alpha/{country_code}")
    return response.json()[0]["name"]["common"]

get_country_name("DE")   # hits the API
get_country_name("DE")   # served from cache, no network call

# @lru_cache(maxsize=N) — bounded LRU cache
# Evicts least-recently-used entries once the cache hits `maxsize`.
# Better when the input domain is large and memory is a concern.
@lru_cache(maxsize=256)
def compute_discount(base_price: float, tier: str) -> float:
    """Heavy computation — price varies by tier. Cache the top 256 combinations."""
    discount_table = load_discount_table()          # expensive DB call
    rate = discount_table.get(tier, 0.0)
    return round(base_price * (1 - rate), 2)

# Inspect cache performance
print(compute_discount.cache_info())
# CacheInfo(hits=142, misses=14, maxsize=256, currsize=14)

from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime

@dataclass
class WebhookEvent:
    event_type: str
    source_id: int
    payload: dict
    received_at: datetime = field(default_factory=datetime.utcnow)
    retry_count: int = 0
    error_message: Optional[str] = None

# @dataclass generates all of this for free:
# - __init__(self, event_type, source_id, payload, received_at=..., retry_count=0, error_message=None)
# - __repr__ that shows all fields
# - __eq__ that compares field-by-field

event = WebhookEvent(event_type="order.created", source_id=9912, payload={"order_id": 44501})
print(event)
# WebhookEvent(event_type='order.created', source_id=9912, payload={...}, retry_count=0, ...)

class UserProfile:
    def __init__(self, first_name: str, last_name: str, email: str):
        self._first_name = first_name
        self._last_name = last_name
        self._email = email.strip().lower()

    @property
    def display_name(self) -> str:
        return f"{self._first_name} {self._last_name}"

    @property
    def email(self) -> str:
        return self._email

    @email.setter
    def email(self, value: str) -> None:
        if "@" not in value:
            raise ValueError(f"Invalid email address: {value!r}")
        self._email = value.strip().lower()

profile = UserProfile("Alice", "Smith", "  [email protected]  ")
print(profile.display_name)   # 'Alice Smith'
print(profile.email)          # '[email protected]'

profile.email = "[email protected]"    # setter runs validation
profile.email = "not-an-email"           # raises ValueError

import functools
from flask import request, jsonify, g

def require_auth(fn):
    """
    Decorator that validates a Bearer token before the route handler runs.
    Sets g.current_user on success; returns 401 JSON on failure.
    """
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return jsonify({"error": "Missing or malformed Authorization header"}), 401

        token = auth_header[len("Bearer "):]
        user = verify_token(token)          # your token validation logic
        if user is None:
            return jsonify({"error": "Invalid or expired token"}), 401

        g.current_user = user               # available to the route handler
        return fn(*args, **kwargs)
    return wrapper

# Usage — the auth check runs before the handler body
@app.route("/api/v1/reports/<int:report_id>")
@require_auth
def get_report(report_id: int):
    report = Report.query.get_or_404(report_id)
    if report.owner_id != g.current_user.id:
        return jsonify({"error": "Forbidden"}), 403
    return jsonify(report.to_dict())

@decorator_a
@decorator_b
@decorator_c
def my_function():
    pass

# This is exactly equivalent to:
my_function = decorator_a(decorator_b(decorator_c(my_function)))

# When my_function() is called:
# 1. decorator_a's wrapper runs first (outermost)
# 2. decorator_b's wrapper runs second
# 3. decorator_c's wrapper runs third (innermost, closest to the real function)
# 4. The real my_function body runs
# 5. Unwinding: decorator_c → decorator_b → decorator_a

# Practical example: order matters for @timer and @retry
# If timer is outermost, it measures total time including retry wait periods.
# If retry is outermost, it measures only the final successful call.

@timer          # measures: total time across all retry attempts + sleep
@retry(max_attempts=3, delay=1.0)
def sync_with_partner_api(partner_id: int) -> dict:
    ...

# vs.

@retry(max_attempts=3, delay=1.0)  # measures: only the final successful call
@timer
def sync_with_partner_api(partner_id: int) -> dict:
    ...

# Usually you want @timer outermost — it tells you the real wall-clock cost of the operation.
# Think about what "calling this function" means to the caller, then wrap in that order.