كل مهمة أتمتة DevOps جادة تتحدث في نهاية المطاف إلى cloud API. يمكنك صياغة طلبات HTTP خام باستخدام مكتبة requests (التي تناولناها في الدرس 4)، لكن مزودي السحابة يشحنون SDKs رسمية تتعامل مع المصادقة وتوقيع الطلبات وإعادة المحاولات التلقائية ونقاط النهاية الإقليمية وترقيم الصفحات وانتظار الموارد خارج الصندوق. بالنسبة لـ AWS، ذلك الـ SDK هو boto3. فهم نموذجه الداخلي — الجلسات والعملاء والموارد وأدوات الترقيم وأدوات الانتظار — ليس اختيارياً إذا كنت تخطط لتشغيل بنية AWS التحتية على نطاق واسع. هذا الدرس يعلم هذا النموذج بالطريقة التي يشرحها مهندس SRE أول لموظف جديد في أسبوعه الأول.
التسلسل الهرمي لكائنات boto3
يعرض boto3 ثلاث طبقات تجريد مميزة. فهم أي منها يُستخدم — ومتى — هو أول درس تتخطاه معظم الدروس التعليمية.
Session (الجلسة): حاوية إعداد. تحمل بيانات الاعتماد والمنطقة والملف الشخصي التي ترثها جميع استدعاءات API اللاحقة. كل تفاعل مع boto3 يتدفق عبر جلسة، سواء أنشأتها صراحةً أم لا.
Client (العميل): غلاف رفيع منخفض المستوى يرتبط 1:1 بعمليات AWS service API. كل إجراء API في وثائق AWS يتوافق بالضبط مع طريقة عميل واحدة. الاستجابات هي قواميس Python عادية. هذه هي الطبقة التي تستخدمها عند الحاجة إلى تحكم دقيق، أو حين لا يغطي التجريد Resource خدمة ما، أو حين تحتاج لتمرير معاملات طلبات خام.
Resource (المورد): غلاف أعلى مستوى وموجه للكائنات فوق العملاء. كائن Bucket في S3 يملك طرقاً كـ .upload_file() و.objects.all() بدلاً من استدعاءات API خام. الموارد مريحة لكنها تغطي فقط مجموعة فرعية من خدمات AWS (S3 وEC2 وDynamoDB وIAM وSQS وSNS). لأي شيء آخر، استخدم العميل مباشرةً.
الفكرة الأساسية: في شركات التقنية الكبرى، معظم أدوات العمليات الداخلية تستخدم العملاء بدلاً من الموارد. الموارد تجرد تفاصيل قد تحتاجها أحياناً (مثل بيانات الاستجابة الأولية لتشخيص الأخطاء أو لتمرير معاملات VersionId لعمليات S3). ابنِ على العملاء؛ استخدم الموارد فقط حين تُبسّط كودك فعلاً.
الجلسات: الطريقة الصحيحة للتعامل مع بيانات الاعتماد
سلسلة بيانات الاعتماد الافتراضية لـ boto3 تقرأ من متغيرات البيئة (AWS_ACCESS_KEY_ID وAWS_SECRET_ACCESS_KEY)، ثم ~/.aws/credentials، ثم ملفات تعريف IAM (على EC2/ECS/Lambda). لمعظم السكريبتات على بنية تحتية مُعدَّة بشكل صحيح، تعمل هذه السلسلة تلقائياً — لا تُرمّز شيئاً.
أنشئ Session صريحة حين يحتاج سكريبتك للعمل مع حسابات متعددة (أتمتة بين الحسابات)، أو مناطق متعددة، أو حين تريد افتراض دور عبر STS. تمرير كائن الجلسة عبر كودك بدلاً من استدعاء boto3.client() على مستوى الوحدة يجعل بيانات الاعتماد مرئية وقابلة للاختبار والمحاكاة.
import boto3
from botocore.config import Config
# --- جلسة صريحة: موصى بها لجميع السكريبتات غير التافهة ---
# جلسة قائمة على ملف تعريف (تقرأ من قسم [prod] في ~/.aws/credentials)
session = boto3.Session(profile_name="prod", region_name="us-east-1")
# أو: افتراض دور عبر STS (نمط أتمتة بين الحسابات)
sts = boto3.client("sts")
assumed = sts.assume_role(
RoleArn="arn:aws:iam::123456789012:role/DeployAutomation",
RoleSessionName="ops-script",
DurationSeconds=3600,
)
creds = assumed["Credentials"]
session = boto3.Session(
aws_access_key_id=creds["AccessKeyId"],
aws_secret_access_key=creds["SecretAccessKey"],
aws_session_token=creds["SessionToken"],
region_name="us-east-1",
)
# أنشئ العملاء من الجلسة لا من وحدة boto3 مباشرة
# botocore.config.Config يتحكم في سلوك إعادة المحاولة والمهلة
config = Config(
retries={"max_attempts": 5, "mode": "adaptive"},
connect_timeout=5,
read_timeout=30,
)
ec2 = session.client("ec2", config=config)
s3 = session.client("s3", config=config)
ممارسة احترافية: استخدم mode="adaptive" في إعداد إعادة المحاولة بدلاً من "standard". الوضع التكيفي يُطبّق تراجعاً أسياً مع اهتزاز ويُبطئ العميل حين تعيد AWS ThrottlingException أو RequestLimitExceeded. الوضع القياسي يعيد المحاولة فوراً مما يُفاقم التقييد تحت الحمل — عكس ما تريده تماماً أثناء نشر جماعي أو سكريبت استجابة للحوادث.
أدوات الترقيم: لا تفترض أبداً أن هناك صفحة استجابة واحدة
أحد أكثر الأخطاء شيوعاً في سكريبتات العمليات هو استدعاء list API وتفويت النتائج بصمت. AWS list APIs مُرقَّمة: استدعاء واحد يعيد بحد أقصى بضع مئات من العناصر وNextToken (أو Marker أو NextPageToken — يختلف حسب الخدمة). إن لم تتبع الرمز المميز، ترى الصفحة الأولى فقط. على حساب صغير يعمل هذا بالصدفة. على حساب يحتوي 10,000 كائن S3 أو 500 نسخة EC2، يفوّت معظمها بصمت.
أدوات الترقيم في boto3 تُلغي هذه المشكلة تماماً. أداة الترقيم هي كائن boto3 يعرف حقل الرمز المميز الذي يجب اتباعه لـ API معين ويُصدر تلقائياً طلبات متتالية حتى تنتهي النتائج. الكود المُستدعي يرى مُكرِّراً واحداً.
import boto3
session = boto3.Session(region_name="us-east-1")
ec2 = session.client("ec2")
# --- خاطئ: قد يفوّت النسخ بصمت إذا كان هناك أكثر من 1000 ---
# response = ec2.describe_instances()
# reservations = response["Reservations"]
# --- صحيح: استخدم أداة ترقيم ---
paginator = ec2.get_paginator("describe_instances")
# page_iterator يُنتج قاموس استجابة واحداً لكل صفحة
page_iterator = paginator.paginate(
Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
)
instances = []
for page in page_iterator:
for reservation in page["Reservations"]:
for instance in reservation["Instances"]:
instances.append({
"id": instance["InstanceId"],
"type": instance["InstanceType"],
"az": instance["Placement"]["AvailabilityZone"],
})
print(f"Found {len(instances)} running instances")
# --- أداة ترقيم مع تصفية النتائج (من جانب الخادم، يُقلل استدعاءات API) ---
s3 = session.client("s3")
s3_paginator = s3.get_paginator("list_objects_v2")
# PaginationConfig يُحدد الصفحات أو إجمالي العناصر المُجلَبة
pages = s3_paginator.paginate(
Bucket="my-data-bucket",
Prefix="logs/2025/",
PaginationConfig={"PageSize": 1000},
)
# .search() يُطبّق تعبير JMESPath عبر جميع الصفحات
large_objects = list(pages.search("Contents[?Size > `10485760`]"))
print(f"Objects >10 MB: {len(large_objects)}")
مصيدة إنتاجية: لا تستدعِ أبداً list_objects_v2 أو describe_instances أو list_users أو أي list API آخر بدون أداة ترقيم في كود الإنتاج. حجم الصفحة الافتراضي لمعظم الخدمات هو 100-1000 عنصر. حساب يحتوي 1,001 مستخدم IAM سيُبلّغ بصمت عن 1,000 فقط في سكريبت امتثال — ولن يلاحظ أحد حتى يفشل مراجعة. اجعل أدوات الترقيم الافتراضية لا الاستثناء.
أدوات الانتظار: الحجب حتى تصبح AWS جاهزة
تستغرق موارد السحابة وقتاً للوصول إلى حالتها المطلوبة. نسخة EC2 تنتقل عبر pending → running. لقطة RDS تنتقل من creating → available. مكدس CloudFormation ينتقل من CREATE_IN_PROGRESS → CREATE_COMPLETE. يجب على سكريبت الأتمتة الانتظار لهذه الانتقالات قبل اتخاذ الإجراء التالي — لكن الاستطلاع يدوياً بحلقات time.sleep() هش ويستهلك حصة API وينتج كوداً قبيحاً.
أدوات انتظار boto3 تُلخّص منطق الاستطلاع الصحيح لكل انتقال حالة مورد: الـ API الصحيح للاستدعاء، الحقل الصحيح للفحص، فترة الاستطلاع الصحيحة (عادة 15 ثانية)، والعدد الأقصى الصحيح من المحاولات قبل الاستسلام. تحصل على استدعاء حجب نظيف إما يعود حين يصبح المورد جاهزاً أو يرفع WaiterError عند انتهاء المهلة.
import boto3
from botocore.exceptions import WaiterError
session = boto3.Session(region_name="us-east-1")
ec2 = session.client("ec2")
# --- تشغيل نسخة والانتظار حتى تكون في حالة التشغيل ---
response = ec2.run_instances(
ImageId="ami-0c02fb55956c7d316", # Amazon Linux 2023، us-east-1
InstanceType="t3.micro",
MinCount=1,
MaxCount=1,
TagSpecifications=[{
"ResourceType": "instance",
"Tags": [{"Key": "Name", "Value": "ops-worker"}],
}],
)
instance_id = response["Instances"][0]["InstanceId"]
print(f"Launched {instance_id}, waiting for running state...")
# ec2.get_waiter("instance_running") يستطلع describe_instances كل 15 ثانية
# الافتراضي: حتى 40 محاولة = 10 دقائق انتظار كحد أقصى
waiter = ec2.get_waiter("instance_running")
try:
waiter.wait(
InstanceIds=[instance_id],
WaiterConfig={"Delay": 15, "MaxAttempts": 40},
)
print(f"{instance_id} is now running")
except WaiterError as exc:
print(f"Waiter timed out or failed: {exc}")
raise SystemExit(1)
# --- إعداد انتظار مخصص للموارد سريعة الانتقال ---
# لقطة RDS: فحص كل 30 ثانية، استسلام بعد 60 دقيقة
rds = session.client("rds")
rds_waiter = rds.get_waiter("db_snapshot_available")
rds_waiter.wait(
DBSnapshotIdentifier="my-db-snap-2025-06-11",
WaiterConfig={"Delay": 30, "MaxAttempts": 120},
)
print("RDS snapshot is available")
هذا سكريبت عمليات واقعي يوقف نسخ EC2 المتوقفة منذ أكثر من 30 يوماً في منطقة معينة. يُوضّح كل مفهوم من هذا الدرس: جلسة صريحة مع إعداد إعادة المحاولة، أداة ترقيم لعداد جميع النسخ، فلتر لمطابقة المتوقفة، وفحص IAM مسبق قبل أي إجراء هدّام.
"""
decommission_stopped.py — ينهي نسخ EC2 المتوقفة لأكثر من 30 يوماً.
الاستخدام:
python decommission_stopped.py --region us-east-1 --dry-run
python decommission_stopped.py --region us-east-1
"""
import argparse
import logging
from datetime import datetime, timezone, timedelta
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)
CUTOFF_DAYS = 30
def get_session(region: str) -> boto3.Session:
return boto3.Session(region_name=region)
def build_client(session: boto3.Session, service: str):
cfg = Config(
retries={"max_attempts": 5, "mode": "adaptive"},
connect_timeout=5,
read_timeout=30,
)
return session.client(service, config=cfg)
def stopped_instances_older_than(ec2, days: int) -> list[dict]:
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
paginator = ec2.get_paginator("describe_instances")
pages = paginator.paginate(
Filters=[{"Name": "instance-state-name", "Values": ["stopped"]}]
)
old = []
for page in pages:
for r in page["Reservations"]:
for inst in r["Instances"]:
state_reason = inst.get("StateTransitionReason", "")
try:
ts_str = state_reason.split("(")[1].rstrip(")")
ts = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S GMT").replace(
tzinfo=timezone.utc
)
except (IndexError, ValueError):
ts = inst["LaunchTime"]
if ts < cutoff:
old.append({"id": inst["InstanceId"], "stopped_at": ts})
return old
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--region", required=True)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
session = get_session(args.region)
ec2 = build_client(session, "ec2")
targets = stopped_instances_older_than(ec2, CUTOFF_DAYS)
if not targets:
log.info("No instances qualify for decommission")
return
ids = [t["id"] for t in targets]
log.info("Candidate instances: %s", ids)
if args.dry_run:
log.info("[DRY RUN] Would terminate %d instance(s): %s", len(ids), ids)
return
try:
ec2.terminate_instances(InstanceIds=ids)
log.info("Termination issued for %s", ids)
except ClientError as exc:
log.error("terminate_instances failed: %s", exc)
raise SystemExit(1)
if __name__ == "__main__":
main()
مصيدة إنتاجية: أضف دائماً علامة --dry-run لأي أتمتة هدّامة. لكن لاحظ أن لـ boto3 معنيين مختلفين لـ "dry run": بعض EC2 APIs تقبل معامل DryRun=True تُقيّمه AWS من جانب الخادم (تعيد نجاح DryRunOperation أو خطأ UnauthorizedOperation للتحقق من أذونات IAM). علامة --dry-run في CLI الخاصة بك كما هو موضح أعلاه تتخطى استدعاء API الهدّام تماماً — وهو أأمن. استخدم كلا الطبقتين معاً: علامتك للاختبار الشامل، وـDryRun=True في AWS للتحقق المسبق من أذونات IAM.
معالجة الأخطاء لاستدعاءات SDK
يُظهر boto3 فئتين من الاستثناءات. botocore.exceptions.ClientError يُغلّف أخطاء AWS service (HTTP 4xx/5xx): أذونات خاطئة، مورد غير موجود، معامل مُشوَّه. botocore.exceptions.BotoCoreError يُغلّف أخطاء مستوى الاتصال: انتهاء المهلة، أخطاء SSL، نقطة نهاية غير قابلة للوصول. في سكريبتات الإنتاج، اصطد كليهما وسجّل كود الخطأ الكامل لا الرسالة فقط.
from botocore.exceptions import ClientError, BotoCoreError
def safe_describe_instance(ec2, instance_id: str) -> dict | None:
try:
resp = ec2.describe_instances(InstanceIds=[instance_id])
return resp["Reservations"][0]["Instances"][0]
except ClientError as exc:
code = exc.response["Error"]["Code"]
if code == "InvalidInstanceID.NotFound":
log.warning("Instance %s not found", instance_id)
return None
# UnauthorizedOperation وInvalidParameterValue وما إلى ذلك — أعد الرفع
log.error("ClientError [%s] for %s: %s", code, instance_id, exc)
raise
except BotoCoreError as exc:
# فشل على مستوى الشبكة — إعداد إعادة المحاولة عالج المحاولات مسبقاً
log.error("Connection error: %s", exc)
raise
الخلاصة الرئيسية: النمط الذي بنيته في هذا الدرس — جلسة صريحة مع إعداد إعادة المحاولة، أداة ترقيم لكل استدعاء قائمة، أداة انتظار لكل انتقال حالة، معالجة منظّمة للأخطاء — هو بالضبط النمط المستخدم في أدوات AWS الداخلية وفي أطر عمل العمليات مفتوحة المصدر مثل awscli وchalice وcloud-custodian. طبّقه على أي SDK سحابي (GCP يستخدم google-cloud-* بمفاهيم مشابهة؛ Azure SDK لـ Python يعكس نفس نموذج العميل/الترقيم) وستكتب أتمتة تصمد في الإنتاج على نطاق واسع.
نستخدم ملفات تعريف الارتباط لتشغيل هذا الموقع وتحليل الزيارات وعرض إعلانات مخصّصة. يمكنك قبول كل ملفات تعريف الارتباط أو رفض غير الأساسية منها.
سياسة الخصوصية