Python 3.15: ленивые импорты, frozendict, sentinel, JIT и жизнь без GIL на реальном FastAPI

uv run 3_12.py                                   # Python 3.12
uv run --python 3.15+gil 3_15.py                 # обычный 3.15
PYTHON_JIT=1 uv run --python 3.15+gil 3_15.py    # 3.15 с JIT
uv run --python 3.15t 3_15.py                    # 3.15 без GIL

lazy import pandas as pd   # ничего не загружает
lazy import uvicorn

df = pd.DataFrame(...)     # вот здесь pandas реально загрузится

# 3.12: два for, читается справа налево
all_tx = [tx for batch in batches for tx in batch]

# 3.15
all_tx = [*batch for batch in batches]
merged = {**stats for stats in partial_stats}   # со словарями так же

@lru_cache
def analyze_cached(params: frozendict):
    return analyze(params)

analyze_cached(frozendict(threshold=4900, top=5))   # повтор — 1–2 мкс из кэша

MISSING = sentinel("MISSING")       # 3.15: в логах так и будет — MISSING

def update_account(acc_id, phone=MISSING, email=MISSING):
    if phone is not MISSING:
        ...

# /// script
# requires-python = ">=3.15"
# dependencies = ["fastapi", "uvicorn", "pandas"]
# ///
"""Сервис аналитики транзакций на Python 3.15.

    uv run --python 3.15+gil 3_15.py               замер: 1 запрос, 8 запросов в 8 потоках, кэш
    PYTHON_JIT=1 uv run --python 3.15+gil 3_15.py  то же с JIT
    uv run --python 3.15t 3_15.py                  сборка без GIL (free-threading)
    uv run --python 3.15t 3_15.py serve            сервер на http://127.0.0.1:8000/docs

+gil обязателен: после запуска на 3.15t просто «3.15» молча берёт ту же сборку без GIL.
"""
import time
START = time.perf_counter()

import random
import sys
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache

# 1. ЛЕНИВЫЕ ИМПОРТЫ (PEP 810)
# pandas грузится сотни миллисекунд, а нужен только в редкой ручке /report.
# `lazy import` ничего не загружает сейчас: модуль подтянется при первом обращении к pd.
lazy import pandas as pd
lazy import uvicorn  # нужен только в режиме serve

from fastapi import FastAPI
from pydantic import BaseModel

print(f"импорты: {(time.perf_counter() - START) * 1000:.0f} мс")

app = FastAPI()


# 2. РАСПАКОВКА В COMPREHENSIONS (PEP 798)
# Транзакции приходят пачками по дням: (клиент, сумма). Склеиваем в один список.
random.seed(42)
batches = [
    [(random.randrange(1000), random.uniform(1, 5000)) for _ in range(100_000)]
    for day in range(10)
]
TXS = [*batch for batch in batches]  # со словарями так же: {**d for d in dicts}


# 3. JIT + FREE-THREADING: чистый Python-цикл, никакого numpy.
# numpy написан на C, его не ускорит ни JIT, ни отсутствие GIL.
def risk_score(amount, threshold):
    # чем ближе к порогу, тем подозрительнее; «круглые» суммы — ещё подозрительнее
    score = amount / threshold
    if amount % 100 < 1:
        score += 0.5
    return score


def analyze(params):
    totals = {}
    suspicious = 0
    for client, amount in TXS:
        totals[client] = totals.get(client, 0) + amount
        if risk_score(amount, params["threshold"]) > 1:
            suspicious += 1
    top = sorted(totals.items(), key=lambda kv: kv[1], reverse=True)[: params["top"]]
    return {"suspicious": suspicious, "top": top}


# 4. FROZENDICT (PEP 814)
# Обычный dict изменяемый, поэтому нехэшируемый: lru_cache падает
# с "unhashable type: 'dict'". frozendict неизменяемый — его можно хэшировать.
@lru_cache
def analyze_cached(params: frozendict):
    return analyze(params)


@app.get("/analyze")
def analyze_endpoint(threshold: float = 4900, top: int = 5):
    # def, а не async def: FastAPI запускает такие ручки в пуле потоков.
    # С GIL потоки считают по очереди, без GIL — на разных ядрах.
    return analyze_cached(frozendict(threshold=threshold, top=top))


# 5. SENTINEL (PEP 661)
# {"phone": null} — «удали телефон», поля нет — «не трогай». None занят, нужен маркер.
MISSING = sentinel("MISSING")  # repr в логах: MISSING
ACCOUNTS = {1: {"phone": "+998901234567", "email": "student@sudoteach.com"}}


def update_account(acc_id, phone=MISSING, email=MISSING):
    print(f"update_account({acc_id}, phone={phone!r}, email={email!r})")
    acc = ACCOUNTS[acc_id]
    if phone is not MISSING:
        acc["phone"] = phone
    if email is not MISSING:
        acc["email"] = email
    return acc


class AccountPatch(BaseModel):
    phone: str | None = None
    email: str | None = None


@app.patch("/accounts/{acc_id}")
def patch_account(acc_id: int, body: AccountPatch):
    # exclude_unset: в словарь попадут только поля, которые клиент реально прислал
    return update_account(acc_id, **body.model_dump(exclude_unset=True))


@app.get("/report")
def report():
    # Только здесь pandas реально загрузится, поэтому первый вызов дольше остальных.
    df = pd.DataFrame(TXS[:10_000], columns=["client", "amount"])
    return df["amount"].describe().round(2).to_dict()


def bench():
    print(f"Python {sys.version.split()[0]} | GIL: {sys._is_gil_enabled()} | JIT: {sys._jit.is_enabled()}")
    params = frozendict(threshold=4900, top=5)

    t = time.perf_counter()
    analyze(params)
    print(f"1 запрос:                {time.perf_counter() - t:.2f} с")

    t = time.perf_counter()
    with ThreadPoolExecutor(8) as pool:
        list(pool.map(analyze, [params] * 8))
    print(f"8 запросов в 8 потоках:  {time.perf_counter() - t:.2f} с")

    analyze_cached(params)
    t = time.perf_counter()
    analyze_cached(params)
    print(f"повтор из кэша:          {(time.perf_counter() - t) * 1e6:.0f} мкс")


if __name__ == "__main__":
    if "serve" in sys.argv:
        uvicorn.run(app)  # вот здесь uvicorn и загрузится
    else:
        bench()