پایتون از صفرپایتون در کار واقعیدرس ۱۰ از ۱۰

پروژه: ابزار و API

پروژه‌ی پایانی دوره‌ی پایتون: ابزار پایش opscheck که از یک فایل JSON چک‌های دیسک، پورت و HTTP را می‌خواند و اجرا می‌کند؛ با کلاس و dataclass، مدیریت خطا، خط فرمان با argparse، گزارش JSON و کد خروج، یک API با Flask برای گزارش، و اجرای خودکار با cron.

حدود ۱۳ دقیقه مطالعه
۳ تمرین با جواب
مقدماتی تا متوسط
به‌روزرسانی: ۲۹ مهر ۱۴۰۵

توی این پروژه می‌سازی یک ابزار پایش واقعی به اسم opscheck: یک فایل config.json می‌گوید چه چیزهایی بررسی شود (پر بودن دیسک، باز بودن یک پورت، جواب دادن یک آدرس HTTP)، برنامه همه را اجرا می‌کند، نتیجه را خوانا چاپ می‌کند، در صورت درخواست گزارش JSON می‌نویسد و با کد خروج می‌گوید همه‌چیز سالم بود یا نه. بعد همان چک‌ها را با یک API کوچک در دسترس می‌گذاری و با cron خودکار می‌کنی. هر چه در این دوره یاد گرفتی اینجا به کار می‌آید: کلاس و وراثت، dataclass، دیکشنری و لیست، فایل و JSON، try/except، ماژول، و Flask.

مسئله

تا حالا هر درس یک مهارت جدا بود. برنامه‌ی واقعی همه را با هم لازم دارد، و چند چیز دیگر که در درس‌ها کمتر دیدی: ساختار پروژه (کد را در چند فایل تقسیم کنی)، قابل‌تنظیم بودن (چک‌ها در فایل باشند، نه داخل کد) و رفتار درست در شکست (یک چک خراب، بقیه را خراب نکند).

نقشه‌ی پروژه

سه فایل پایتون و یک فایل تنظیمات: checks.py منطق چک‌ها، opscheck.py خط فرمان، api.py نسخه‌ی وب.

config.json خوانده می‌شود، برای هر عضو یک شیء چک ساخته می‌شود، همه اجرا می‌شوند و نتیجه‌ها به سه شکل خروجی می‌روند.

آماده‌سازی:

mkdir -p ~/py/opscheck
cd ~/py/opscheck
python3 -m venv .venv
source .venv/bin/activate
pip install requests flask

قدم ۱: فایل تنظیمات

[
  {"type": "disk", "name": "root-disk", "path": "/", "max_percent": 90},
  {"type": "port", "name": "postgres", "host": "127.0.0.1", "port": 5432},
  {"type": "port", "name": "ssh", "host": "127.0.0.1", "port": 22},
  {"type": "http", "name": "web", "url": "http://127.0.0.1:8000/"}
]

هر چک یک دیکشنری است: type نوعش را می‌گوید و بقیه‌ی کلیدها تنظیمات همان نوع‌اند. افزودن یک چک جدید یعنی یک خط به این فایل، بدون دست زدن به کد. (دیتابیس PostgreSQL را در دوره‌ی SQL نصب می‌کنی؛ اینجا عمداً گذاشته‌ایم تا یک چک ناموفق هم ببینی.)

قدم ۲: چک‌ها، `checks.py`

"""چک‌های پایش: دیسک، پورت و HTTP."""
import shutil
import socket
from dataclasses import dataclass

import requests


@dataclass
class Result:
    name: str
    ok: bool
    detail: str


class Check:
    def __init__(self, name):
        self.name = name

    def run(self):
        raise NotImplementedError

    def safe_run(self):
        try:
            return self.run()
        except Exception as e:  # هر خطای پیش‌بینی‌نشده = چک ناموفق
            return Result(self.name, False, f"خطا: {e}")


class DiskCheck(Check):
    def __init__(self, name, path="/", max_percent=90):
        super().__init__(name)
        self.path = path
        self.max_percent = max_percent

    def run(self):
        usage = shutil.disk_usage(self.path)
        percent = round(usage.used / usage.total * 100, 1)
        ok = percent <= self.max_percent
        return Result(self.name, ok, f"{self.path} {percent}٪ پر (حد {self.max_percent}٪)")


