تُعرِّف قدرتان أساسيتان معظمَ سكريبتات DevOps الحقيقية: التنقل في نظام الملفات واستدعاء برامج خارجية. سواء كنت تُحلِّل ملف إعداد خدمة، أو تُدير ملفات السجل، أو تُغلِّف أداة CLI ضمن أتمتة، فأنت بحاجة إلى إتقان هذه المهام بأمان وبشكل محمول دون مفاجآت في بيئة الإنتاج. تتناول هذه الدرس الأنماط الحديثة في Python لكلا القدرتين.
لماذا pathlib بدلاً من os.path
قبل Python 3.6 كان النهج المعتاد هو وحدة os.path — مجموعة دوال تتعامل مع المسارات كنصوص عادية. أما pathlib (أُضيفت في 3.4 وأصبحت اصطلاحية من 3.6 فصاعداً) فتمنحك كائنات مسار تعرف أنها مسارات. يعتمد دمج المسارات على عامل /، وتُعالَج الفروقات بين الأنظمة (مائل للخلف في Windows مقابل مائل للأمام في POSIX) تلقائياً، ويحمل الكائن توابع لكل عملية شائعة.
from pathlib import Path
# بناء المسارات بأمان — دون الحاجة لـ os.path.join() يدوياً
base = Path("/etc/myapp")
config = base / "config" / "settings.yaml"
print(config) # /etc/myapp/config/settings.yaml
print(config.name) # settings.yaml
print(config.stem) # settings
print(config.suffix) # .yaml
print(config.parent) # /etc/myapp/config
# فحوصات شائعة
print(config.exists())
print(config.is_file())
print(config.is_dir())
# التكرار على شجرة مجلد
log_dir = Path("/var/log/nginx")
for log_file in log_dir.glob("*.log"):
print(log_file)
# glob متكرر — البحث عن كل .conf تحت /etc
for conf in Path("/etc").rglob("*.conf"):
print(conf)
في سكريبتات الإنتاج احرص دائماً على تحويل المسارات إلى صيغتها المطلقة الكاملة باستخدام Path(...).resolve(). يُزيل ذلك غموض الروابط الرمزية ويضمن أن السكريبت يتصرف بالطريقة نفسها سواء استُدعي من جذر المشروع أو من مهمة cron بمجلد عمل مختلف.
قراءة الملفات وكتابتها بأمان
يُعدّ استخدام open() المدمجة مع كتلة with النمط المعياري — فمدير السياق يضمن إغلاق مقبض الملف حتى عند وقوع استثناء. بالنسبة لملفات الإعداد الصغيرة (أقل من بضعة ميجابايتات) فإن read_text() وwrite_text() على كائن Path أكثر إيجازاً.
from pathlib import Path
config_path = Path("/etc/myapp/settings.yaml")
# قراءة الملف كاملاً كنص (UTF-8 افتراضياً)
raw = config_path.read_text(encoding="utf-8")
# القراءة سطراً سطراً — مفضّلة للملفات الكبيرة
with config_path.open(encoding="utf-8") as fh:
for line in fh:
line = line.rstrip("\n")
print(line)
# الكتابة الذرية — اكتب في ملف مؤقت ثم أعِد التسمية
# إعادة التسمية على نفس نظام الملفات ذرية في Linux/macOS؛
# أي تعطل أثناء الكتابة لن يترك ملف إعداد منقوصاً.
import tempfile, os
def write_atomic(path: Path, content: str) -> None:
tmp_fd, tmp_path = tempfile.mkstemp(
dir=path.parent, prefix=".tmp_"
)
try:
with os.fdopen(tmp_fd, "w", encoding="utf-8") as fh:
fh.write(content)
os.replace(tmp_path, path) # ذري في POSIX
except Exception:
os.unlink(tmp_path)
raise
write_atomic(config_path, raw.replace("debug: true", "debug: false"))
لا تستخدم أبداً path.write_text(content) مباشرةً على ملف إعداد حي في الإنتاج. إذا أُوقفت العملية في منتصف الكتابة، سيكون الملف مبتوراً وستفشل الخدمة التي تقرأه عند البدء. نمط الملف المؤقت ثم إعادة التسمية المذكور أعلاه هو النهج الصحيح الذي تعتمده أدوات مثل systemd وnginx ومعظم مديري الحزم.
تشغيل الأوامر الخارجية مع subprocess
تستدعي سكريبتات DevOps باستمرار أدوات CLI: git، وkubectl، وterraform، وaws، وdocker. وحدة subprocess في Python هي الطريقة الصحيحة لذلك. الأساليب القديمة مثل os.system() ووحدة commands مُهملة؛ لا تستخدمها أبداً.
نقطتا الدخول الرئيسيتان هما subprocess.run() للأوامر الفردية وsubprocess.Popen() للعمليات المتدفقة أو التفاعلية. ابدأ دائماً بـrun().
import subprocess
# --- الأسلوب الآمن: مرِّر الأوامر كقائمة، لا كنص shell ---
result = subprocess.run(
["git", "rev-parse", "--short", "HEAD"],
capture_output=True, # التقاط stdout وstderr بدل طباعتهما
text=True, # فك ترميز البايتات إلى str تلقائياً
check=True, # إطلاق CalledProcessError عند خروج غير صفري
)
commit_sha = result.stdout.strip()
print(f"Current commit: {commit_sha}")
# --- فحص رمز الخروج يدوياً بدل check=True ---
result = subprocess.run(
["systemctl", "is-active", "--quiet", "nginx"],
capture_output=True, text=True
)
if result.returncode == 0:
print("nginx is running")
else:
print("nginx is NOT running")
# --- تمرير متغيرات البيئة ---
import os
env = {**os.environ, "KUBECONFIG": "/home/deploy/.kube/config"}
result = subprocess.run(
["kubectl", "get", "nodes", "-o", "wide"],
capture_output=True, text=True, check=True, env=env
)
print(result.stdout)
# --- بث الناتج للأوامر طويلة الأمد ---
with subprocess.Popen(
["terraform", "apply", "-auto-approve"],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True
) as proc:
for line in proc.stdout:
print(line, end="") # ناتج فوري
proc.wait()
if proc.returncode != 0:
raise RuntimeError(f"terraform failed (exit {proc.returncode})")
كيف يُنشئ subprocess.run() عملية فرعية ويلتقط ناتجها ويُعيد كائن CompletedProcess إلى السكريبت.
حقن Shell: القاعدة الأمنية الحرجة
القاعدة الأهم عند استخدام subprocess في سكريبتات DevOps: لا تبنِ نصاً من مدخلات غير موثوقة وتمرره مع shell=True. يُمرِّر خيار shell=True النص الخاص بك إلى /bin/sh -c، مما يعني أن أي محرف خاص بالـ shell (;، |، $()، &&) في البيانات المُقدَّمة من المستخدم أو من البيئة يصبح ثغرة حقن.
# خطأ — خطر حقن shell إذا جاء branch_name من مصدر خارجي
branch_name = "main; rm -rf /"
result = subprocess.run(
f"git checkout {branch_name}",
shell=True, # لا تفعل هذا أبداً مع مدخلات خارجية
capture_output=True
)
# صحيح — صيغة القائمة؛ لا يُستدعى الـ shell أبداً
result = subprocess.run(
["git", "checkout", branch_name], # branch_name مجرد وسيط
capture_output=True, text=True, check=True
)
# استخدام مقبول لـ shell=True: فقط للتوجيهات المدمجة أو pipelines
# بسلاسل نصية مُكوَّدة بالكامل (لا بيانات مستخدم في أي مكان)
result = subprocess.run(
"df -h | grep /dev/sda",
shell=True, capture_output=True, text=True
)
على نطاق شركات التقنية الكبرى، تُشغَّل سكريبتات الأتمتة في الغالب بواسطة أنظمة CI/CD أو خطافات ويب أو مدخلات المشغِّلين. حقن shell في سكريبت نشر يعمل بصلاحيات root يعني اختراقاً كاملاً للخادم. تعامل مع صيغة القائمة في subprocess.run() باعتبارها الأصل، واعتبر shell=True إشارة تحذير في مراجعات الكود ما لم يكن النص ثابتاً مُكوَّداً.
التعامل مع الأخطاء والمهلة الزمنية
يجب أن تتعامل سكريبتات ops في الإنتاج مع الفشل بشكل لائق. عيِّن دائماً timeout حتى لا يُعطِّل أمر خارجي معلَّق خط الأنابيب الخاص بك إلى أجل غير مسمى. التقط subprocess.CalledProcessError لتسجيل التشخيصات والقرار بشأن إعادة المحاولة أو التنبيه أو الإيقاف.
import subprocess, logging
log = logging.getLogger(__name__)
def run_kubectl(args: list[str], timeout: int = 30) -> str:
"""تشغيل kubectl بأمان؛ إرجاع stdout؛ رفع استثناء عند الفشل."""
cmd = ["kubectl"] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True,
timeout=timeout,
)
return result.stdout
except subprocess.CalledProcessError as exc:
log.error(
"kubectl failed (exit %d): %s",
exc.returncode,
exc.stderr.strip(),
)
raise
except subprocess.TimeoutExpired:
log.error("kubectl timed out after %ds: %s", timeout, cmd)
raise
# الاستخدام
nodes = run_kubectl(["get", "nodes", "-o", "name"])
print(nodes)
ملخص
استخدم pathlib.Path لجميع عمليات نظام الملفات — فهو محمول وقابل للقراءة ويتجنب أخطاء دمج النصوص.
اكتب ملفات الإعداد بشكل ذري عبر الملف المؤقت ثم إعادة التسمية لحماية الخدمات الجارية.
استخدم subprocess.run() مع قائمة من الوسائط، وcapture_output=True، وtext=True، وcheck=True، وtimeout.
لا تُمرِّر أبداً بيانات مستخدم عبر shell=True.
التقط CalledProcessError وTimeoutExpired؛ سجِّل stderr عند كل فشل.
نستخدم ملفات تعريف الارتباط لتشغيل هذا الموقع وتحليل الزيارات وعرض إعلانات مخصّصة. يمكنك قبول كل ملفات تعريف الارتباط أو رفض غير الأساسية منها.
سياسة الخصوصية