Dekoratorlar va kontekst menejerlari

def wrapper_factory(fn):
    def wrapper(*args, **kwargs):
        print(f"→ {fn.__name__} chaqirildi")
        result = fn(*args, **kwargs)
        print(f"← {fn.__name__} tugadi")
        return result
    return wrapper
 
 
def salom(ism):
    return f"Salom, {ism}"
 
 
salom = wrapper_factory(salom)    # funksiyani "o'rab" qayta tayinladik
salom("Aziz")

→ salom chaqirildi
← salom tugadi

@wrapper_factory
def salom(ism):
    return f"Salom, {ism}"
 
# aynan shu bilan bir xil:
# def salom(ism): ...
# salom = wrapper_factory(salom)

salom.__name__      # 'wrapper' — asl nom yo'qoldi!
salom.__doc__       # None — docstring yo'qoldi

import functools
import time
 
 
def timer(fn):
    @functools.wraps(fn)                 # nom, docstring, imzo va h.k. ko'chiriladi
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            print(f"{fn.__name__}: {elapsed:.3f}s")
    return wrapper
 
 
@timer
def embed_batch(texts):
    """Matnlarni vektorlarga aylantiradi."""
    time.sleep(0.1)
    return [[0.0] * 4 for _ in texts]
 
 
embed_batch(["a", "b"])          # embed_batch: 0.105s
embed_batch.__name__             # 'embed_batch'
embed_batch.__wrapped__          # asl (o'ralmagan) funksiya — testlarda foydali