class PortCheck(Check):
    def __init__(self, name, host, port, timeout=2):
        super().__init__(name)
        self.host = host
        self.port = port
        self.timeout = timeout

    def run(self):
        try:
            with socket.create_connection((self.host, self.port), timeout=self.timeout):
                return Result(self.name, True, f"{self.host}:{self.port} باز است")
        except OSError as e:
            return Result(self.name, False, f"{self.host}:{self.port} بسته است ({e.strerror or e})")


class HttpCheck(Check):
    def __init__(self, name, url, expect=200, timeout=3):
        super().__init__(name)
        self.url = url
        self.expect = expect
        self.timeout = timeout

    def run(self):
        try:
            r = requests.get(self.url, timeout=self.timeout)
        except requests.exceptions.RequestException as e:
            return Result(self.name, False, f"{self.url} در دسترس نیست ({type(e).__name__})")
        ok = r.status_code == self.expect
        return Result(self.name, ok, f"{self.url} کد {r.status_code}")


TYPES = {"disk": DiskCheck, "port": PortCheck, "http": HttpCheck}


def build_checks(config):
    checks = []
    for item in config:
        item = dict(item)
        kind = item.pop("type")
        if kind not in TYPES:
            raise ValueError(f"نوع چک ناشناخته: {kind}")
        checks.append(TYPES[kind](**item))
    return checks

تصمیم‌های طراحی، یکی‌یکی:

Result یک dataclass است (درس کلاس): هر چک، از هر نوعی، یک نتیجه‌ی هم‌شکل برمی‌گرداند. بقیه‌ی برنامه فقط با Result کار می‌کند و لازم نیست نوع چک را بداند.
Check کلاس پایه است و هر نوع، run خودش را بازنویسی می‌کند (وراثت). raise NotImplementedError در کلاس پایه یعنی «فرزندها باید این را بنویسند».
safe_run تور ایمنی است: اگر یک چک خطای پیش‌بینی‌نشده داد (مثلاً مسیر دیسکی که وجود ندارد)، برنامه نمی‌میرد؛ آن چک فقط «ناموفق» می‌شود. این یکی از معدود جاهایی است که except Exception معنی دارد: لایه‌ی بیرونی‌ای که باید همه‌چیز را ادامه بدهد، و خطا را هم گم نمی‌کند (در detail می‌نویسد).
shutil.disk_usage همان df است، socket.create_connection همان nc -z (درس پورت‌ها) و requests.get همان curl. هر سه timeout دارند تا یک سرور مرده کل برنامه را گیر نیندازد.
TYPES یک دیکشنری از اسم نوع به کلاس است؛ TYPES[kind](**item) کلاس مناسب را با بقیه‌ی کلیدهای دیکشنری به‌عنوان آرگومان با اسم می‌سازد. ** یعنی «این دیکشنری را به آرگومان‌های کلید=مقدار باز کن». برای افزودن نوع جدید، فقط یک کلاس و یک خط در TYPES لازم است (تمرین سخت).

⚡ بررسی سریع

اگر DiskCheck برای مسیر /no/such خطای FileNotFoundError بدهد، opscheck چه می‌کند؟

جواب درست: safe_run خطا را می‌گیرد و آن چک را ناموفق با پیام خطا گزارش می‌کند؛ بقیه‌ی چک‌ها اجرا می‌شوند

safe_run هر خطا را به یک Result ناموفق تبدیل می‌کند.

قدم ۳: خط فرمان، `opscheck.py`

"""opscheck: اجرای چک‌ها از روی فایل تنظیمات و گزارش."""
import argparse
import json
import sys
from dataclasses import asdict
from datetime import datetime

from checks import build_checks


def load_config(path):
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        sys.exit(f"خطا: فایل تنظیمات {path} پیدا نشد")
    except json.JSONDecodeError as e:
        sys.exit(f"خطا: {path} خط {e.lineno}: {e.msg}")


def run_all(config):
    return [c.safe_run() for c in build_checks(config)]


def main():
    parser = argparse.ArgumentParser(description="پایش ساده‌ی سرور")
    parser.add_argument("-c", "--config", default="config.json", help="فایل تنظیمات")
    parser.add_argument("--json", metavar="FILE", help="گزارش را در این فایل JSON هم بنویس")
    args = parser.parse_args()

    try:
        results = run_all(load_config(args.config))
    except (ValueError, TypeError) as e:
        sys.exit(f"خطا در تنظیمات: {e}")
    for r in results:
        mark = "✔" if r.ok else "✘"
        print(f"{mark} {r.name:<10} {r.detail}")

    failed = [r for r in results if not r.ok]
    print(f"--- {len(results) - len(failed)} از {len(results)} سالم")

    if args.json:
        report = {
            "time": datetime.now().isoformat(timespec="seconds"),
            "ok": not failed,
            "results": [asdict(r) for r in results],
        }
        with open(args.json, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)

    sys.exit(1 if failed else 0)


