Data Science va sun'iy intellekt/MLOps va deploy7/14-dars36 daqiqa

Mundarija (26)

27.7-dars: FastAPI bilan model API

27-QISM — MLOPS VA DEPLOY · 7-dars

1. Kirish va motivatsiya

Oldingi darsda 27.6-bob modelni paketga aylantirdik va unga bitta Predictor interfeysini berdik. Endi savol: bu Predictor ni boshqalar qanday ishlatadi? Mobil ilova Kotlin da, veb-sayt JavaScript da, CRM tizimi Java da yozilgan — ularning hech biri Python obyektini import qila olmaydi. Umumiy til kerak.

Bu til — HTTP va JSON. Model tarmoq orqali so'rov qabul qiladigan xizmatga aylanadi: mijoz POST /predict ga JSON yuboradi, javobda JSON oladi. Bu model API deyiladi. Xizmatning ichida Python, sklearn yoki torch borligini mijoz bilmaydi va bilishi shart ham emas — u faqat shartnomani (qaysi maydonlar, qaysi turlar, qaysi javob) biladi.

Python da model API uchun eng qulay vositalardan biri — FastAPI. Uning kuchli tomoni: Python turlari (type hints) va pydantic sxemalari orqali siz bir marta yozgan sxemadan uch narsa avtomatik chiqadi:

kiruvchi ma'lumotni tekshirish (noto'g'ri turga — tushunarli 422 javob);
hujjat — OpenAPI sxemasi va interaktiv /docs sahifasi;
muharrirda avtomatik to'ldirish va tur tekshiruvi.

Real vaziyat. Jamoa modelni Flask bilan tez "o'rab" qo'ydi: request.get_json() dan maydonlarni oladi va to'g'ridan-to'g'ri modelga beradi. Bir kuni mobil ilova yangilanishida tolov maydoni boshqa birlikda yuborila boshladi va qiymatlar 100 marta katta keldi. API hech qanday xato qaytarmadi — model shunchaki hammaga "ketadi" deb bashorat qildi, marketing esa bir hafta davomida sodiq mijozlarga chegirma yubordi. Maydon uchun le=1000 chegarasi bo'lganida birinchi so'rovning o'zi 422 qaytarardi.

Bu darsda Predictor ni FastAPI orqali taqdim etamiz: pydantic sxemalar, to'rtta endpoint, modelni bir marta yuklash, xato javoblar va OpenAPI — hammasini TestClient bilan sinab.

Bu darsda:

Nega API: onlayn va batch bashorat
REST asoslari: endpoint, metod, status kodlar, JSON
FastAPI va pydantic sxemalar: Field chegaralari, Enum, ixtiyoriy maydonlar
Modelni bir marta yuklash: lifespan va Depends
/predict, /predict_batch, /health, /version
Xato javoblar: 422, 400, 500, 503
Avtomatik OpenAPI sxemasi
TestClient bilan sinash, uvicorn bilan ishga tushirish
Flask bilan qisqa taqqoslash
Tuzoqlar

ℹ Misollar real FastAPI 0.141, pydantic 2.13, Flask 3.1 bilan (Python 3.14). Server ishga tushirilmaydi va port ochilmaydi: barcha so'rovlar TestClient orqali jarayon ichida bajariladi.

2. Nazariya — chuqur tushuntirish

2.1. Nega API: onlayn va batch bashorat

text

                  ONLAYN (API)                    BATCH (rejali ish)
QACHON          so'rov kelganda, darhol          jadval bo'yicha (har kecha)
KIRISH          1 yoki bir necha qator           millionlab qator
KECHIKISH       millisekundlar muhim             soatlar ham mumkin
NATIJA          javobda qaytadi                  jadvalga / faylga yoziladi
MISOL           kredit arizasi, firibgarlik,     ertangi kun uchun barcha mijozlarning
                tavsiya sahifada                  ketish ehtimoli, oylik hisobot
MURAKKABLIK     yuqori: xizmat doim ishlashi,    past: skript + rejalashtiruvchi
                kechikish, yuklama, xavfsizlik

QOIDA: agar natija bir necha soat kechiksa ham bo'lsa - BATCH yetarli va arzon
       agar qaror foydalanuvchi kutib turganda kerak bo'lsa - ONLAYN (API)

Ko'p jamoalar "model = API" deb o'ylaydi va batch yetarli bo'lgan joyda murakkab xizmat quradi. Marketing uchun ketish ehtimoli har kecha hisoblansa yetarli — bu 27.3 dagi quvurning bir bosqichi. Kredit arizasi esa mijoz ekran oldida kutib turganda baholanadi — bu API.

Ko'pincha ikkalasi birga kerak bo'ladi, va shunda muhim qoida: bir xil Predictor 27.6-bob. Batch skript ham, API ham bitta paketni ishlatadi — aks holda bir mijoz uchun ikki xil javob chiqadi.

Avval so'rang: natija qanchalik tez kerak? Soatlar — batch; soniyalar — API.

2.2. REST asoslari

text