import functools
 
 
def my_decorator(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        # ... oldin
        result = fn(*args, **kwargs)
        # ... keyin
        return result
    return wrapper

import functools
import logging
import random
import time
 
log = logging.getLogger(__name__)
 
 
def retry(times: int = 3, exceptions: tuple[type[Exception], ...] = (Exception,),
          base_delay: float = 1.0):
    """Xato bo'lsa qayta urinadi: exponential backoff + jitter."""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    if attempt == times:
                        raise
                    delay = base_delay * 2 ** (attempt - 1) * random.uniform(0.5, 1.5)
                    log.warning("%s: %d-urinish muvaffaqiyatsiz (%r), %.1fs kutamiz",
                                fn.__name__, attempt, e, delay)
                    time.sleep(delay)
        return wrapper
    return decorator
 
 
@retry(times=5, exceptions=(ConnectionError, TimeoutError))
def fetch_embeddings(texts):
    ...

@timer
@retry(times=3)
def call_api(): ...
 
# aynan: call_api = timer(retry(times=3)(call_api))

import inspect
 
TOOLS: dict[str, dict] = {}
 
 
def tool(fn):
    """Funksiyani LLM agent uchun asbob sifatida ro'yxatga oladi."""
    sig = inspect.signature(fn)
    TOOLS[fn.__name__] = {
        "fn": fn,
        "description": inspect.getdoc(fn) or "",
        "params": list(sig.parameters),
    }
    return fn                      # funksiya o'zgarmaydi
 
 
@tool
def get_weather(city: str) -> str:
    """Shahar uchun joriy ob-havoni qaytaradi."""
    return f"{city}: +18°C, quyoshli"
 
 
@tool
def calculator(expression: str) -> str:
    """Oddiy arifmetik ifodani hisoblaydi."""
    ...
 
 
print(list(TOOLS))                          # ['get_weather', 'calculator']
print(TOOLS["get_weather"]["description"])   # Shahar uchun joriy ob-havoni qaytaradi.

with open("data.txt", encoding="utf-8") as f:   # kirish: fayl ochiladi
    data = f.read()
# chiqish: fayl yopiladi — hatto read() xato bergan bo'lsa ham

import time
 
 
class Timer:
    def __init__(self, name: str = "blok") -> None:
        self.name = name
 
    def __enter__(self) -> "Timer":
        self.start = time.perf_counter()
        return self                         # `as` dan keyingi nomga shu bog'lanadi
 
    def __exit__(self, exc_type, exc, tb) -> bool:
        self.elapsed = time.perf_counter() - self.start
        status = "xato bilan" if exc_type else "muvaffaqiyatli"
        print(f"{self.name}: {self.elapsed:.3f}s ({status})")
        return False                        # False — xato bo'lsa, u yuqoriga ko'tarilsin
 
 
with Timer("tokenizatsiya") as t:
    tokens = [s.split() for s in ["a b c"] * 100_000]
print(t.elapsed)

import time
from contextlib import contextmanager
 
 
@contextmanager
def timed(name: str):
    start = time.perf_counter()
    try:
        yield                          # ← with bloki shu yerda bajariladi
    finally:
        print(f"{name}: {time.perf_counter() - start:.2f}s")
 
 
with timed("ma'lumot yuklash"):
    time.sleep(0.05)

import os
from contextlib import contextmanager
 
 
@contextmanager
def temp_env(**overrides: str):
    old = {k: os.environ.get(k) for k in overrides}
    os.environ.update(overrides)
    try:
        yield
    finally:
        for k, v in old.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v
 
 
with temp_env(MODEL="claude-sonnet-5-5", DEBUG="1"):
    print(os.environ["MODEL"])        # claude-sonnet-5-5
print(os.environ.get("MODEL"))        # None — avvalgi holat tiklandi

from contextlib import contextmanager
 
 
@contextmanager
def evaluating(model):
    """Model'ni vaqtincha eval rejimiga o'tkazadi, keyin avvalgi rejimni tiklaydi."""
    was_training = model.training
    model.eval()
    try:
        yield model
    finally:
        if was_training:
            model.train()
 
# with evaluating(model), torch.no_grad():
#     val_loss = compute_loss(model, val_loader)

from contextlib import suppress, ExitStack, nullcontext, closing
 
# suppress — ma'lum xatoni ataylab e'tiborsiz qoldirish
with suppress(FileNotFoundError):
    os.remove("tmp.lock")              # yo'q bo'lsa — mayli
 
# ExitStack — oldindan noma'lum sondagi kontekst menejerlar
with ExitStack() as stack:
    files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
    ...                                # chiqishda hammasi yopiladi
 
# nullcontext — "hech narsa qilmaydigan" menejer (shartli holatlar uchun)
ctx = timed("debug") if debug else nullcontext()
with ctx:
    run()

with open("in.txt", encoding="utf-8") as src, open("out.txt", "w", encoding="utf-8") as dst:
    dst.write(src.read().upper())
 
with (                                 # Python 3.10+: qavslar bilan ko'p qatorli
    open("a.txt", encoding="utf-8") as a,
    open("b.txt", encoding="utf-8") as b,
):
    ...

import functools
import json
import logging
import time
from contextlib import contextmanager
 
log = logging.getLogger("llm")
 
 
@contextmanager
def span(name: str, **attrs):
    """Bitta amalni kuzatadi va natijani JSON qatori sifatida logga yozadi."""
    record = {"span": name, **attrs}
    start = time.perf_counter()
    try:
        yield record                   # chaqiruvchi record'ga ma'lumot qo'shishi mumkin
        record["status"] = "ok"
    except Exception as e:
        record["status"] = "error"
        record["error"] = repr(e)
        raise
    finally:
        record["ms"] = round((time.perf_counter() - start) * 1000, 1)
        log.info(json.dumps(record, ensure_ascii=False))
 
 
def traced(fn):
    """Funksiyaning har bir chaqiruvini span ichida bajaradi."""
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        with span(fn.__name__) as rec:
            result = fn(*args, **kwargs)
            if isinstance(result, dict) and "usage" in result:
                rec["tokens"] = result["usage"]
            return result
    return wrapper
 
 
@traced
def fake_llm(prompt: str) -> dict:
    time.sleep(0.02)
    return {"text": prompt.upper(), "usage": {"input": len(prompt), "output": len(prompt)}}
 
 
logging.basicConfig(level=logging.INFO, format="%(message)s")
fake_llm("salom")
# {"span": "fake_llm", "tokens": {"input": 5, "output": 5}, "status": "ok", "ms": 20.3}