Python ParamSpec and Concatenate Practice Problems & Exercises
Practice: ParamSpec and Concatenate
← Back to lessonEasy
Write a timed decorator using ParamSpec so that the wrapped function retains its exact parameter signature in type checkers.
import time
import functools
from typing import TypeVar, Callable
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
def timed(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__}({', '.join(str(a) for a in args)}) took {elapsed:.3f}s")
return result
return wrapper
@timed
def slow_add(a: int, b: int) -> int:
time.sleep(0.001)
return a + b
result = slow_add(2, 3)
print(f"Result: {result}")Expected Output
slow_add(2, 3) took 0.XXXs
Result: 5Hints
Hint 1: Declare P = ParamSpec("P") and T = TypeVar("T"). The wrapper signature is (*args: P.args, **kwargs: P.kwargs).
Hint 2: The return type of the decorator is Callable[P, T] -> Callable[P, T]. This preserves the original function signature.
Write a retry(max_attempts) parametrized decorator using ParamSpec that retries a function on exception, preserving the original type signature.
import functools
from typing import TypeVar, Callable
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
def retry(max_attempts: int) -> Callable[[Callable[P, T]], Callable[P, T]]:
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
last_exc: Exception = RuntimeError("no attempts")
for attempt in range(1, max_attempts + 1):
try:
result = func(*args, **kwargs)
print(f"Attempt {attempt} succeeded: {result}")
return result
except Exception as e:
last_exc = e
print(f"Attempt {attempt} failed: {e}")
raise last_exc
return wrapper
return decorator
call_count = 0
@retry(max_attempts=3)
def flaky_operation() -> int:
global call_count
call_count += 1
if call_count < 3:
raise RuntimeError("temporary error")
return 42
flaky_operation()Expected Output
Attempt 1 failed: temporary error
Attempt 2 failed: temporary error
Attempt 3 succeeded: 42Hints
Hint 1: Use P = ParamSpec("P") and T = TypeVar("T") to build a retry wrapper that preserves the wrapped function signature.
Hint 2: The retry loop calls func(*args, **kwargs) inside a try/except. After max_attempts, re-raise the last exception.
Create a logged decorator with ParamSpec that logs function name, arguments, return value, and any exception raised.
import functools
from typing import TypeVar, Callable
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
def logged(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
print(f"[LOG] {func.__name__} called with args={args}, kwargs={kwargs}")
try:
result = func(*args, **kwargs)
print(f"[LOG] {func.__name__} returned {result}")
return result
except Exception as e:
print(f"[LOG] {func.__name__} raised {type(e).__name__}: {e}")
raise
return wrapper
@logged
def divide(a: float, b: float) -> float:
return a / b
result = divide(10, 2)
print(f"Result: {result}")
try:
divide(10, 0)
except ZeroDivisionError:
passExpected Output
[LOG] divide called with args=(10, 2), kwargs={}
[LOG] divide returned 5.0
Result: 5.0
[LOG] divide called with args=(10, 0), kwargs={}
[LOG] divide raised ZeroDivisionError: division by zeroHints
Hint 1: Wrap the function call in try/except. Log both successful returns and exceptions.
Hint 2: Use P.args and P.kwargs in the wrapper signature to preserve the parameter spec.
Medium
Use Concatenate to write an authenticated decorator that injects a user string as the first argument to the wrapped function, prepending it to the original parameters.
import functools
from typing import TypeVar, Callable
try:
from typing import ParamSpec, Concatenate
except ImportError:
from typing_extensions import ParamSpec, Concatenate
P = ParamSpec("P")
T = TypeVar("T")
class Request:
def __init__(self, user: str, is_auth: bool) -> None:
self.user = user
self.is_auth = is_auth
def authenticated(
func: Callable[Concatenate[str, P], T]
) -> Callable[Concatenate[Request, P], T]:
@functools.wraps(func)
def wrapper(request: Request, *args: P.args, **kwargs: P.kwargs) -> T:
if not request.is_auth:
raise PermissionError("Unauthenticated access blocked")
return func(request.user, *args, **kwargs)
return wrapper
@authenticated
def greet(user: str, greeting: str = "Hello") -> str:
return f"{greeting}, {user}!"
req_ok = Request("admin", is_auth=True)
req_bad = Request("guest", is_auth=False)
print(f"authenticated: {req_ok.is_auth}")
print(f"user: {req_ok.user}")
result = greet(req_ok, greeting="Hello")
print(f"greet result: {result}")
try:
greet(req_bad)
except PermissionError as e:
print(e)Expected Output
authenticated: True
user: admin
greet result: Hello, admin!
Unauthenticated access blockedHints
Hint 1: Concatenate[ExtraType, P] prepends an extra positional argument to the parameter spec P.
Hint 2: The outer function signature is Callable[Concatenate[RequestType, P], T] and the wrapper adds the request argument before forwarding.
Write a memoize decorator using ParamSpec that caches return values. Track cache hits to prove the cache is being used.
import functools
from typing import TypeVar, Callable, Dict, Tuple, Any
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
def memoize(func: Callable[P, T]) -> Callable[P, T]:
cache: Dict[Tuple, T] = {}
hits = [0]
misses = [0]
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
key = args + tuple(sorted(kwargs.items()))
if key in cache:
hits[0] += 1
return cache[key]
misses[0] += 1
result = func(*args, **kwargs)
cache[key] = result
return result
wrapper.cache_hits = hits # type: ignore[attr-defined]
wrapper.cache_misses = misses # type: ignore[attr-defined]
return wrapper
@memoize
def fib(n: int) -> int:
if n <= 1:
return n
return fib(n - 1) + fib(n - 2)
r1 = fib(10)
print(f"fib(10) = {r1}")
r2 = fib(10)
print(f"fib(10) = {r2} (cached)")
print(f"Cache hits: {fib.cache_hits[0]}") # type: ignore[attr-defined]
print(f"Unique calls: {fib.cache_misses[0]}") # type: ignore[attr-defined]Expected Output
fib(10) = 55
fib(10) = 55 (cached)
Cache hits: 1
Unique calls: 11Hints
Hint 1: The memoize decorator uses a dict keyed by (args, frozenset(kwargs.items())). P preserves the original signature.
Hint 2: Track cache hits and misses separately so you can verify caching is working.
Build a rate_limit(calls, period) decorator using ParamSpec. It should raise RateLimitError if the function is called more than calls times in the given period seconds.
import time
import functools
from collections import deque
from typing import TypeVar, Callable
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
class RateLimitError(Exception):
pass
def rate_limit(calls: int, period: float) -> Callable[[Callable[P, T]], Callable[P, T]]:
def decorator(func: Callable[P, T]) -> Callable[P, T]:
timestamps: deque = deque()
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
now = time.monotonic()
# Remove timestamps outside the window
while timestamps and now - timestamps[0] > period:
timestamps.popleft()
if len(timestamps) >= calls:
raise RateLimitError(
f"max {calls} calls per {period}s window"
)
timestamps.append(now)
return func(*args, **kwargs)
return wrapper
return decorator
successful = 0
@rate_limit(calls=3, period=1.0)
def api_call(endpoint: str) -> str:
return f"ok"
for i in range(1, 5):
try:
result = api_call("/data")
successful += 1
print(f"Call {i}: {result}")
except RateLimitError as e:
print(f"Call {i}: RateLimitError: {e}")
print(f"Total successful: {successful}")Expected Output
Call 1: ok
Call 2: ok
Call 3: ok
Call 4: RateLimitError: max 3 calls per 1.0s window
Total successful: 3Hints
Hint 1: Store the call timestamps in a deque. Before each call, purge entries older than the window. If len(deque) >= limit, raise RateLimitError.
Hint 2: Use P and T from ParamSpec and TypeVar to preserve the wrapped function signature.
Create a configurable debug_log(log_args, log_result) decorator factory using ParamSpec that optionally logs call arguments and the return value.
import functools
from typing import TypeVar, Callable
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
def debug_log(
log_args: bool = True,
log_result: bool = True,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
print(f"[DEBUG] {func.__name__} called")
if log_args:
print(f"[DEBUG] {func.__name__} args: {args} kwargs: {kwargs}")
result = func(*args, **kwargs)
if log_result:
print(f"[DEBUG] {func.__name__} returned: {result}")
return result
return wrapper
return decorator
@debug_log(log_args=True, log_result=True)
def process(data: str) -> str:
return f"PROCESSED: {data.upper()}"
result = process("data")
print(f"Result: {result}")Expected Output
[DEBUG] process called
[DEBUG] process args: ('data',) kwargs: {}
[DEBUG] process returned: PROCESSED: DATA
Result: PROCESSED: DATAHints
Hint 1: The outer factory returns a decorator. The decorator uses P and T to preserve types. Configuration (log_args, log_result) is captured by the factory closure.
Hint 2: Conditionally print args and return value based on the flags.
Hard
Build a middleware pipeline where each middleware is a decorator using ParamSpec. Chain them to process a "request" string through auth, logging, and validation layers.
import functools
from typing import TypeVar, Callable, List
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
MiddlewareType = Callable[[Callable[P, T]], Callable[P, T]]
def auth_middleware(token: str) -> MiddlewareType:
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
print(f"[auth] checking token: {token}")
return func(*args, **kwargs)
return wrapper
return decorator
def logging_middleware(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
print(f"[log] calling {func.__name__} with {args}")
return func(*args, **kwargs)
return wrapper
def validation_middleware(min_len: int) -> MiddlewareType:
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
# Validate first positional argument if it's a string
if args and isinstance(args[0], str):
if len(args[0]) < min_len:
raise ValueError(f"Input too short: min {min_len} chars")
print(f"[validate] input length OK: {len(args[0])}")
return func(*args, **kwargs)
return wrapper
return decorator
def handler(request: str) -> str:
return f"PROCESSED: {request.upper()}"
# Chain middleware: validation -> logging -> auth -> handler
pipeline = (
auth_middleware("secret")(
logging_middleware(
validation_middleware(5)(
handler
)
)
)
)
result = pipeline("request data")
print(f"Final: {result}")Expected Output
[auth] checking token: secret
[log] calling handler with ('request data',)
[validate] input length OK: 12
Handler result: PROCESSED: REQUEST DATA
Final: PROCESSED: REQUEST DATAHints
Hint 1: Each middleware wraps a handler of type Callable[P, T]. It intercepts the call, does its work, and calls the inner handler.
Hint 2: Use functools.wraps to preserve names. Chain middleware by nesting wrappers: apply(m3, apply(m2, apply(m1, handler))).
Implement a type-safe partial_apply using Concatenate that binds the first argument of a function and returns a new callable with the remaining parameters.
import functools
from typing import TypeVar, Callable
try:
from typing import ParamSpec, Concatenate
except ImportError:
from typing_extensions import ParamSpec, Concatenate
P = ParamSpec("P")
T = TypeVar("T")
A = TypeVar("A")
def partial_apply(
func: Callable[Concatenate[A, P], T],
first_arg: A,
) -> Callable[P, T]:
@functools.wraps(func)
def partial(*args: P.args, **kwargs: P.kwargs) -> T:
return func(first_arg, *args, **kwargs)
return partial
def add(a: int, b: int) -> int:
return a + b
def greet(greeting: str, name: str) -> str:
return f"{greeting}, {name}!"
add_5 = partial_apply(add, 5)
greet_hello = partial_apply(greet, "Hello")
print(f"add_5(10) = {add_5(10)}")
print(f"add_5(20) = {add_5(20)}")
print(f"greet_hello('Alice') = {greet_hello('Alice')}")Expected Output
add_5(10) = 15
add_5(20) = 25
greet_hello('Alice') = Hello, Alice!Hints
Hint 1: partial_apply(func, first_arg) returns a Callable[P, T] where P is the original parameters minus the first argument.
Hint 2: Use Concatenate[FirstArg, P] to describe the original function signature and P for the returned partial.
Use Concatenate to build a transactional decorator that injects a Transaction object as the first argument to the decorated function, handling commit/rollback automatically.
import functools
import uuid
from typing import TypeVar, Callable
try:
from typing import ParamSpec, Concatenate
except ImportError:
from typing_extensions import ParamSpec, Concatenate
P = ParamSpec("P")
T = TypeVar("T")
class Transaction:
def __init__(self) -> None:
self.id = f"tx-{uuid.uuid4().hex[:3]}"
def begin(self) -> None:
print(f"[TXN] begin transaction {self.id}")
def commit(self) -> None:
print(f"[TXN] commit transaction {self.id}")
def rollback(self, reason: str) -> None:
print(f"[TXN] rollback transaction {self.id}: {reason}")
def transactional(
func: Callable[Concatenate[Transaction, P], T]
) -> Callable[P, T]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
txn = Transaction()
txn.begin()
try:
result = func(txn, *args, **kwargs)
txn.commit()
return result
except Exception as e:
txn.rollback(str(e))
raise
return wrapper
@transactional
def transfer(txn: Transaction, amount: float, from_acc: str, to_acc: str) -> str:
if amount > 50 and from_acc == "Bob":
raise ValueError("Insufficient funds")
return f"Transfer of {amount} from {from_acc} to {to_acc} completed"
result = transfer(100, "Alice", "Bob")
print(f"transfer result: {result}")
try:
transfer(100, "Bob", "Alice")
except ValueError:
passExpected Output
[TXN] begin transaction tx-001
transfer result: Transfer of 100 from Alice to Bob completed
[TXN] commit transaction tx-001
[TXN] begin transaction tx-002
[TXN] rollback transaction tx-002: Insufficient fundsHints
Hint 1: Concatenate[Transaction, P] means the wrapped function receives a Transaction as its first argument (injected by the decorator).
Hint 2: The decorator creates a Transaction, calls begin(), invokes the function, and calls commit() on success or rollback() on exception.
Build a typed event bus where handlers for a specific event all share the same ParamSpec. Emitting the event calls all handlers with the correct arguments.
import functools
from typing import TypeVar, Callable, List, Any
try:
from typing import ParamSpec
except ImportError:
from typing_extensions import ParamSpec
P = ParamSpec("P")
T = TypeVar("T")
class EventBus:
def __init__(self) -> None:
self._handlers: List[Callable[..., None]] = []
self._event_count = 0
def on(self, func: Callable[P, None]) -> Callable[P, None]:
self._handlers.append(func)
return func
def emit(self, *args: Any, **kwargs: Any) -> None:
self._event_count += 1
for handler in self._handlers:
handler(*args, **kwargs)
@property
def event_count(self) -> int:
return self._event_count
login_bus = EventBus()
@login_bus.on
def audit_handler(user: str, ip: str) -> None:
print(f"Handler 1: user_login event, user={user}, ip={ip}")
@login_bus.on
def metrics_handler(user: str, ip: str) -> None:
print(f"Handler 2: user_login event, user={user}")
login_bus.emit(user="alice", ip="192.168.1.1")
login_bus.emit(user="alice", ip="192.168.1.1")
login_bus.emit(user="bob", ip="10.0.0.1")
print(f"Total events processed: {login_bus.event_count}")Expected Output
Handler 1: user_login event, user=alice, ip=192.168.1.1
Handler 2: user_login event, user=alice
Handler 1: user_login event, user=bob, ip=10.0.0.1
Total events processed: 3Hints
Hint 1: Use P = ParamSpec("P") to define the handler signature. EventBus[P] stores handlers typed as Callable[P, None].
Hint 2: emit(**kwargs) calls each handler with the kwargs. Type safety ensures all handlers accept the same parameter spec.