RESURS va ENDPOINT:
  /mijozlar           mijozlar to'plami
  /mijozlar/m1        bitta mijoz (yo'l parametri)
  /kvadrat?x=3        so'rov (query) parametri
  /predict            harakat (model API da odatiy)

METODLAR:
  GET     o'qish, tanasiz, xavfsiz va idempotent (takrorlash zararsiz)
  POST    yaratish / hisoblash, tanasi bor (JSON)
  PUT     to'liq almashtirish, idempotent
  DELETE  o'chirish, idempotent

STATUS KODLAR:
  2xx muvaffaqiyat   200 OK, 201 Created
  4xx MIJOZ xatosi   400 Bad Request (biznes qoidasi)
                     401 Unauthorized (kim ekaning noma'lum)
                     403 Forbidden (ma'lum, lekin ruxsat yo'q)
                     404 Not Found, 405 Method Not Allowed
                     422 Unprocessable Entity (sxema/validatsiya)
                     429 Too Many Requests 27.8-bob
  5xx SERVER xatosi  500 Internal Server Error (bizning aybimiz)
                     503 Service Unavailable (hali tayyor emas / band)

Status kod mijozga "nima qilish kerak" ni aytadi: 4xx — "so'rovingizni tuzating, takrorlash foydasiz"; 5xx — "biz tomonda muammo, keyinroq qayta urinib ko'ring". Shuning uchun validatsiya xatosiga 500 qaytarish — xato: mijoz behuda qayta urinadi.

Model API da bashorat POST /predict bilan so'raladi, garchi u hech narsa "yaratmasa" ham: kirish murakkab JSON (GET ning so'rov satriga sig'maydi), va ko'p proksi/keshlar GET javobini keshlaydi.

2.3. FastAPI: turlar — shartnoma

text

from fastapi import FastAPI
app = FastAPI(title="Mijoz ketishi API", version="1.0.0")

@app.get("/mijozlar/{mijoz_id}")          # yo'l parametri
def mijoz_ol(mijoz_id: str): ...

@app.get("/kvadrat")                        # x - majburiy query, aniqlik - ixtiyoriy
def kvadrat(x: float, aniqlik: int = 2): ...

@app.post("/predict", response_model=Bashorat)
def predict(mijoz: Mijoz): ...              # Mijoz - pydantic model -> JSON TANA

FASTAPI TURLARDAN NIMA OLADI:
  yo'l/query parametrlari  -> turga o'giradi (x="3" -> 3.0), bo'lmasa 422
  pydantic parametri       -> JSON tanani tekshiradi
  response_model           -> javobni ham tekshiradi va filtrlaydi
  hammasi                  -> OpenAPI sxemasi (/openapi.json, /docs)

Funksiya ichida siz allaqachon tekshirilgan obyekt bilan ishlaysiz: mijoz.oylar — butun son, 0 va 600 oralig'ida. Tekshiruv kodi biznes mantiq bilan aralashmaydi.

2.4. Pydantic sxemalar

text

class Tarif(str, Enum):
    oylik = "oylik"; yillik = "yillik"; ikki_yillik = "ikki_yillik"

class Mijoz(BaseModel):
    model_config = ConfigDict(extra="forbid")          # noma'lum maydon -> xato
    oylar: int = Field(ge=0, le=600)                   # 0 <= oylar <= 600
    tolov: float = Field(gt=0, le=1000)               # 0 < tolov <= 1000
    shikoyat: int | None = Field(default=None, ge=0)   # IXTIYORIY
    tarif: Tarif                                       # faqat ro'yxatdagilar
    hudud: str = Field(default="nomalum", min_length=2, max_length=40)

CHEGARALAR:  gt, ge, lt, le, min_length, max_length, pattern, multiple_of
IXTIYORIY:   T | None = None   (qiymat kelmasa - None)
             T = sukut         (kelmasa - sukut)
REJIM:       lax (sukut): "12" -> 12, 12.0 -> 12;  12.5 -> int uchun XATO
             strict:      "12" -> XATO (tur aynan mos bo'lishi kerak)
XATO:        ValidationError.errors() -> [{"loc", "type", "msg", "input"}, ...]
             BARCHA xatolar birdaniga

extra="forbid" model API uchun foydali: mijoz tolov_som deb yozsa (imlo xatosi), sukut bo'yicha pydantic uni jimgina e'tiborsiz qoldiradi va tolov yo'qligi haqida xato beradi — lekin mijoz hudud o'rniga region yuborsa va hudud ixtiyoriy bo'lsa, xato umuman chiqmaydi, sukut qiymati ishlatiladi. forbid bilan bunday imlo xatolari darhol ko'rinadi.

Sxema — API shartnomasi. Chegaralar biznes domenidan (27.6 dagi sxema bilan bir xil), ixtiyoriy maydonlar aniq belgilangan, noma'lum maydonlar rad etiladi.

2.5. Modelni bir marta yuklash: `lifespan`

text

YOMON:                                  YAXSHI:
@app.post("/predict")                   @asynccontextmanager
def predict(m: Mijoz):                  async def lifespan(app):
    model = joblib.load(...)  # HAR          app.state.predictor = Predictor(papka)  # BIR MARTA
    return model.predict(...) # so'rovda     yield                                   # xizmat ishlaydi
                                             app.state.predictor = None              # to'xtashda tozalash
  - har so'rov: disk + deserializatsiya
  - 100 so'rov/s -> 100 marta yuklash   app = FastAPI(lifespan=lifespan)

MODELNI ENDPOINTGA BERISH - Depends:
  def predictor_ol(request: Request):
      p = getattr(request.app.state, "predictor", None)
      if p is None: raise HTTPException(503, "model yuklanmagan")
      return p

  @app.post("/predict")
  def predict(mijoz: Mijoz, pr: Predictor = Depends(predictor_ol)): ...

lifespan — ilova ishga tushganda bir marta bajariladigan kod (yield gacha) va to'xtaganda bajariladigan kod (yield dan keyin). Modelni yuklash, ma'lumotlar bazasi ulanishi, keshni isitish — hammasi shu yerda. Depends esa endpointga kerakli obyektni beradi va testlarda uni almashtirish (app.dependency_overrides) imkonini beradi.

Model yuklanib bo'lmasa (xesh mos emas, nazorat namunalari yiqildi — 27.6), ilova ishga tushmasligi kerak — noto'g'ri model bilan ishlagandan ko'ra, umuman ishlamagan yaxshi. Orkestrator 27.10-bob buni ko'radi va eski versiyani qoldiradi.

2.6. To'rt endpoint

text

GET  /health         xizmat tirikmi va TAYYORmi?
                     200 {"holat": "ok"} | 503 {"holat": "tayyor emas"}
                     orkestrator (Kubernetes, 27.10) shu bilan tekshiradi:
                       liveness  - jarayon tirikmi (qayta ishga tushirish kerakmi)
                       readiness - so'rov qabul qila oladimi (model yuklanganmi)

GET  /version        {"model": "mijoz_ketishi", "model_versiyasi": "1.2.0",
                      "api_versiyasi": "1.0.0"}
                     incident paytida BIRINCHI savol: "prodda aynan nima?"

POST /predict        1 mijoz -> {"ehtimol": 0.83, "sinf": 1, "versiya": "1.2.0"}

POST /predict_batch  {"mijozlar": [...]} (1..100) -> {"natijalar": [...], "soni": N}
                     bitta model chaqiruvi - ko'p qator uchun samarali
                     hajm CHEGARALANGAN (27.8)

Har javobda versiya bor: mijoz tomonidagi loglar ham qaysi model javob berganini saqlaydi. Rollback 27.5-bob paytida bu bebaho.

2.7. Xato javoblar

text

KOD  QACHON                              KIM TUZATADI    MISOL
422  sxema buzilgan (FastAPI avtomatik)  mijoz           tolov = -5, tarif = "haftalik"
400  sxema to'g'ri, BIZNES qoidasi       mijoz           partiyada takroriy mijoz_id
     buzilgan (siz raise qilasiz)
404  bunday endpoint/resurs yo'q         mijoz           /predikt (imlo)
503  xizmat hali tayyor emas             kutish          model yuklanmoqda
500  kutilmagan xato - BIZNING xato      biz             model fayli buzilgan

500 JAVOBIDA:
  YO'Q:  traceback, fayl yo'llari, SQL so'rovi, kalitlar (hujumchi uchun xarita!)
  BOR:   {"xato": "ichki xato", "sorov_id": "..."} - to'liq tafsilot faqat LOGDA (27.8)

2.8. Avtomatik OpenAPI sxemasi

FastAPI barcha endpointlar, parametrlar, pydantic sxemalar va javob kodlaridan OpenAPI (avvalgi nomi Swagger) sxemasini yasaydi:

text

app.openapi()      -> Python lug'ati (JSON sxema)
GET /openapi.json  -> xuddi shu, HTTP orqali
GET /docs          -> interaktiv sahifa (Swagger UI): endpointni brauzerda sinash
GET /redoc         -> o'qish uchun chiroyli hujjat

FOYDASI:
  mijoz jamoalari hujjatni KODDAN oladi - eskirmaydi
  sxemadan mijoz kutubxonasi generatsiya qilinadi (openapi-generator)
  kontrakt testlari 27.8-bob - sxema o'zgarmaganini tekshirish

2.9. `TestClient` va import tuzog'i

TestClient ilovani jarayon ichida chaqiradi: haqiqiy tarmoq, port va server yo'q, lekin so'rov FastAPI ning butun zanjiridan (validatsiya, Depends, xato ishlovchilari, middleware) o'tadi. Shuning uchun testlar tez va deterministik.

text

with TestClient(app) as klient:        # 'with' - lifespan ISHLAYDI (model yuklanadi)
    r = klient.post("/predict", json={...})
    r.status_code, r.json(), r.headers

TestClient(app)                        # 'with' siz - lifespan ISHLAMAYDI
TestClient(app, raise_server_exceptions=False)
                                       # 500 ni istisno emas, JAVOB sifatida olish

Bu mashinadagi versiyalarda from fastapi.testclient import TestClient qatori StarletteDeprecationWarning chiqaradi (Starlette test mijozining httpx ga bog'liqligi haqida). Bu bizning kodimizga aloqasi yo'q ogohlantirish, lekin u stderr ni ifloslantiradi. Shuning uchun misollarda import shunday o'raladi:

python

import warnings

with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from fastapi.testclient import TestClient

Ogohlantirish faqat shu import uchun o'chiriladi — boshqa joylardagi ogohlantirishlar ko'rinaveradi.

2.10. Ishga tushirish: uvicorn

Haqiqiy xizmat ASGI server — odatda uvicorn — orqali ishga tushiriladi (bu mashinada uvicorn yo'q; buyruqlar faqat tanishish uchun):

bash

# ishlab chiqish: kod o'zgarsa qayta yuklanadi
uvicorn ilova:app --reload --port 8000

# ishlab chiqarish: bir necha jarayon, barcha interfeyslarda
uvicorn ilova:app --host 0.0.0.0 --port 8000 --workers 4

# tekshirish
curl -s localhost:8000/health
curl -s -X POST localhost:8000/predict -H "Content-Type: application/json" \
     -d '{"oylar": 3, "tolov": 95.5, "tarif": "oylik"}'

--workers 4 — to'rtta alohida jarayon, har birida model alohida yuklanadi (xotira 4 barobar). --reload faqat ishlab chiqishda. Konteynerga joylash — 27.9 mavzusi.

2.11. Flask bilan qisqa taqqoslash

text

                    FASTAPI                         FLASK
VALIDATSIYA         avtomatik (pydantic, turlar)    qo'lda (yoki pydantic ni qo'lda chaqirish)
XATO SXEMA          422, standart detail            o'zingiz belgilaysiz
HUJJAT              OpenAPI + /docs avtomatik       kengaytma kerak
ASINXRONLIK         async def tabiiy (ASGI)         asosan sinxron (WSGI)
YETUKLIK            yangiroq                        juda yetuk, ko'p kengaytma
TEST                TestClient                      app.test_client()

XULOSA: ikkalasi ham model API uchun yaraydi; muhimi - VALIDATSIYA bo'lishi.
        Flask da pydantic ni qo'lda chaqirsangiz, deyarli bir xil natija.

2.12. Tuzoqlar

Asosiy tuzoqlar: batch yetarli joyda API qurish; modelni har so'rovda yuklash; sxemasiz (dict) kirish; chegara va Enum siz maydonlar; extra maydonlarni jim e'tiborsiz qoldirish; validatsiya xatosiga 500 yoki 200 qaytarish; 500 javobida traceback; /health model yuklanmagan paytda ham 200 qaytarishi; javobda model versiyasi yo'qligi; API va batch da turli kod; TestClient ni with siz ishlatib, lifespan ishlamaganini sezmaslik; --workers da xotira ko'payishini unutish.

3. Tez ma'lumotnoma

python

import warnings
from contextlib import asynccontextmanager
from enum import Enum

from fastapi import Depends, FastAPI, HTTPException, Request
from pydantic import BaseModel, ConfigDict, Field

with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from fastapi.testclient import TestClient


class Tarif(str, Enum):
    oylik = "oylik"
    yillik = "yillik"


class Mijoz(BaseModel):
    model_config = ConfigDict(extra="forbid")
    oylar: int = Field(ge=0, le=600)
    tolov: float = Field(gt=0, le=1000)
    shikoyat: int | None = None
    tarif: Tarif


@asynccontextmanager
async def lifespan(app):
    app.state.predictor = object()          # Predictor(papka) - bir marta
    yield
    app.state.predictor = None


app = FastAPI(title="Model API", version="1.0.0", lifespan=lifespan)


def predictor_ol(request: Request):
    p = getattr(request.app.state, "predictor", None)
    if p is None:
        raise HTTPException(status_code=503, detail="model yuklanmagan")
    return p


@app.post("/predict")
def predict(mijoz: Mijoz, pr=Depends(predictor_ol)):
    return {"oylar": mijoz.oylar}


# with TestClient(app) as k:
#     r = k.post("/predict", json={"oylar": 3, "tolov": 90, "tarif": "oylik"})

Tuzilma xulosasi

API faqat natija tez kerak bo'lganda; aks holda batch - bitta Predictor
REST: GET o'qish, POST hisoblash; 4xx - mijoz, 5xx - server
pydantic: Field (ge, le, gt, min_length), Enum, T | None, extra="forbid"
lifespan - model BIR marta; Depends - endpointga berish (503 agar yo'q)
/health, /version, /predict, /predict_batch; javobda versiya
422 sxema, 400 biznes, 500 bizning xato (tracebacksiz), 503 tayyor emas
app.openapi() - avtomatik hujjat; TestClient - serversiz sinov

4. Batafsil misollar

Misollar real FastAPI/pydantic/Flask bilan (Python 3.14). Server ishga tushirilmaydi — hamma so'rovlar TestClient / test_client() orqali jarayon ichida.

Misol 1 — REST asoslari: minimal FastAPI ilova

python

"""Endpoint, metod, parametrlar, status kodlar va OpenAPI - TestClient bilan."""

import warnings

from fastapi import FastAPI, HTTPException

with warnings.catch_warnings():
    warnings.simplefilter("ignore")        # StarletteDeprecationWarning (httpx)
    from fastapi.testclient import TestClient

app = FastAPI(title="Salom API", version="0.1.0")
MIJOZLAR = {"m1": {"ism": "Ali", "tarif": "oylik"},
            "m2": {"ism": "Vali", "tarif": "yillik"}}


@app.get("/health")
def health():
    return {"holat": "ok"}


@app.get("/mijozlar/{mijoz_id}")
def mijoz_ol(mijoz_id: str):
    if mijoz_id not in MIJOZLAR:
        raise HTTPException(status_code=404, detail=f"mijoz topilmadi: {mijoz_id}")
    return MIJOZLAR[mijoz_id]


@app.post("/mijozlar", status_code=201)
def mijoz_qosh(malumot: dict):
    yangi_id = f"m{len(MIJOZLAR) + 1}"
    MIJOZLAR[yangi_id] = malumot
    return {"id": yangi_id}


@app.get("/kvadrat")
def kvadrat(x: float, aniqlik: int = 2):
    return {"x": x, "natija": round(x * x, aniqlik)}


def main() -> None:
    klient = TestClient(app)
    jurnal = []

    def sorov(metod, yol, **kw):
        r = klient.request(metod, yol, **kw)
        jurnal.append((metod, yol, r.status_code))
        return r

    print("=== 1. GET - o'qish ===")
    r = sorov("GET", "/health")
    print(f"  {r.status_code} {r.json()}  content-type: "
          f"{r.headers['content-type']}")
    r = sorov("GET", "/mijozlar/m1")
    print(f"  {r.status_code} {r.json()}")
    r = sorov("GET", "/mijozlar/m9")
    print(f"  {r.status_code} {r.json()}")

    print("\n=== 2. Query parametrlar va avtomatik tur o'girish ===")
    for yol in ["/kvadrat?x=3", "/kvadrat?x=1.5&aniqlik=1", "/kvadrat?x=abc",
                "/kvadrat"]:
        r = sorov("GET", yol)
        if r.status_code == 200:
            print(f"  {yol:<26} {r.status_code} {r.json()}")
        else:
            d = r.json()["detail"][0]
            print(f"  {yol:<26} {r.status_code} type={d['type']}, loc={d['loc']}")

    print("\n=== 3. POST - yaratish (201) ===")
    r = sorov("POST", "/mijozlar", json={"ism": "Olim", "tarif": "oylik"})
    print(f"  {r.status_code} {r.json()}")
    r = sorov("GET", f"/mijozlar/{r.json()['id']}")
    print(f"  qayta o'qish: {r.status_code} {r.json()}")
    r = sorov("POST", "/mijozlar", content="bu json emas",
              headers={"content-type": "application/json"})
    print(f"  buzilgan JSON: {r.status_code} type={r.json()['detail'][0]['type']}")

    print("\n=== 4. Noto'g'ri yo'l va metod ===")
    r = sorov("GET", "/yoq")
    print(f"  GET /yoq:        {r.status_code} {r.json()}")
    r = sorov("DELETE", "/health")
    print(f"  DELETE /health:  {r.status_code} {r.json()}")

    print("\n=== 5. Jurnal: status kodlar guruhlari ===")
    for guruh in ["2xx", "4xx", "5xx"]:
        mos = [f"{m} {y} -> {k}" for m, y, k in jurnal
               if str(k)[0] == guruh[0]]
        print(f"  {guruh}: {len(mos)} ta")
        for q in mos[:4]:
            print(f"    {q}")
        if len(mos) > 4:
            print(f"    (+{len(mos) - 4} ta)")

    print("\n=== 6. Avtomatik OpenAPI sxemasi ===")
    sxema = app.openapi()
    print(f"  openapi {sxema['openapi']}, {sxema['info']['title']} "
          f"{sxema['info']['version']}")
    for yol, amallar in sorted(sxema["paths"].items()):
        for metod, amal in sorted(amallar.items()):
            kodlar = sorted(amal["responses"])
            print(f"  {metod.upper():<5} {yol:<22} {amal['summary']:<12} {kodlar}")
    print("  ⭐ Turlar - shartnoma: FastAPI ularni tekshiradi va hujjatlaydi")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. GET - o'qish ===
  200 {'holat': 'ok'}  content-type: application/json
  200 {'ism': 'Ali', 'tarif': 'oylik'}
  404 {'detail': 'mijoz topilmadi: m9'}

=== 2. Query parametrlar va avtomatik tur o'girish ===
  /kvadrat?x=3               200 {'x': 3.0, 'natija': 9.0}
  /kvadrat?x=1.5&aniqlik=1   200 {'x': 1.5, 'natija': 2.2}
  /kvadrat?x=abc             422 type=float_parsing, loc=['query', 'x']
  /kvadrat                   422 type=missing, loc=['query', 'x']

=== 3. POST - yaratish (201) ===
  201 {'id': 'm3'}
  qayta o'qish: 200 {'ism': 'Olim', 'tarif': 'oylik'}
  buzilgan JSON: 422 type=json_invalid

=== 4. Noto'g'ri yo'l va metod ===
  GET /yoq:        404 {'detail': 'Not Found'}
  DELETE /health:  405 {'detail': 'Method Not Allowed'}

=== 5. Jurnal: status kodlar guruhlari ===
  2xx: 6 ta
    GET /health -> 200
    GET /mijozlar/m1 -> 200
    GET /kvadrat?x=3 -> 200
    GET /kvadrat?x=1.5&aniqlik=1 -> 200
    (+2 ta)
  4xx: 6 ta
    GET /mijozlar/m9 -> 404
    GET /kvadrat?x=abc -> 422
    GET /kvadrat -> 422
    POST /mijozlar -> 422
    (+2 ta)
  5xx: 0 ta

=== 6. Avtomatik OpenAPI sxemasi ===
  openapi 3.1.0, Salom API 0.1.0
  GET   /health                Health       ['200']
  GET   /kvadrat               Kvadrat      ['200', '422']
  POST  /mijozlar              Mijoz Qosh   ['201', '422']
  GET   /mijozlar/{mijoz_id}   Mijoz Ol     ['200', '422']
  ⭐ Turlar - shartnoma: FastAPI ularni tekshiradi va hujjatlaydi

Natija tahlili. 1-bo'limda uchta GET so'rov: xizmat holati, mavjud mijoz (200) va mavjud bo'lmagan mijoz — biz o'zimiz HTTPException(404) bilan qaytargan tushunarli detail. Javob turi har doim application/json. 2-bo'lim FastAPI ning turlardan foydalanishini ko'rsatadi: x=3 so'rov satridan matn bo'lib keladi, lekin funksiyaga 3.0 (float) bo'lib tushdi; aniqlik berilmasa sukut 2 ishlatildi. x=abc va x umuman yo'q bo'lganda biz bitta ham if yozmagan bo'lsak ham 422 qaytdi, xato turi (float_parsing, missing) va joyi (['query', 'x']) bilan. 3-bo'limda POST 201 Created qaytardi va yaratilgan resurs GET bilan o'qildi; JSON emas matn yuborilganda — json_invalid bilan 422. 4-bo'limda noto'g'ri yo'l (404) va noto'g'ri metod (405, /health faqat GET ni qabul qiladi). 5-bo'limdagi jurnal: 6 ta 2xx, 6 ta 4xx va birorta ham 5xx yo'q — barcha xatolar mijoz tomonida edi va xizmat ularni to'g'ri tasnifladi. 6-bo'limda app.openapi() biz yozgan to'rt endpointni avtomatik hujjatlashtirdi: parametrli endpointlar uchun 422 javobi ham qo'shilgan, POST uchun esa biz belgilagan 201.

Misol 2 — Pydantic sxemalar: chegaralar, Enum, ixtiyoriy maydonlar

python

"""pydantic: Field chegaralari, Enum, ixtiyoriy maydon, extra, strict, JSON sxema."""

import json
import warnings
from enum import Enum

from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field, ValidationError

with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from fastapi.testclient import TestClient


class Tarif(str, Enum):
    oylik = "oylik"
    yillik = "yillik"
    ikki_yillik = "ikki_yillik"


class Mijoz(BaseModel):
    model_config = ConfigDict(extra="forbid")
    oylar: int = Field(ge=0, le=600, description="mijozlik muddati, oy")
    tolov: float = Field(gt=0, le=1000, description="oylik to'lov, ming so'm")
    shikoyat: int | None = Field(default=None, ge=0, le=100)
    tarif: Tarif
    hudud: str = Field(default="nomalum", min_length=2, max_length=40)


app = FastAPI()


@app.post("/tekshir")
def tekshir(mijoz: Mijoz):
    return {"qabul": mijoz.model_dump(mode="json")}


def xatolar(malumot, **kw):
    try:
        Mijoz.model_validate(malumot, **kw)
        return []
    except ValidationError as e:
        return [(".".join(map(str, x["loc"])), x["type"]) for x in e.errors()]


def main() -> None:
    print("=== 1. To'g'ri kirish va sukut qiymatlar ===")
    m = Mijoz.model_validate({"oylar": 12, "tolov": 55.5, "tarif": "oylik"})
    print(f"  {m.model_dump(mode='json')}")
    print(f"  tarif turi: {type(m.tarif).__name__}, qiymati: {m.tarif.value}")

    print("\n=== 2. Lax va strict rejim ===")
    kirish = {"oylar": "12", "tolov": "55.5", "tarif": "yillik"}
    m = Mijoz.model_validate(kirish)
    print(f"  lax:    oylar={m.oylar!r} ({type(m.oylar).__name__}), "
          f"tolov={m.tolov!r}")
    print(f"  strict: {xatolar(kirish, strict=True)}")
    for oylar in [12.0, 12.5]:
        natija = xatolar({"oylar": oylar, "tolov": 50, "tarif": "oylik"})
        print(f"  oylar={oylar}: {natija or 'qabul'}")

    print("\n=== 3. Bir nechta xato - hammasi birdaniga ===")
    yomon = {"oylar": -3, "tolov": 0, "shikoyat": 500, "tarif": "haftalik",
             "hudud": "X"}
    for joy, tur in xatolar(yomon):
        print(f"  {joy:<9} {tur}")

    print("\n=== 4. extra='forbid' - imlo xatosi darhol ko'rinadi ===")
    imlo = {"oylar": 5, "tolov": 40, "tarif": "oylik", "region": "Buxoro"}
    print(f"  {xatolar(imlo)}")

    print("\n=== 5. Sxemadan avtomatik JSON Schema ===")
    js = Mijoz.model_json_schema()
    for nom, xos in js["properties"].items():
        cheklov = {k: v for k, v in xos.items()
                   if k in ("minimum", "maximum", "exclusiveMinimum",
                            "minLength", "maxLength", "default")}
        tur = xos.get("type") or xos.get("$ref", "").split("/")[-1] or "anyOf"
        print(f"  {nom:<9} {tur:<8} {cheklov}")
    print(f"  majburiy: {js['required']}")
    print(f"  Tarif: {js['$defs']['Tarif']['enum']}")

    print("\n=== 6. FastAPI ichida: 422 javob ===")
    klient = TestClient(app)
    r = klient.post("/tekshir", json={"oylar": 3, "tolov": 95.5,
                                      "tarif": "oylik"})
    print(f"  to'g'ri:  {r.status_code} {json.dumps(r.json(), ensure_ascii=False)}")
    r = klient.post("/tekshir", json=yomon)
    detail = r.json()["detail"]
    print(f"  yomon:    {r.status_code}, xatolar soni: {len(detail)}")
    d = detail[0]
    print(f"  birinchisi: loc={d['loc']}, type={d['type']}, input={d['input']}")
    print(f"    msg: {d['msg']}")
    print("  ⭐ Sxema bir marta yoziladi: validatsiya + xato + hujjat")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. To'g'ri kirish va sukut qiymatlar ===
  {'oylar': 12, 'tolov': 55.5, 'shikoyat': None, 'tarif': 'oylik', 'hudud': 'nomalum'}
  tarif turi: Tarif, qiymati: oylik

=== 2. Lax va strict rejim ===
  lax:    oylar=12 (int), tolov=55.5
  strict: [('oylar', 'int_type'), ('tolov', 'float_type'), ('tarif', 'is_instance_of')]
  oylar=12.0: qabul
  oylar=12.5: [('oylar', 'int_from_float')]

=== 3. Bir nechta xato - hammasi birdaniga ===
  oylar     greater_than_equal
  tolov     greater_than
  shikoyat  less_than_equal
  tarif     enum
  hudud     string_too_short

=== 4. extra='forbid' - imlo xatosi darhol ko'rinadi ===
  [('region', 'extra_forbidden')]

=== 5. Sxemadan avtomatik JSON Schema ===
  oylar     integer  {'maximum': 600, 'minimum': 0}
  tolov     number   {'exclusiveMinimum': 0, 'maximum': 1000}
  shikoyat  anyOf    {'default': None}
  tarif     Tarif    {}
  hudud     string   {'default': 'nomalum', 'maxLength': 40, 'minLength': 2}
  majburiy: ['oylar', 'tolov', 'tarif']
  Tarif: ['oylik', 'yillik', 'ikki_yillik']

=== 6. FastAPI ichida: 422 javob ===
  to'g'ri:  200 {"qabul": {"oylar": 3, "tolov": 95.5, "shikoyat": null, "tarif": "oylik", "hudud": "nomalum"}}
  yomon:    422, xatolar soni: 5
  birinchisi: loc=['body', 'oylar'], type=greater_than_equal, input=-3
    msg: Input should be greater than or equal to 0
  ⭐ Sxema bir marta yoziladi: validatsiya + xato + hujjat

Natija tahlili. 1-bo'limda faqat majburiy maydonlar berildi: shikoyat None, hudud esa "nomalum" sukutini oldi; tarif oddiy satr emas, Tarif enum a'zosi. 2-bo'lim lax va strict farqini ko'rsatadi: sukut (lax) rejimda "12" butun 12 ga, "55.5" float ga o'girildi — JSON mijozlari sonlarni ba'zan matn sifatida yuboradi, bu qulay. Strict rejimda esa uchala maydon ham rad etildi. 12.0 butun son sifatida qabul qilindi (kasr qismi nol), 12.5 esa int_from_float xatosi berdi — pydantic ma'lumotni jimgina yaxlitlamaydi. 3-bo'limda beshta maydonning hammasi noto'g'ri edi va beshta xato birdaniga qaytdi, har biri o'z turi bilan (greater_than_equal, greater_than, less_than_equal, enum, string_too_short). 4-bo'limda hudud o'rniga region yozilgan: extra="forbid" bo'lmaganida bu so'rov xatosiz o'tardi va hudud jimgina "nomalum" bo'lib qolardi. 5-bo'limda xuddi shu sxemadan JSON Schema chiqdi: chegaralar (minimum, exclusiveMinimum, maxLength), sukutlar, majburiy maydonlar ro'yxati va enum qiymatlari — bu OpenAPI hujjatining asosi. 6-bo'limda FastAPI xuddi shu sxema bilan: to'g'ri kirish 200, yomon kirish — beshta xatoli 422 javob, har xatoda loc (['body', 'oylar']), type, input va inson o'qiy oladigan msg.

Misol 3 — To'liq model API: lifespan va to'rt endpoint

python

"""Model API: lifespan, Depends, /predict, /predict_batch, /health, /version."""

import warnings
from contextlib import asynccontextmanager
from enum import Enum

import numpy as np
import pandas as pd
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from fastapi.testclient import TestClient

SONLI = ["oylar", "tolov", "shikoyat"]
KATEGORIAL = ["tarif", "hudud"]
YUKLASHLAR = {"soni": 0}


def malumot(n, seed):
    rng = np.random.default_rng(seed)
    df = pd.DataFrame({
        "oylar": rng.integers(1, 72, n),
        "tolov": rng.normal(60, 20, n).clip(10, 150).round(2),
        "shikoyat": rng.poisson(0.8, n).astype(float),
        "tarif": rng.choice(["oylik", "yillik", "ikki_yillik"], n,
                            p=[0.55, 0.3, 0.15]),
        "hudud": rng.choice(["Toshkent", "Samarqand", "Buxoro", "Andijon"], n),
    })
    logit = (-1.0 - 0.04 * df["oylar"] + 0.02 * (df["tolov"] - 60)
             + 0.5 * df["shikoyat"]
             + np.where(df["tarif"] == "oylik", 1.0, -0.5))
    df["ketdi"] = (rng.random(n) < 1 / (1 + np.exp(-logit))).astype(int)
    return df


class Predictor:
    """27.6 dagi Predictor ning qisqa nusxasi (paket o'rniga shu yerda o'rgatiladi)."""
    nom, versiya, chegara = "mijoz_ketishi", "1.2.0", 0.35

    def __init__(self):
        YUKLASHLAR["soni"] += 1
        df = malumot(3000, seed=0)
        pre = ColumnTransformer([
            ("son", Pipeline([("bosh", SimpleImputer(strategy="median")),
                              ("olchov", StandardScaler())]), SONLI),
            ("kat", OneHotEncoder(handle_unknown="ignore"), KATEGORIAL)])
        self.model = Pipeline([("pre", pre), ("clf", LogisticRegression())])
        self.model.fit(df[SONLI + KATEGORIAL], df["ketdi"])
        self.chaqiruvlar = 0

    def predict_proba(self, df):
        self.chaqiruvlar += 1
        return self.model.predict_proba(df[SONLI + KATEGORIAL])[:, 1]


class Tarif(str, Enum):
    oylik = "oylik"
    yillik = "yillik"
    ikki_yillik = "ikki_yillik"


class Mijoz(BaseModel):
    model_config = ConfigDict(extra="forbid")
    mijoz_id: str | None = Field(default=None, max_length=32)
    oylar: int = Field(ge=0, le=600)
    tolov: float = Field(gt=0, le=1000)
    shikoyat: int | None = Field(default=None, ge=0, le=100)
    tarif: Tarif
    hudud: str = Field(default="nomalum", min_length=2, max_length=40)


class Partiya(BaseModel):
    mijozlar: list[Mijoz] = Field(min_length=1, max_length=100)


class Bashorat(BaseModel):
    mijoz_id: str | None
    ehtimol: float
    sinf: int
    versiya: str


class PartiyaJavob(BaseModel):
    natijalar: list[Bashorat]
    soni: int


@asynccontextmanager
async def lifespan(app):
    app.state.predictor = Predictor()          # BIR MARTA, ishga tushishda
    yield
    app.state.predictor = None                 # to'xtashda tozalash


app = FastAPI(title="Mijoz ketishi API", version="1.0.0", lifespan=lifespan)


@app.exception_handler(Exception)
async def ichki_xato(request, exc):
    # tafsilot faqat logga 27.8-bob; mijozga - umumiy xabar
    return JSONResponse(status_code=500,
                        content={"xato": "ichki xato", "tur": type(exc).__name__})


def predictor_ol(request: Request):
    p = getattr(request.app.state, "predictor", None)
    if p is None:
        raise HTTPException(status_code=503, detail="model yuklanmagan")
    return p


def bashoratlar(pr, mijozlar):
    df = pd.DataFrame([m.model_dump(mode="json") for m in mijozlar])
    p = pr.predict_proba(df)
    return [Bashorat(mijoz_id=m.mijoz_id, ehtimol=round(float(x), 4),
                     sinf=int(x >= pr.chegara), versiya=pr.versiya)
            for m, x in zip(mijozlar, p)]


@app.get("/health", summary="tayyorlik")
def health(request: Request):
    if getattr(request.app.state, "predictor", None) is None:
        return JSONResponse(status_code=503, content={"holat": "tayyor emas"})
    return {"holat": "ok"}


@app.get("/version", summary="versiyalar")
def version(pr: Predictor = Depends(predictor_ol)):
    return {"model": pr.nom, "model_versiyasi": pr.versiya,
            "api_versiyasi": app.version}


@app.post("/predict", response_model=Bashorat, summary="bitta mijoz",
          responses={503: {"description": "model yuklanmagan"}})
def predict(mijoz: Mijoz, pr: Predictor = Depends(predictor_ol)):
    return bashoratlar(pr, [mijoz])[0]


@app.post("/predict_batch", response_model=PartiyaJavob, summary="partiya",
          responses={400: {"description": "takroriy mijoz_id"},
                     503: {"description": "model yuklanmagan"}})
def predict_batch(partiya: Partiya, pr: Predictor = Depends(predictor_ol)):
    idlar = [m.mijoz_id for m in partiya.mijozlar if m.mijoz_id is not None]
    if len(idlar) != len(set(idlar)):
        raise HTTPException(status_code=400, detail="takroriy mijoz_id")
    natijalar = bashoratlar(pr, partiya.mijozlar)
    return PartiyaJavob(natijalar=natijalar, soni=len(natijalar))


MIJOZ = {"mijoz_id": "a1", "oylar": 3, "tolov": 95.5, "shikoyat": 2,
         "tarif": "oylik", "hudud": "Buxoro"}


def main() -> None:
    print("=== 1. Lifespan ishlamagan (with siz TestClient) ===")
    k0 = TestClient(app)
    for metod, yol in [("GET", "/health"), ("POST", "/predict")]:
        r = k0.request(metod, yol, json=MIJOZ if metod == "POST" else None)
        print(f"  {metod} {yol:<9} {r.status_code} {r.json()}")
    print(f"  model yuklashlar soni: {YUKLASHLAR['soni']}")

    with TestClient(app, raise_server_exceptions=False) as k:
        print("\n=== 2. Ishga tushdi: /health va /version ===")
        print(f"  model yuklashlar soni: {YUKLASHLAR['soni']}")
        print(f"  /health  {k.get('/health').status_code} {k.get('/health').json()}")
        print(f"  /version {k.get('/version').json()}")

        print("\n=== 3. /predict ===")
        r = k.post("/predict", json=MIJOZ)
        print(f"  {r.status_code} {r.json()}")
        sodda = {"oylar": 48, "tolov": 40, "tarif": "ikki_yillik"}
        r = k.post("/predict", json=sodda)
        print(f"  {r.status_code} {r.json()}")

        print("\n=== 4. /predict_batch ===")
        rng = np.random.default_rng(1)
        partiya = [{"mijoz_id": f"b{i}", "oylar": int(rng.integers(1, 72)),
                    "tolov": round(float(rng.uniform(20, 120)), 2),
                    "tarif": str(rng.choice(["oylik", "yillik"]))}
                   for i in range(40)]
        oldin = k.app.state.predictor.chaqiruvlar
        r = k.post("/predict_batch", json={"mijozlar": partiya})
        j = r.json()
        print(f"  {r.status_code} soni={j['soni']}, "
              f"model chaqiruvlari: {k.app.state.predictor.chaqiruvlar - oldin}")
        for b in j["natijalar"][:3]:
            print(f"    {b}")
        print(f"    (+{j['soni'] - 3} ta)")
        yakka = [k.post("/predict", json=m).json()["ehtimol"]
                 for m in partiya[:10]]
        mos = all(abs(a - b["ehtimol"]) < 1e-9
                  for a, b in zip(yakka, j["natijalar"][:10]))
        print(f"  10 ta yakka /predict bilan mos: {mos}")

        print("\n=== 5. Xato javoblar ===")
        holatlar = [
            ("tolov manfiy", "/predict", {**MIJOZ, "tolov": -5}),
            ("noma'lum tarif", "/predict", {**MIJOZ, "tarif": "haftalik"}),
            ("bo'sh partiya", "/predict_batch", {"mijozlar": []}),
            ("101 ta mijoz", "/predict_batch",
             {"mijozlar": [{**MIJOZ, "mijoz_id": None}] * 101}),
            ("takroriy id", "/predict_batch", {"mijozlar": [MIJOZ, MIJOZ]}),
            ("imlo: /predikt", "/predikt", MIJOZ),
        ]
        for nom, yol, tana in holatlar:
            r = k.post(yol, json=tana)
            d = r.json().get("detail")
            izoh = d[0]["type"] if isinstance(d, list) else d
            print(f"  {nom:<16} {r.status_code}  {izoh}")
        asl_model = k.app.state.predictor.model
        k.app.state.predictor.model = None            # model "buzildi"
        r = k.post("/predict", json=MIJOZ)
        print(f"  {'model buzildi':<16} {r.status_code}  {r.json()}")
        k.app.state.predictor.model = asl_model

        print("\n=== 6. Model necha marta yuklandi? ===")
        print(f"  jami so'rovlar: 20 dan ortiq, model yuklashlar: "
              f"{YUKLASHLAR['soni']}")
    print(f"  to'xtagandan keyin predictor: {app.state.predictor}")

    print("\n=== 7. OpenAPI: endpointlar ro'yxati ===")
    sxema = app.openapi()
    for yol, amallar in sorted(sxema["paths"].items()):
        for metod, amal in amallar.items():
            print(f"  {metod.upper():<5} {yol:<15} {amal['summary']:<11} "
                  f"{sorted(amal['responses'])}")
    print(f"  sxemalar: {sorted(sxema['components']['schemas'])}")
    print("  ⭐ Model bir marta yuklanadi; har javobda versiya; xato kodlari aniq")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. Lifespan ishlamagan (with siz TestClient) ===
  GET /health   503 {'holat': 'tayyor emas'}
  POST /predict  503 {'detail': 'model yuklanmagan'}
  model yuklashlar soni: 0

=== 2. Ishga tushdi: /health va /version ===
  model yuklashlar soni: 1
  /health  200 {'holat': 'ok'}
  /version {'model': 'mijoz_ketishi', 'model_versiyasi': '1.2.0', 'api_versiyasi': '1.0.0'}

=== 3. /predict ===
  200 {'mijoz_id': 'a1', 'ehtimol': 0.8268, 'sinf': 1, 'versiya': '1.2.0'}
  200 {'mijoz_id': None, 'ehtimol': 0.0312, 'sinf': 0, 'versiya': '1.2.0'}

=== 4. /predict_batch ===
  200 soni=40, model chaqiruvlari: 1
    {'mijoz_id': 'b0', 'ehtimol': 0.2899, 'sinf': 0, 'versiya': '1.2.0'}
    {'mijoz_id': 'b1', 'ehtimol': 0.8259, 'sinf': 1, 'versiya': '1.2.0'}
    {'mijoz_id': 'b2', 'ehtimol': 0.4116, 'sinf': 1, 'versiya': '1.2.0'}
    (+37 ta)
  10 ta yakka /predict bilan mos: True

=== 5. Xato javoblar ===
  tolov manfiy     422  greater_than
  noma'lum tarif   422  enum
  bo'sh partiya    422  too_short
  101 ta mijoz     422  too_long
  takroriy id      400  takroriy mijoz_id
  imlo: /predikt   404  Not Found
  model buzildi    500  {'xato': 'ichki xato', 'tur': 'AttributeError'}

=== 6. Model necha marta yuklandi? ===
  jami so'rovlar: 20 dan ortiq, model yuklashlar: 1
  to'xtagandan keyin predictor: None

=== 7. OpenAPI: endpointlar ro'yxati ===
  GET   /health         tayyorlik   ['200']
  POST  /predict        bitta mijoz ['200', '422', '503']
  POST  /predict_batch  partiya     ['200', '400', '422', '503']
  GET   /version        versiyalar  ['200']
  sxemalar: ['Bashorat', 'HTTPValidationError', 'Mijoz', 'Partiya', 'PartiyaJavob', 'Tarif', 'ValidationError']
  ⭐ Model bir marta yuklanadi; har javobda versiya; xato kodlari aniq

Natija tahlili. 1-bo'lim muhim tuzoqni ko'rsatadi: TestClient ni with siz yaratganda lifespan ishlamadi — model yuklanmadi (0 marta), /health halol 503 qaytardi, /predict esa Depends(predictor_ol) orqali 503 model yuklanmagan berdi. Bu ishlab chiqarishdagi "xizmat ishga tushdi, lekin model hali yuklanmoqda" holatining aynan o'zi. 2-bo'limda with bilan lifespan ishladi va model bir marta yuklandi. 3-bo'limda to'liq kirish (0.8268, sinf 1) va faqat majburiy maydonli kirish (0.0312, shikoyat bo'sh — ichki SimpleImputer to'ldirdi); har javobda model versiyasi. 4-bo'limda 40 mijozlik partiya bitta model chaqiruvi bilan hisoblandi va birinchi 10 tasi yakka /predict natijalari bilan mos keldi — batch va onlayn yo'l bir xil Predictor dan o'tadi. 5-bo'limda har xato o'z kodini oldi: sxema buzilishi — 422 (greater_than, enum, too_short, too_long), biznes qoidasi (takroriy mijoz_id) — 400, noto'g'ri yo'l — 404. Model "buzilganda" (model = None) mijoz faqat 500 va umumiy xabar oldi — traceback yoki fayl yo'li yo'q. 6-bo'limda yigirmadan ortiq so'rovdan keyin ham model yuklashlar soni 1, xizmat to'xtaganda esa lifespan ning yield dan keyingi qismi predictor ni tozaladi. 7-bo'limda OpenAPI biz belgilagan qisqa tavsiflar (summary) va javob kodlarini (400, 503) ham hujjatladi, sxemalar ro'yxatida esa bizning pydantic modellarimiz va FastAPI ning standart validatsiya xatosi sxemasi bor.

Misol 4 — Flask bilan taqqoslash va onlayn/batch

python

"""Flask: validatsiyasiz va pydantic bilan; FastAPI bilan taqqoslash; batch."""

import warnings
from enum import Enum

import numpy as np
import pandas as pd
from fastapi import FastAPI
from flask import Flask, jsonify, request
from pydantic import BaseModel, ConfigDict, Field, ValidationError
from sklearn.linear_model import LogisticRegression
from werkzeug.exceptions import HTTPException

with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from fastapi.testclient import TestClient

USTUNLAR = ["oylar", "tolov", "oylik_tarif"]


class Model:
    def __init__(self):
        rng = np.random.default_rng(0)
        X = pd.DataFrame({"oylar": rng.integers(1, 72, 2000),
                          "tolov": rng.normal(60, 20, 2000),
                          "oylik_tarif": rng.integers(0, 2, 2000)})
        logit = -1 - 0.04 * X["oylar"] + 0.02 * (X["tolov"] - 60) + X["oylik_tarif"]
        y = (rng.random(2000) < 1 / (1 + np.exp(-logit))).astype(int)
        self.clf = LogisticRegression(max_iter=1000).fit(X, y)
        self.chaqiruvlar = 0
        self.qatorlar = 0

    def ehtimol(self, qatorlar):
        self.chaqiruvlar += 1
        self.qatorlar += len(qatorlar)
        df = pd.DataFrame(qatorlar)
        df["oylik_tarif"] = (df["tarif"] == "oylik").astype(int)
        return self.clf.predict_proba(df[USTUNLAR])[:, 1]


MODEL = Model()


class Tarif(str, Enum):
    oylik = "oylik"
    yillik = "yillik"


class Mijoz(BaseModel):
    model_config = ConfigDict(extra="forbid")
    oylar: int = Field(ge=0, le=600)
    tolov: float = Field(gt=0, le=1000)
    tarif: Tarif


class Partiya(BaseModel):
    mijozlar: list[Mijoz] = Field(min_length=1, max_length=100)


# ---------- FastAPI ----------
fapi = FastAPI()


@fapi.post("/predict")
def f_predict(m: Mijoz):
    return {"ehtimol": round(float(MODEL.ehtimol([m.model_dump(mode="json")])[0]), 4)}


@fapi.post("/predict_batch")
def f_batch(p: Partiya):
    e = MODEL.ehtimol([m.model_dump(mode="json") for m in p.mijozlar])
    return {"ehtimollar": [round(float(x), 4) for x in e]}


# ---------- Flask ----------
flask_ilova = Flask(__name__)


@flask_ilova.post("/predict_sodda")
def fl_sodda():
    malumot = request.get_json()          # validatsiyasiz
    return jsonify(ehtimol=round(float(MODEL.ehtimol([malumot])[0]), 4))


@flask_ilova.post("/predict")
def fl_predict():
    malumot = request.get_json(silent=True)
    if not isinstance(malumot, dict):
        return jsonify(xato="JSON obyekt kerak"), 400
    try:
        m = Mijoz.model_validate(malumot)
    except ValidationError as e:
        return jsonify(detail=e.errors(include_url=False,
                                       include_context=False)), 422
    return jsonify(ehtimol=round(float(MODEL.ehtimol([m.model_dump(mode="json")])[0]), 4))


@flask_ilova.errorhandler(Exception)
def fl_xato(e):
    if isinstance(e, HTTPException):
        return e
    return jsonify(xato="ichki xato", tur=type(e).__name__), 500


def main() -> None:
    fk = TestClient(fapi)
    flk = flask_ilova.test_client()

    print("=== 1. Bir xil kirish - uch xil API ===")
    holatlar = [
        ("to'g'ri", {"oylar": 3, "tolov": 95.5, "tarif": "oylik"}),
        ("tolov matn", {"oylar": 3, "tolov": "ellik", "tarif": "oylik"}),
        ("oylar manfiy", {"oylar": -40, "tolov": 95.5, "tarif": "oylik"}),
        ("tiyinda", {"oylar": 3, "tolov": 9550, "tarif": "oylik"}),
        ("noma'lum tarif", {"oylar": 3, "tolov": 95.5, "tarif": "haftalik"}),
    ]
    print(f"  {'holat':<15} {'FastAPI':<14} {'Flask+pydantic':<16} Flask sodda")
    for nom, tana in holatlar:
        qatorlar = []
        for r in [fk.post("/predict", json=tana),
                  flk.post("/predict", json=tana),
                  flk.post("/predict_sodda", json=tana)]:
            j = r.json() if callable(r.json) else r.json
            qatorlar.append(f"{r.status_code} {j.get('ehtimol', '')}".strip())
        print(f"  {nom:<15} {qatorlar[0]:<14} {qatorlar[1]:<16} {qatorlar[2]}")

    print("\n=== 2. Flask sodda: jim xatolar ===")
    r = flk.post("/predict_sodda", json={"oylar": 3, "tolov": 9550,
                                        "tarif": "oylik"})
    print(f"  tolov tiyinda (100x): {r.status_code}, ehtimol {r.get_json()['ehtimol']}"
          f" - xato YO'Q, lekin javob ma'nosiz")
    r = flk.post("/predict_sodda", json={"oylar": 3, "tolov": 95.5,
                                        "tarif": "haftalik"})
    print(f"  noma'lum tarif:      {r.status_code}, ehtimol {r.get_json()['ehtimol']}"
          f" - 'oylik emas' deb qabul qilindi")

    print("\n=== 3. Onlayn va batch: 200 mijoz ===")
    rng = np.random.default_rng(2)
    mijozlar = [{"oylar": int(rng.integers(1, 72)),
                 "tolov": round(float(rng.uniform(20, 120)), 2),
                 "tarif": str(rng.choice(["oylik", "yillik"]))}
                for _ in range(200)]
    MODEL.chaqiruvlar = 0
    onlayn = [fk.post("/predict", json=m).json()["ehtimol"] for m in mijozlar]
    onlayn_chaqiruv = MODEL.chaqiruvlar
    MODEL.chaqiruvlar = 0
    batch = []
    for i in range(0, 200, 100):
        r = fk.post("/predict_batch", json={"mijozlar": mijozlar[i:i + 100]})
        batch += r.json()["ehtimollar"]
    batch_chaqiruv = MODEL.chaqiruvlar
    # faraziy narx: har HTTP so'rov 2 ms, har model chaqiruvi 1 ms + 0.01 ms/qator
    narx = lambda sorov, chaq: sorov * 2 + chaq * 1 + 200 * 0.01
    print(f"  onlayn: 200 so'rov, {onlayn_chaqiruv} model chaqiruvi, "
          f"faraziy {narx(200, onlayn_chaqiruv):.0f} ms")
    print(f"  batch:  2 so'rov,   {batch_chaqiruv} model chaqiruvi, "
          f"faraziy {narx(2, batch_chaqiruv):.0f} ms")
    farq = max(abs(a - b) for a, b in zip(onlayn, batch))
    print(f"  natijalar max farqi: {farq}")
    print("  ⭐ Validatsiya freymvorkdan muhimroq; batch - kam chaqiruv")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. Bir xil kirish - uch xil API ===
  holat           FastAPI        Flask+pydantic   Flask sodda
  to'g'ri         200 0.6148     200 0.6148       200 0.6148
  tolov matn      422            422              500
  oylar manfiy    422            422              200 0.8928
  tiyinda         422            422              200 1.0
  noma'lum tarif  422            422              200 0.3481

=== 2. Flask sodda: jim xatolar ===
  tolov tiyinda (100x): 200, ehtimol 1.0 - xato YO'Q, lekin javob ma'nosiz
  noma'lum tarif:      200, ehtimol 0.3481 - 'oylik emas' deb qabul qilindi

=== 3. Onlayn va batch: 200 mijoz ===
  onlayn: 200 so'rov, 200 model chaqiruvi, faraziy 602 ms
  batch:  2 so'rov,   2 model chaqiruvi, faraziy 8 ms
  natijalar max farqi: 0.0
  ⭐ Validatsiya freymvorkdan muhimroq; batch - kam chaqiruv

Natija tahlili. 1-bo'lim jadvali darsning asosiy xulosasini ko'rsatadi: freymvork emas, validatsiya hal qiladi. To'g'ri kirishda uchala variant bir xil 0.6148 berdi. Pydantic bilan yozilgan Flask endpointi to'rtta yomon kirishning hammasida FastAPI bilan bir xil 422 qaytardi. Validatsiyasiz Flask esa faqat bitta holatda "yiqildi" (tolov matn — 500) — qolgan uchtasida esa 200 va ma'nosiz javob berdi. Manfiy oylar uchun 0.8928, 100 marta katta to'lov uchun 1.0, noma'lum tarif uchun 0.3481. 2-bo'lim bu jim xatolarni alohida ko'rsatadi: ular eng xavfli, chunki monitoring 500 larni ko'radi, lekin 200 bilan qaytgan noto'g'ri javoblarni ko'rmaydi — kirishdagi "tiyin" hodisasining aynan o'zi. Noma'lum tarif esa model kodidagi tarif == "oylik" taqqoslash tufayli jimgina "oylik emas" deb talqin qilindi. 3-bo'limda 200 mijoz ikki yo'l bilan baholandi. Onlayn yo'lda 200 HTTP so'rov va 200 model chaqiruvi bo'ldi, batch yo'lda — 2 so'rov va 2 chaqiruv. Natijalar aynan bir xil (0.0 farq). Faraziy narx modelida (har so'rov 2 ms, har chaqiruv 1 ms) bu taxminan 602 ga qarshi 8 ms. Raqamlar faraziy, lekin nisbat real: ko'p mijozni bitta so'rovda yuborish mumkin bo'lsa, /predict_batch ancha tejamkor.

5. To'g'ri va noto'g'ri tushunishlar

Noto'g'ri fikr	To'g'risi
"Har model API bo'lishi kerak"	Natija soatlab kutishi mumkin bo'lsa — batch yetarli va arzon
"Validatsiya — ortiqcha kod"	Pydantic bilan u sxemaning o'zi; usiz jim ma'nosiz javoblar
"Xato bo'lsa 500 qaytaraman"	Mijoz xatosi — 4xx (422/400); 500 — faqat bizning xato
"Model har so'rovda yangi yuklansin — eng so'nggisi bo'ladi"	Har so'rovda disk va deserializatsiya; lifespan da bir marta
"`/health` doim 200 qaytaradi"	Model yuklanmagan bo'lsa — 503, aks holda orkestrator aldanadi
"Hujjatni alohida yozamiz"	OpenAPI koddan avtomatik — eskirmaydi
"`TestClient` — haqiqiy sinov emas"	U butun FastAPI zanjiridan o'tadi; faqat tarmoq qatlami yo'q
"Flask yomon, FastAPI yaxshi"	Ikkalasi ham yaraydi; muhimi — validatsiya
"`--workers 4` bepul tezlik"	Har jarayon modelni alohida yuklaydi — xotira 4x

6. Keng tarqalgan xatolar va yechimlari

1. Modelni har so'rovda yuklash

python

@app.post("/predict")
def predict(m: Mijoz):
    model = joblib.load("model.joblib")                            # ⚠️
    ...
# lifespan ichida: app.state.predictor = Predictor(papka)          # ✅

2. Sxemasiz kirish

python

@app.post("/predict")
def predict(malumot: dict): ...                                    # ⚠️
@app.post("/predict")
def predict(mijoz: Mijoz): ...                                     # ✅ pydantic

3. Chegarasiz maydonlar

python

tolov: float                                                       # ⚠️ tiyin ham o'tadi
tolov: float = Field(gt=0, le=1000)                               # ✅

4. Biznes xatosiga 500

python

raise ValueError("takroriy id")                                    # ⚠️ -> 500
raise HTTPException(status_code=400, detail="takroriy mijoz_id")   # ✅

5. Tracebackni mijozga berish

python

return JSONResponse(500, {"xato": traceback.format_exc()})         # ⚠️
return JSONResponse(status_code=500, content={"xato": "ichki xato"})  # ✅ + log

6. Aldamchi /health

python

@app.get("/health")
def health(): return {"holat": "ok"}                               # ⚠️ doim
# predictor yo'q bo'lsa 503                                         # ✅

7. with siz TestClient

python

klient = TestClient(app); klient.post("/predict", ...)             # ⚠️ lifespan yo'q
with TestClient(app) as klient: ...                                # ✅

7. Integratsiya — bu bilim qayerda kerak bo'ladi

25.6-dars (o'tilgan): LLM API bilan ishlash — o'shanda API mijozi edik, endi xizmat qilamiz
27.3-dars (o'tilgan): ma'lumot quvuri — batch bashorat uning bir bosqichi
27.5-dars (o'tilgan): registr — /version javobi registrdagi versiyaga mos bo'lishi kerak
27.6-dars (o'tilgan): Predictor — API uning ustidagi yupqa qatlam
27.8-dars: API ni mustahkamlash va testlash — kontrakt, golden va chegara testlari, rate limiting, autentifikatsiya
27.9-dars: Docker bilan konteynerlash — uvicorn bilan ilova obrazga joylanadi
27.10-dars: Bulut va deploy strategiyalari — /health orkestrator uchun
27.11-dars: Monitoring — har so'rov metrikasi va logi

8. Eng yaxshi amaliyotlar

Avval so'rang: onlayn kerakmi? Aks holda batch — va ikkalasi bitta Predictor bilan.
Har kirish — pydantic sxema: Field chegaralari domen bo'yicha, Enum, aniq ixtiyoriy maydonlar, extra="forbid".
Model lifespan da bir marta yuklanadi, endpointga Depends bilan beriladi.
To'rt endpoint: /health (haqiqiy tayyorlik), /version, /predict, /predict_batch (hajm chegarasi bilan).
Har javobda model versiyasi.
Status kodlar ma'noli: 422 sxema, 400 biznes, 503 tayyor emas, 500 — faqat kutilmagan (tracebacksiz).
response_model — javob ham shartnomaga mos.
Hamma narsa TestClient bilan sinaladi, with bilan (lifespan ishlashi uchun).
OpenAPI sxemasini hujjat sifatida bering va o'zgarishini kuzating 27.8-bob.

9. Amaliy topshiriq

Vazifa 1: Bashorat qiling

python

1.  # kredit arizasi - onlayn yoki batch?
2.  # oylik marketing ro'yxati - onlayn yoki batch?
3.  # sxema buzilsa FastAPI qaysi kod qaytaradi?
4.  # takroriy mijoz_id uchun qaysi kod?
5.  # model yuklanmagan bo'lsa /health nima qaytaradi?
6.  # Field(ge=0, le=600) nimani anglatadi?
7.  # lax rejimda "12" int maydonga nima bo'ladi? 12.5 chi?
8.  # extra="forbid" nima uchun?
9.  # lifespan qachon bajariladi?
10. # TestClient ni with siz ishlatsak?
11. # app.openapi() nima qaytaradi?
12. # --workers 4 xotiraga qanday ta'sir qiladi?

Javoblar

Onlayn — mijoz kutib turibdi
Batch
422
400
503
0 <= oylar <= 600
"12" -> 12; 12.5 -> xato (int_from_float)
Noma'lum (imlo xatoli) maydonlarni rad etish
Ilova ishga tushganda (yield gacha) va to'xtaganda (yield dan keyin)
Lifespan ishlamaydi — model yuklanmaydi
OpenAPI sxemasi (lug'at): yo'llar, metodlar, sxemalar, javob kodlari
Har jarayon modelni alohida yuklaydi — taxminan 4 barobar

Vazifa 2: Xatolarni tuzating

python

1.  @app.post("/predict")
    def predict(malumot: dict):
        return {"p": joblib.load("m.joblib").predict_proba(pd.DataFrame([malumot]))[0, 1]}

2.  tolov: float

3.  @app.get("/health")
    def health(): return {"holat": "ok"}

4.  if takroriy: raise ValueError("takroriy id")

5.  mijozlar: list[Mijoz]

Javoblar

python

1.  @app.post("/predict", response_model=Bashorat)
    def predict(mijoz: Mijoz, pr: Predictor = Depends(predictor_ol)): ...
    # model lifespan da bir marta yuklanadi

2.  tolov: float = Field(gt=0, le=1000)

3.  @app.get("/health")
    def health(request: Request):
        if getattr(request.app.state, "predictor", None) is None:
            return JSONResponse(status_code=503, content={"holat": "tayyor emas"})
        return {"holat": "ok"}

4.  if takroriy: raise HTTPException(status_code=400, detail="takroriy mijoz_id")

5.  mijozlar: list[Mijoz] = Field(min_length=1, max_length=100)

Vazifa 3: Sxema

2-misoldagi Mijoz ga qo'shing:

telefon: str | None — faqat +998 bilan boshlanib, 9 raqam (pattern)
tugilgan_yil: int — 1920 va 2010 oralig'ida
model_validator — tarif == "ikki_yillik" bo'lsa, oylar >= 1 bo'lishi shart

Har qoida uchun to'g'ri va noto'g'ri kirish bilan sinang, xato type larini chop eting.

Vazifa 4: Readiness va liveness

3-misolga ikki endpoint qo'shing: /health/live (jarayon tirik — doim 200) va /health/ready (model yuklangan va nazorat namunalari o'tgan — aks holda 503). Lifespan ichida nazorat yiqilsa, ilova ishga tushmasin — TestClient bilan sinang (with bloki istisno berishi kerak).

Vazifa 5: dependency_overrides

3-misoldagi predictor_ol ni testda soxta Predictor bilan almashtiring (app.dependency_overrides[predictor_ol] = ...) — u doim 0.5 qaytarsin. /predict javobida sinf chegaraga ko'ra to'g'ri hisoblanishini tekshiring. Nega bu usul haqiqiy modelsiz tez testlar yozish imkonini beradi?

Vazifa 6: Flask versiyasi

3-misoldagi to'rt endpointni Flask da yozing (pydantic ni qo'lda chaqirib). Bir xil 10 ta kirish (to'g'ri va noto'g'ri) uchun FastAPI va Flask javoblari status kod va ehtimol bo'yicha bir xil ekanini jadval bilan tekshiring.

Vazifa 7: O'ylash

Mahsulot jamoasi so'radi: "Modelingizni API qilib bering, har sahifa ochilganda mijozning ketish ehtimolini ko'rsatamiz". Saytga kuniga 2 million sahifa ochiladi, mijozlar soni 300 ming, model esa mijozning oylik ma'lumotlaridan foydalanadi. Qanday arxitekturani taklif qilasiz?

Javob

Qisqa javob: onlayn bashorat emas, onlayn o'qish. Model oylik ma'lumotdan foydalanadi — demak bir mijozning ehtimoli oy davomida deyarli o'zgarmaydi. Har sahifa ochilishida modelni chaqirish — kuniga 2 million marta bir xil hisobni takrorlash.

1. Batch hisob + tez o'qish

Har kecha (yoki ma'lumot yangilanganda) batch skript 300 ming mijozning hammasi uchun ehtimolni hisoblaydi — bitta Predictor, bir necha daqiqa.
Natija kalit-qiymat omboriga yoziladi (Redis, ma'lumotlar bazasi jadvali): mijoz_id -> (ehtimol, model_versiyasi, hisoblangan_sana).
Sayt API si faqat shu ombordan o'qiydi — millisekundlar, model yo'q.

2. Solishtirish

	Har sahifada model	Batch + o'qish
Model chaqiruvlari/kun	2 000 000	300 000 (bitta batch da)
Kechikish	model + tarmoq	faqat o'qish
Model API ishdan chiqsa	sahifa ta'sirlanadi	kechagi natija ko'rsatiladi
Murakkablik	yuqori (masshtablash)	past

3. Qachon haqiqiy onlayn API kerak bo'ladi

Kirish ma'lumoti so'rov paytida paydo bo'lsa (masalan, joriy savat tarkibi, hozirgi sessiya harakati)
Yangi mijoz uchun (hali batch da yo'q) — bu holatda gibrid: ombordan topilmasa, /predict chaqiriladi

4. Nimani saqlab qolish kerak

Ombordagi har yozuvda model versiyasi va hisoblangan sana — xuddi /version kabi: incident paytida "bu ehtimolni qaysi model hisoblagan?" savoliga javob.

Xulosa: API — natijani yetkazish usuli, hisoblash usuli emas. Kirish ma'lumoti sekin o'zgarsa, batch da hisoblang va onlayn faqat o'qing; model API ni haqiqatan yangi ma'lumot so'rov bilan kelganda ishlating.

Nimani mustahkamlaydi: 2.1, 2.5, 2.6-bo'limlar.

Xulosa

Bu darsda Predictor ni FastAPI orqali taqdim etdik va hamma narsani serversiz, TestClient bilan sinadik.

Eng muhim uch fikr:

Turlar va pydantic sxemasi — API shartnomasi. 1-misolda FastAPI yo'l va query parametrlarini turga o'girdi va noto'g'ri qiymatlarga biz hech qanday if yozmasdan 422 qaytardi. 2-misolda bitta Mijoz sxemasi Field chegaralari, Enum, ixtiyoriy maydonlar va extra="forbid" bilan beshta xatoni birdaniga ushladi, imlo xatoli region maydonini rad etdi va JSON Schema ni avtomatik berdi.
Model bir marta yuklanadi, xatolar ma'noli kod oladi. 3-misolda lifespan modelni bitta marta yukladi — yigirmadan ortiq so'rovdan keyin ham yuklashlar soni 1. Lifespan ishlamaganda esa /health va /predict halol 503 qaytardi. Sxema xatolari 422, biznes qoidasi 400, noto'g'ri yo'l 404, ichki xato esa tracebacksiz 500 oldi. Har javobda model versiyasi bor, OpenAPI esa endpointlar va javob kodlarini avtomatik hujjatladi.
Freymvork emas, validatsiya muhim; batch — kam chaqiruv. 4-misolda pydantic bilan Flask FastAPI bilan bir xil ishladi. Validatsiyasiz Flask esa uchta yomon kirishga 200 va ma'nosiz javob berdi (masalan, 100 marta katta to'lovga 1.0). 200 mijoz uchun /predict_batch 200 o'rniga 2 ta model chaqiruvi bilan aynan bir xil natija berdi.

Keyingi darsda API ni mustahkamlash va testlash: kontrakt, golden va chegara testlari, so'rov hajmini cheklash, timeout va zaxira javob, kesh, idempotentlik, strukturali loglar, API kaliti va rate limiting.

Ulashish:Telegram'da

Izohlar (0)

Izoh yozish uchun kiring.

Hozircha izoh yo'q. Birinchi bo'ling!

27.7-dars: FastAPI bilan model API

1. Kirish va motivatsiya

2. Nazariya — chuqur tushuntirish

2.1. Nega API: onlayn va batch bashorat

2.2. REST asoslari

2.3. FastAPI: turlar — shartnoma

2.4. Pydantic sxemalar

2.5. Modelni bir marta yuklash: lifespan

2.6. To'rt endpoint

2.7. Xato javoblar

2.8. Avtomatik OpenAPI sxemasi

2.9. TestClient va import tuzog'i

2.10. Ishga tushirish: uvicorn

2.11. Flask bilan qisqa taqqoslash

2.12. Tuzoqlar

3. Tez ma'lumotnoma

4. Batafsil misollar

Misol 1 — REST asoslari: minimal FastAPI ilova

Misol 2 — Pydantic sxemalar: chegaralar, Enum, ixtiyoriy maydonlar

Misol 3 — To'liq model API: lifespan va to'rt endpoint

Misol 4 — Flask bilan taqqoslash va onlayn/batch

5. To'g'ri va noto'g'ri tushunishlar

6. Keng tarqalgan xatolar va yechimlari

7. Integratsiya — bu bilim qayerda kerak bo'ladi

8. Eng yaxshi amaliyotlar

9. Amaliy topshiriq

Xulosa

Izohlar (0)

2.5. Modelni bir marta yuklash: `lifespan`

2.9. `TestClient` va import tuzog'i