if __name__ == "__main__":
    main()

argparse (کتابخانه‌ی استاندارد) آرگومان‌های خط فرمان را با گزینه‌های -c و --json می‌خواند و -h (راهنما) را خودکار می‌سازد. همان کاری که در دوره‌ی Bash با getopts می‌کردی، ولی خیلی کامل‌تر.
sys.exit("پیام") پیام را روی stderr چاپ می‌کند و با کد خروج ۱ تمام می‌شود؛ برای خطاهای تنظیمات که ادامه دادن معنی ندارد.
{r.name:<10} اسم را در ۱۰ کاراکتر چپ‌چین می‌کند تا ستون‌ها مرتب شوند.
asdict(r) یک dataclass را به دیکشنری تبدیل می‌کند تا json.dump بتواند بنویسدش.
کد خروج: صفر اگر همه سالم، یک اگر نه. این اسکریپت را قابل‌ترکیب می‌کند: ./opscheck.py && echo ok یا cron و systemd که از روی کد خروج تصمیم می‌گیرند (درس ترکیب دستورها).

قدم ۴: اجرا

یک وب‌سرور ساده روی پورت ۸۰۰۰ برای چک web بالا بیاور (در یک ترمینال جدا؛ همان ماژول http.server پایتون):

mkdir -p site && echo hi > site/index.html
python3 -m http.server 8000 --bind 127.0.0.1 --directory site

و در ترمینال اصلی:

python3 opscheck.py --json report.json
echo "کد خروج: $?"

✔ root-disk  / 5.2٪ پر (حد 90٪)
✘ postgres   127.0.0.1:5432 بسته است (Connection refused)
✔ ssh        127.0.0.1:22 باز است
✔ web        http://127.0.0.1:8000/ کد 200
--- 3 از 4 سالم
کد خروج: 1

cat report.json

{
  "time": "2026-10-21T20:02:18",
  "ok": false,
  "results": [
    {
      "name": "root-disk",
      "ok": true,
      "detail": "/ 5.2٪ پر (حد 90٪)"
    },
    {
      "name": "postgres",
      "ok": false,
      "detail": "127.0.0.1:5432 بسته است (Connection refused)"
    },
    {
      "name": "ssh",
      "ok": true,
      "detail": "127.0.0.1:22 باز است"
    },
    {
      "name": "web",
      "ok": true,
      "detail": "http://127.0.0.1:8000/ کد 200"
    }
  ]
}

راهنمای خودکار:

python3 opscheck.py -h

usage: opscheck.py [-h] [-c CONFIG] [--json FILE]

پایش ساده‌ی سرور

options:
  -h, --help            show this help message and exit
  -c CONFIG, --config CONFIG
                        فایل تنظیمات
  --json FILE           گزارش را در این فایل JSON هم بنویس

و رفتار با تنظیمات خراب (همه با کد خروج ۱):

python3 opscheck.py -c nope.json
echo '[{"type": "ftp", "name": "x"}]' > bad.json
python3 opscheck.py -c bad.json
echo '[{"type": "disk", "name": "x", "pathh": "/"}]' > bad2.json
python3 opscheck.py -c bad2.json

خطا: فایل تنظیمات nope.json پیدا نشد
خطا در تنظیمات: نوع چک ناشناخته: ftp
خطا در تنظیمات: DiskCheck.__init__() got an unexpected keyword argument 'pathh'

سومی جالب است: غلط تایپی در فایل تنظیمات (pathh) به‌خاطر **item به یک TypeError تبدیل شد که دقیقاً اسم کلید اشتباه را می‌گوید.

قدم ۵: API، `api.py`

"""API برای دیدن آخرین وضعیت چک‌ها."""
from dataclasses import asdict

from flask import Flask

from opscheck import load_config, run_all

app = Flask(__name__)
app.json.ensure_ascii = False
app.json.sort_keys = False


@app.get("/report")
def report():
    results = run_all(load_config("config.json"))
    ok = all(r.ok for r in results)
    body = {"ok": ok, "results": [asdict(r) for r in results]}
    return body, (200 if ok else 503)


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5001)

