Closures & Lexical Scope in Real Projects

x = 10  # Global scope

def outer():
    y = 20  # outer() scope
    
    def inner():
        # inner() can see BOTH x and y!
        print(f"x = {x}, y = {y}")
    
    return inner  # Return the inner function

func = outer()  # outer() finishes, but...
func()  # inner() STILL remembers y! → prints "x = 10, y = 20"

def make_multiplier(factor):
    # 'factor' will be "packed" into the closure
    def multiply(x):
        return x * factor  # Uses 'factor' from outer scope
    return multiply

times_10 = make_multiplier(10)  # factor=10 is captured
times_5 = make_multiplier(5)    # factor=5 is captured separately

print(times_10(3))  # 30 (uses its own factor=10)
print(times_5(3))   # 15 (uses its own factor=5)

# You can actually SEE the closure!
print(times_10.__closure__)  # Shows the cell objects
print(t
...

def counter():
    count = 0  # This variable lives in the closure
    
    def increment():
        nonlocal count  # "I want to MODIFY the outer 'count'"
        count += 1
        return count
    
    return increment

# Create two SEPARATE counters
counter_a = counter()
counter_b = counter()

print(counter_a())  # 1 - counter_a's count
print(counter_a())  # 2 - counter_a's count
print(counter_b())  # 1 - counter_b has its OWN count!
print(counter_a())  # 3 - counter_a continues

def make_logger(level):
    def log(message):
        print(f"[{level}] {message}")
    return log

info = make_logger("INFO")
error = make_logger("ERROR")

info("Server started")
error("Connection failed")

def require_auth(role):
    def decorator(func):
        def wrapper(user, *args):
            if user.get("role") != role:
                raise PermissionError("Access denied")
            return func(user, *args)
        return wrapper
    return decorator

@require_auth("admin")
def delete_user(user, user_id):
    print(f"Deleting user {user_id}")

admin = {"role": "admin"}
delete_user(admin, 123)

def range_validator(min_val, max_val):
    def validate(value):
        return min_val <= value <= max_val
    return validate

age_valid = range_validator(18, 100)
print(age_valid(25))  # True
print(age_valid(5))   # False

def memoize(func):
    cache = {}
    def wrapper(x):
        if x not in cache:
            cache[x] = func(x)
        return cache[x]
    return wrapper

@memoize
def expensive_calc(n):
    print("Computing...")
    return n * n

print(expensive_calc(5))  # Computing... 25
print(expensive_calc(5))  # 25 (cached)

import time

def rate_limiter(max_calls, period):
    calls = []
    def decorator(func):
        def wrapper(*args, **kwargs):
            nonlocal calls
            now = time.time()
            calls = [t for t in calls if now - t < period]
            if len(calls) >= max_calls:
                raise Exception("Rate limit exceeded")
            calls.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limiter(3, 5)
def api_call():
    print
...

def query_builder(table):
    def query(**filters):
        conditions = " AND ".join(f"{k}='{v}'" for k, v in filters.items())
        return f"SELECT * FROM {table} WHERE {conditions}"
    return query

users = query_builder("users")
print(users(age=25, active=True))

def compose(f, g):
    def composed(x):
        return f(g(x))
    return composed

def double(x): return x * 2
def add_5(x): return x + 5

pipeline = compose(double, add_5)
print(pipeline(10))  # (10+5)*2 = 30

# CLOSURE approach - simple, lightweight
def make_counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc

# CLASS approach - more features
class Counter:
    def __init__(self):
        self.count = 0
    
    def inc(self):
        self.count += 1
        return self.count
    
    def reset(self):  # Easy to add methods!
        self.count = 0

# Both work, choose based on needs!
closure_counter = make_counter()
class_counter = Cou
...

def make_adder(x):
    def add(y):
        return x + y
    return add

add_5 = make_adder(5)
print(add_5.__closure__)
print(add_5.__closure__[0].cell_contents)

def counter_broken():
    count = 0
    def inc():
        # count += 1  # ❌ UnboundLocalError!
        # Python thinks 'count' is a NEW local variable
        pass
    return inc

# The fix:
def counter_fixed():
    count = 0
    def inc():
        nonlocal count  # ✅ "Use the outer count!"
        count += 1
        return count
    return inc

c = counter_fixed()
print(c())  # 1
print(c())  # 2

# ❌ WRONG - all functions capture the SAME 'i'
funcs_bad = []
for i in range(3):
    funcs_bad.append(lambda: i)  # All will print 2!

print([f() for f in funcs_bad])  # [2, 2, 2] - oops!

# ✅ FIX - capture 'i' as a default argument
funcs_good = []
for i in range(3):
    funcs_good.append(lambda i=i: i)  # Each captures its own 'i'

print([f() for f in funcs_good])  # [0, 1, 2] - correct!

def event_system():
    handlers = {}
    
    def on(event, callback):
        if event not in handlers:
            handlers[event] = []
        handlers[event].append(callback)
    
    def emit(event, data):
        if event in handlers:
            for callback in handlers[event]:
                callback(data)
    
    return on, emit

subscribe, publish = event_system()

subscribe("login", lambda u: print(f"Welcome {u}"))
subscribe("login", lambda u: print(f"Logging {u}"))

publish("login
...

def provide_db(connector):
    def run_query(q):
        db = connector()
        return db.execute(q) if hasattr(db, 'execute') else f"Query: {q} on {db}"
    return run_query

def sqlite_connector():
    return "SQLite connection"

def mysql_connector():
    return "MySQL connection"

query_local = provide_db(sqlite_connector)
query_prod = provide_db(mysql_connector)

print(query_local("SELECT * FROM users"))
print(query_prod("SELECT * FROM orders"))

def middleware(next_handler):
    def wrapper(request):
        print("Before")
        response = next_handler(request)
        print("After")
        return response
    return wrapper

def authenticate(next_handler):
    def wrapper(request):
        if not request.get("auth"):
            return "Unauthorized"
        return next_handler(request)
    return wrapper

def endpoint(request):
    return f"Hello {request.get('user', 'Guest')}"

# Chain multiple
handler = middleware(authenticate(e
...

import time
import random

def retry(times, delay=1):
    def decorator(func):
        def wrapper(*a, **kw):
            for i in range(times):
                try:
                    return func(*a, **kw)
                except Exception as e:
                    print(f"Attempt {i+1} failed: {e}")
                    time.sleep(delay)
            raise Exception("Failed after retries")
        return wrapper
    return decorator

@retry(3, delay=0.1)
def unstable_api():
    if random.random(
...

import time

def timed_cache(seconds):
    def decorator(func):
        cache = {}
        timestamps = {}

        def wrapper(*a):
            if a in cache and time.time() - timestamps[a] < seconds:
                print("Cache hit!")
                return cache[a]
            result = func(*a)
            cache[a] = result
            timestamps[a] = time.time()
            return result

        return wrapper
    return decorator

@timed_cache(5)
def expensive_calc(x):
    print("Computin
...

def feature_flag(enabled):
    def decorator(func):
        def wrapper(*a, **kw):
            if enabled:
                return func(*a, **kw)
            return "Feature disabled"
        return wrapper
    return decorator

@feature_flag(False)
def new_checkout():
    return "New checkout flow!"

@feature_flag(True)
def new_dashboard():
    return "New dashboard!"

print(new_checkout())  # Feature disabled
print(new_dashboard())  # New dashboard!

def tracker(event_name):
    def decorator(func):
        def wrapper(*a, **kw):
            print(f"[TRACK] {event_name}")
            return func(*a, **kw)
        return wrapper
    return decorator

@tracker("user_signup")
def register_user(email):
    print(f"Registering {email}")

register_user("alice@example.com")

COMMANDS = {}

def command(name):
    def decorator(func):
        COMMANDS[name] = func
        return func
    return decorator

@command("hello")
def hello():
    print("Hello world!")

@command("add")
def add():
    print(1 + 1)

# Execute
COMMANDS["hello"]()
COMMANDS["add"]()
print("Available commands:", list(COMMANDS.keys()))

def pipeline(*steps):
    def run(value):
        for step in steps:
            value = step(value)
        return value
    return run

def trim(x): return x.strip()
def lower(x): return x.lower()
def reverse(x): return x[::-1]

clean = pipeline(trim, lower, reverse)
print(clean("  HELLO  "))  # olleh

def managed_resource(resource):
    def wrapper(func):
        def run(*a, **kw):
            r = resource()
            try:
                return func(r, *a, **kw)
            finally:
                r.close()
        return run
    return wrapper

class MockResource:
    def close(self):
        print("Resource closed")

@managed_resource(MockResource)
def use_resource(r):
    print("Using resource")

use_resource()

def fixture(setup):
    def decorator(func):
        def wrapper():
            env = setup()
            return func(env)
        return wrapper
    return decorator

def create_env():
    return {"db": "mock_db"}

@fixture(create_env)
def test_user(env):
    assert env["db"] == "mock_db"
    print("Test passed!")

test_user()

def on_click(message):
    def handler():
        print(message)
    return handler

# Bind in games
button_handler = on_click("Start!")
button_handler()  # Start!

pause_handler = on_click("Paused")
pause_handler()

import inspect

def make_adder(x):
    def add(y):
        return x + y
    return add

adder = make_adder(10)
print(adder.__closure__)
print(inspect.getclosurevars(adder))

funcs = [lambda: i for i in range(3)]
print([f() for f in funcs])  # [2, 2, 2] - not [0,1,2]

funcs = [(lambda x: lambda: x)(i) for i in range(3)]
print([f() for f in funcs])  # [0, 1, 2]

import asyncio

def async_retry(times):
    def decorator(func):
        async def wrapper(*a, **kw):
            for _ in range(times):
                try:
                    return await func(*a, **kw)
                except Exception:
                    await asyncio.sleep(0.1)
            raise Exception("Failed after retries")
        return wrapper
    return decorator

@async_retry(3)
async def unstable_fetch():
    return "Data fetched"

# In async context:
# result = await unstable_f
...

def state_machine(initial):
    state = initial

    def transition(new_state):
        nonlocal state
        state = new_state
        return state
    
    def current():
        return state
    
    return current, transition

get_state, set_state = state_machine("IDLE")

set_state("RUNNING")
set_state("PAUSED")
print(get_state())  # PAUSED

def scaler(mean, std):
    def transform(x):
        return (x - mean) / std
    return transform

def clip(min_val, max_val):
    def transform(x):
        return max(min_val, min(x, max_val))
    return transform

# Chain transformations
normalize = scaler(120, 30)
clamp = clip(0, 255)

data = 150
result = clamp(normalize(data))
print(result)

def token_rule(pattern, action):
    def processor(token):
        if pattern(token):
            return action(token)
        return token
    return processor

is_number = lambda t: t.isdigit()
to_int = lambda t: int(t)

process_token = token_rule(is_number, to_int)
print(process_token("123"))  # 123
print(process_token("abc"))  # abc

ROUTES = {}

def route(path):
    def decorator(func):
        ROUTES[path] = func
        return func
    return decorator

@route("/hello")
def hello():
    return {"msg": "world"}

@route("/users")
def users():
    return {"users": []}

print(ROUTES["/hello"]())

import time

def schedule(interval):
    def decorator(func):
        last_run = [0]  # Use list for mutable state
        def wrapper():
            if time.time() - last_run[0] >= interval:
                last_run[0] = time.time()
                return func()
            return "Too soon!"
        return wrapper
    return decorator

@schedule(1)
def update_prices():
    return "Updating prices..."

print(update_prices())
print(update_prices())  # Too soon!

def field(name):
    def getter(obj):
        return obj[name]
    return getter

user_name = field("name")
user_age = field("age")

user = {"name": "Alice", "age": 25}
print(user_name(user))  # Alice
print(user_age(user))   # 25

def use_state(init):
    state = [init]  # Use list for mutable state
    def get(): return state[0]
    def set(v):
        state[0] = v
    return get, set

get_count, set_count = use_state(0)
set_count(5)
print(get_count())  # 5
set_count(10)
print(get_count())  # 10

def require_role(role):
    def decorator(func):
        def wrapper(user, *a):
            if user.get("role") != role:
                raise PermissionError("Forbidden")
            return func(user, *a)
        return wrapper
    return decorator

@require_role("admin")
def delete_user(user, target_id):
    print("Deleting", target_id)

admin = {"role": "admin"}
delete_user(admin, 123)

def message_queue():
    queue = []
    
    def publish(msg):
        queue.append(msg)
    
    def consume():
        if queue:
            return queue.pop(0)
    
    return publish, consume

send, receive = message_queue()
send("Hello")
send("World")
print(receive())  # Hello
print(receive())  # World

def polynomial(a, b, c):
    def f(x):
        return a*x*x + b*x + c
    return f

quadratic = polynomial(1, -3, 2)
print(quadratic(0))  # 2
print(quadratic(1))  # 0
print(quadratic(2))  # 0

def partial(func, *preset):
    def wrapper(*a):
        return func(*preset, *a)
    return wrapper

def add(a, b, c):
    return a + b + c

add_5 = partial(add, 5)
print(add_5(10, 2))  # 17

def stack(*middlewares):
    def build(handler):
        for m in reversed(middlewares):
            handler = m(handler)
        return handler
    return build

def logger(h):
    def wrapper(x):
        print("Logging...")
        return h(x)
    return wrapper

def auth(h):
    def wrapper(x):
        print("Authenticating...")
        return h(x)
    return wrapper

# Build stack
def endpoint(x): return x * 2
final = stack(logger, auth)(endpoint)
print(final(5))

def watch():
    listeners = []
    
    def subscribe(fn):
        listeners.append(fn)
    
    def emit(value):
        for fn in listeners:
            fn(value)
    
    return subscribe, emit

on_change, notify = watch()

on_change(lambda v: print(f"Value changed: {v}"))
on_change(lambda v: print(f"Logging: {v}"))

notify(42)

import time

def memoize_with_ttl(seconds):
    cache = {}
    timestamps = {}
    
    def decorator(func):
        def wrapper(x):
            if x in cache and time.time() - timestamps[x] < seconds:
                return cache[x]
            result = func(x)
            cache[x] = result
            timestamps[x] = time.time()
            return result
        return wrapper
    return decorator

@memoize_with_ttl(5)
def expensive_calc(x):
    print("Computing...")
    return x * x

print(ex
...