Tashqi dunyo: HTTP, timeout va qayta urinish

"""Sinov uchun kichik HTTP server: turli xatolarni ataylab qaytaradi."""
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

HOLAT = {"urinishlar": 0}


class Ishlovchi(BaseHTTPRequestHandler):
    def log_message(self, *args):          # jim turadi
        pass

    def _javob(self, kod: int, tana: dict) -> None:
        xom = json.dumps(tana).encode()
        self.send_response(kod)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(xom)))
        self.end_headers()
        self.wfile.write(xom)

    def do_GET(self):
        yol = self.path
        if yol == "/ok":
            self._javob(200, {"holat": "yaxshi", "talabalar": 3})
        elif yol == "/sekin":
            time.sleep(3)
            self._javob(200, {"holat": "kech javob"})
        elif yol == "/xato500":
            self._javob(500, {"xato": "server ichki xatosi"})
        elif yol == "/topilmadi":
            self._javob(404, {"xato": "topilmadi"})
        elif yol == "/uchinchida":
            HOLAT["urinishlar"] += 1
            if HOLAT["urinishlar"] < 3:
                self._javob(503, {"xato": "band", "urinish": HOLAT["urinishlar"]})
            else:
                self._javob(200, {"holat": "nihoyat", "urinish": HOLAT["urinishlar"]})
        elif yol == "/nolla":
            HOLAT["urinishlar"] = 0
            self._javob(200, {"holat": "nollandi"})
        else:
            self._javob(404, {"xato": "noma'lum yo'l"})


def ishga_tushir(port: int = 0) -> tuple[HTTPServer, str]:
    server = HTTPServer(("127.0.0.1", port), Ishlovchi)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, f"http://127.0.0.1:{server.server_port}"

import sys
sys.path.insert(0, str(ISH))
import time
import urllib.error
import urllib.request
from sinov_server import ishga_tushir

server, manzil = ishga_tushir()
print("server:", manzil)

def sorov(yol, timeout):
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(manzil + yol, timeout=timeout) as j:
            return j.status, round((time.perf_counter() - t0) * 1000), None
    except Exception as e:
        return None, round((time.perf_counter() - t0) * 1000), type(e).__name__

import pandas as pd
qatorlar = []
for yol, t in [("/ok", 5), ("/sekin", 1), ("/sekin", 5)]:
    kod, ms, xato = sorov(yol, t)
    qatorlar.append({"yo'l": yol, "timeout, s": t, "kod": kod, "vaqt, ms": ms, "xato": xato or "-"})
show(pd.DataFrame(qatorlar))

import urllib.error, urllib.request, json

def olish(manzil, yol, timeout=5):
    """(ma'lumot, xato_matni) qaytaradi."""
    try:
        with urllib.request.urlopen(manzil + yol, timeout=timeout) as j:
            return json.loads(j.read()), None
    except urllib.error.HTTPError as e:
        return None, f"HTTP {e.code}"
    except urllib.error.URLError as e:
        return None, f"ulanib bo'lmadi: {e.reason}"
    except json.JSONDecodeError:
        return None, "javob JSON emas"

import pandas as pd
qatorlar = []
for yol in ["/ok", "/xato500", "/topilmadi", "/nomalum"]:
    malumot, xato = olish(manzil, yol)
    qatorlar.append({"yo'l": yol, "ma'lumot": str(malumot), "xato": xato or "-"})
qatorlar.append({"yo'l": "(o'chiq port)", "ma'lumot": str(olish("http://127.0.0.1:1", "/x", 2)[0]),
                 "xato": olish("http://127.0.0.1:1", "/x", 2)[1]})
show(pd.DataFrame(qatorlar))

"""HTTP so'rovlari: timeout, qayta urinish, backoff."""
import json
import logging
import time
import urllib.error
import urllib.request
from typing import Any

log = logging.getLogger(__name__)

QAYTA_URINSA_BOLADI = {408, 429, 500, 502, 503, 504}


class TarmoqXato(Exception):
    """So'rov bajarilmadi."""


def olish(
    manzil: str,
    *,
    timeout: float = 5.0,
    urinishlar: int = 3,
    boshlangich_kutish: float = 0.1,
) -> dict[str, Any]:
    """JSON oladi. Vaqtinchalik xatolarda qayta urinadi (eksponensial backoff).

    Doimiy xatolarda (404, 401) darhol to'xtaydi — qayta urinish ma'nosiz.
    """
    oxirgi = ""
    for urinish in range(1, urinishlar + 1):
        try:
            with urllib.request.urlopen(manzil, timeout=timeout) as j:
                malumot: dict[str, Any] = json.loads(j.read())
                return malumot
        except urllib.error.HTTPError as e:
            oxirgi = f"HTTP {e.code}"
            if e.code not in QAYTA_URINSA_BOLADI:
                log.debug("doimiy xato %s — qayta urinilmaydi", oxirgi)
                raise TarmoqXato(f"{manzil}: {oxirgi}") from None
        except urllib.error.URLError as e:
            oxirgi = f"ulanib bo'lmadi: {e.reason}"
        except TimeoutError:
            oxirgi = f"timeout ({timeout} s)"

        if urinish < urinishlar:
            kutish = boshlangich_kutish * (2 ** (urinish - 1))
            log.warning("%s (%d-urinish), %.2f s dan keyin qayta urinaman", oxirgi, urinish, kutish)
            time.sleep(kutish)

    raise TarmoqXato(f"{manzil}: {urinishlar} urinishdan keyin ham bajarilmadi ({oxirgi})")