همان تابع‌های opscheck.py را import کردیم (و به لطف if __name__ == "__main__"، main() آن اجرا نمی‌شود). کد ۵۰۳ (Service Unavailable) برای «چیزی خراب است» انتخاب شده تا ابزارهای پایش (و Nginx، و حتی healthcheck داکر) فقط با کد بفهمند وضعیت چیست:

python3 api.py &
curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:5001/report
curl -s http://127.0.0.1:5001/report | python3 -m json.tool

503
{
    "ok": false,
    "results": [
        {
            "name": "root-disk",
            "ok": true,
            "detail": "/ 5.2٪ پر (حد 90٪)"
        },
...

json.tool فارسی را به \u... تبدیل می‌کند؛ با python3 -m json.tool --no-ensure-ascii خوانا می‌ماند. بدون debug، Flask JSON را فشرده (در یک خط) می‌فرستد که برای ماشین‌ها بهتر است.

قدم ۶: خودکار کردن با cron

برای اجرای هر پنج دقیقه، با crontab -e (دوره‌ی لینوکس، درس cron) این خط را اضافه کن. چون cron محیط محدودی دارد، مسیرهای کامل می‌دهیم و پایتونِ داخل venv را صدا می‌زنیم:

*/5 * * * * cd /home/ali/py/opscheck && .venv/bin/python opscheck.py --json /home/ali/py/opscheck/report.json >> /home/ali/py/opscheck/cron.log 2>&1

.venv/bin/python خودش محیط مجازی را به کار می‌گیرد، پس activate لازم نیست. حالا report.json همیشه تازه است و cron.log تاریخچه را نگه می‌دارد.

چک‌لیست نهایی

✓ چک‌لیست پروژه‌ی opscheck

config.json چک‌ها را تعریف می‌کند و برای افزودن چک، کد عوض نمی‌شود
هر چک یک Result هم‌شکل برمی‌گرداند و خطای یک چک بقیه را متوقف نمی‌کند
همه‌ی اتصال‌های شبکه timeout دارند
-h راهنمای خوانا می‌دهد و تنظیمات خراب پیام روشن و کد خروج ۱ می‌دهد
کد خروج صفر یعنی همه سالم
--json گزارش ماندگار با زمان می‌نویسد
API کد ۲۰۰ یا ۵۰۳ می‌دهد
cron با مسیر کامل و پایتونِ venv اجرایش می‌کند

اشتباهات رایج

۱) اجرا با پایتونِ بیرون از venv

ModuleNotFoundError: No module named 'requests' یعنی محیط فعال نیست. راه‌حل: source .venv/bin/activate یا مستقیم .venv/bin/python opscheck.py.

۲) مسیر نسبی در cron

cron در پوشه‌ی خانه اجرا می‌شود و config.json را پیدا نمی‌کند. راه‌حل: cd /مسیر/کامل && ... یا -c /مسیر/کامل/config.json.

۳) چک بدون timeout

یک requests.get بدون timeout به سرور مرده می‌تواند دقیقه‌ها گیر کند و اجرای بعدی cron روی قبلی سوار شود. راه‌حل: همیشه timeout.

۴) گرفتن همه‌ی خطاها در همه‌جا

except Exception فقط در safe_run (لایه‌ی بیرونی، با ثبت پیام) معنی دارد. داخل runها خطاهای مشخص را بگیر (OSError، RequestException) تا باگ‌های واقعی پنهان نشوند.

تمرین

تمرینآسان

یک چک دیسک دوم برای /tmp با حد ۵۰٪ به config.json اضافه کن و دوباره اجرا کن. کدام فایل‌ها را عوض کردی؟

دیدن جواب

فقط config.json:

  {"type": "disk", "name": "tmp-disk", "path": "/tmp", "max_percent": 50},

✔ tmp-disk   /tmp 5.2٪ پر (حد 50٪)

هیچ کدی عوض نشد؛ هدف اصلی طراحی همین بود. (اگر /tmp روی همان پارتیشن / باشد، درصدش هم یکی است.)

تمرینمتوسط

تمرین اصلی: گزینه‌ی --quiet به opscheck.py اضافه کن که فقط چک‌های ناموفق را چاپ کند (و خط خلاصه را). برای cron مفید است: لاگ فقط وقتی چیزی خراب است پر می‌شود. (راهنما: action="store_true" یک گزینه‌ی بدون مقدار می‌سازد.)