import logging, sys, importlib, time
sys.path.insert(0, str(ISH / "src"))
import hisobot.tarmoq as tarmoq
importlib.reload(tarmoq)

logging.basicConfig(level=logging.WARNING, format="  [%(levelname)s] %(message)s",
                    stream=sys.stdout, force=True)

print("1) Uchinchi urinishda muvaffaqiyat:")
tarmoq.olish(manzil + "/nolla")
t0 = time.perf_counter()
natija = tarmoq.olish(manzil + "/uchinchida")
print("   natija:", natija, f"| jami {(time.perf_counter() - t0) * 1000:.0f} ms")

print("\n2) Doimiy xato (404) — qayta urinish YO'Q:")
t0 = time.perf_counter()
try:
    tarmoq.olish(manzil + "/topilmadi")
except tarmoq.TarmoqXato as e:
    print("   xato:", e, f"| {(time.perf_counter() - t0) * 1000:.0f} ms")

print("\n3) Vaqtinchalik xato (500) — uch urinish ham yetmadi:")
t0 = time.perf_counter()
try:
    tarmoq.olish(manzil + "/xato500")
except tarmoq.TarmoqXato as e:
    print("   xato:", e, f"| {(time.perf_counter() - t0) * 1000:.0f} ms")

import pandas as pd
qatorlar = []
for urinish in range(1, 7):
    qatorlar.append({
        "urinish": urinish,
        "doimiy kutish (0.5 s)": f"{0.5:.2f} s",
        "eksponensial (0.1 s dan)": f"{0.1 * 2 ** (urinish - 1):.2f} s",
    })
show(pd.DataFrame(qatorlar))
print("6 urinishda jami kutish — doimiy:", 6 * 0.5, "s | eksponensial:",
      round(sum(0.1 * 2 ** i for i in range(6)), 2), "s")

import sys
sys.path.insert(0, str(ISH))
try:
    import requests
    j = requests.get(manzil + "/ok", timeout=5)
    print("requests:", j.status_code, j.json())

    s = requests.Session()                      # ulanishni qayta ishlatadi — tezroq
    import time
    t0 = time.perf_counter()
    for _ in range(20):
        requests.get(manzil + "/ok", timeout=5)
    sessiyasiz = (time.perf_counter() - t0) * 1000
    t0 = time.perf_counter()
    for _ in range(20):
        s.get(manzil + "/ok", timeout=5)
    sessiya_bilan = (time.perf_counter() - t0) * 1000
    print(f"20 so'rov — sessiyasiz: {sessiyasiz:.0f} ms | sessiya bilan: {sessiya_bilan:.0f} ms")
    s.close()
except ImportError:
    print("requests o'rnatilmagan")

import json
import urllib.error
from unittest.mock import MagicMock, patch

import pytest

from hisobot.tarmoq import TarmoqXato, olish


def soxta_javob(malumot):
    j = MagicMock()
    j.read.return_value = json.dumps(malumot).encode()
    j.__enter__.return_value = j
    j.__exit__.return_value = False
    return j


def test_muvaffaqiyat():
    with patch("urllib.request.urlopen", return_value=soxta_javob({"a": 1})):
        assert olish("http://x/y") == {"a": 1}


def test_404_qayta_urinmaydi():
    xato = urllib.error.HTTPError("http://x/y", 404, "yo'q", {}, None)
    with (
        patch("urllib.request.urlopen", side_effect=xato) as m,
        pytest.raises(TarmoqXato, match="404"),
    ):
        olish("http://x/y", urinishlar=3)
    assert m.call_count == 1                     # faqat BIR marta chaqirilgan


def test_500_uch_marta_urinadi():
    xato = urllib.error.HTTPError("http://x/y", 500, "xato", {}, None)
    with (
        patch("urllib.request.urlopen", side_effect=xato) as m,
        pytest.raises(TarmoqXato),
    ):
        olish("http://x/y", urinishlar=3, boshlangich_kutish=0.001)
    assert m.call_count == 3


def test_ikkinchida_tuzaladi():
    xato = urllib.error.HTTPError("http://x/y", 503, "band", {}, None)
    with patch("urllib.request.urlopen", side_effect=[xato, soxta_javob({"ok": True})]) as m:
        assert olish("http://x/y", urinishlar=3, boshlangich_kutish=0.001) == {"ok": True}
    assert m.call_count == 2

.venv/bin/python3 -m pytest tests/test_tarmoq.py -v