دیدن جواب

    parser.add_argument("-q", "--quiet", action="store_true", help="فقط ناموفق‌ها را چاپ کن")
    ...
    for r in results:
        if args.quiet and r.ok:
            continue
        mark = "✔" if r.ok else "✘"
        print(f"{mark} {r.name:<10} {r.detail}")

python3 opscheck.py -q

✘ postgres   127.0.0.1:5432 بسته است (Connection refused)
--- 3 از 4 سالم

args.quiet بدون -q برابر False و با آن True است.

تمرینسخت

یک نوع چک جدید بساز: FileAgeCheck که بررسی کند یک فایل (مثلاً آخرین بکاپ) کمتر از N ساعت پیش تغییر کرده باشد. با type: "file_age" و کلیدهای path و max_hours. (راهنما: Path(p).stat().st_mtime زمان آخرین تغییر به ثانیه است و time.time() زمان فعلی.)

دیدن جواب

import time
from pathlib import Path


class FileAgeCheck(Check):
    def __init__(self, name, path, max_hours=24):
        super().__init__(name)
        self.path = path
        self.max_hours = max_hours

    def run(self):
        age_h = (time.time() - Path(self.path).stat().st_mtime) / 3600
        ok = age_h <= self.max_hours
        return Result(self.name, ok, f"{self.path} {age_h:.1f} ساعت پیش (حد {self.max_hours})")


TYPES = {"disk": DiskCheck, "port": PortCheck, "http": HttpCheck, "file_age": FileAgeCheck}

  {"type": "file_age", "name": "backup", "path": "/home/ali/py/opscheck/report.json", "max_hours": 1},

✔ backup     /home/ali/py/opscheck/report.json 0.1 ساعت پیش (حد 1)

اگر فایل نباشد، stat() خطای FileNotFoundError می‌دهد و safe_run آن را به یک نتیجه‌ی ناموفق تبدیل می‌کند؛ بدون اینکه در FileAgeCheck کد اضافه‌ای بنویسی.

آزمونک

چرا چک‌ها در config.json هستند نه در کد؟
TYPES[kind](**item) چه می‌کند؟
چرا opscheck با کد خروج 1 تمام می‌شود وقتی چکی ناموفق است؟
API چرا در حالت خرابی 503 برمی‌گرداند نه 200؟
در cron چرا .venv/bin/python opscheck.py؟

جمع‌بندی

یک ابزار واقعی ساختی: تنظیمات در JSON، منطق در کلاس‌ها، خط فرمان با argparse، گزارش JSON، کد خروج، API و cron.
طراحی خوب: نتیجه‌ی هم‌شکل (Result)، وراثت برای انواع چک، دیکشنری TYPES برای افزودن نوع جدید با یک خط.
رفتار درست در شکست: safe_run برای هر چک، timeout برای هر اتصال، پیام روشن و کد خروج برای تنظیمات خراب.
برنامه‌ات با بقیه‌ی دنیا حرف می‌زند: کد خروج برای shell، JSON برای داشبورد، کد HTTP برای پایش.
قدم بعد: gunicorn + systemd + Nginx (دوره‌ی Nginx)، و ذخیره‌ی تاریخچه در دیتابیس (دوره‌ی SQL).

برگه‌ی تقلب این درس

دستور	کاری که می‌کند
`python3 opscheck.py`	اجرای همه‌ی چک‌ها
`python3 opscheck.py -c other.json`	فایل تنظیمات دیگر
`python3 opscheck.py --json report.json`	نوشتن گزارش
`python3 opscheck.py -h`	راهنمای خودکار argparse
`echo $?`	۰ همه سالم، ۱ مشکل
`python3 api.py`	API روی 127.0.0.1:5001/report
`/5 * * * cd ... && .venv/bin/python opscheck.py`	اجرای هر پنج دقیقه با cron

مسئله

نقشه‌ی پروژه

قدم ۱: فایل تنظیمات

قدم ۲: چک‌ها، checks.py

قدم ۳: خط فرمان، opscheck.py

قدم ۴: اجرا

قدم ۵: API، api.py

قدم ۶: خودکار کردن با cron

چک‌لیست نهایی

اشتباهات رایج

۱) اجرا با پایتونِ بیرون از venv

۲) مسیر نسبی در cron

۳) چک بدون timeout

۴) گرفتن همه‌ی خطاها در همه‌جا

تمرین

آزمونک

جمع‌بندی

قدم ۲: چک‌ها، `checks.py`

قدم ۳: خط فرمان، `opscheck.py`

قدم ۵: API، `api.py`