کنترل خطا (Exceptions) در پایتون
جلسه دوازدهم - کنترل خطا (Exceptions) در پایتون
مفهوم، انواع مهم استثناها و ساختارهای try/except/else/finally و نحوه استفاده از raise
معرفی
در برنامهنویسی ممکن است اتفاقات غیرمنتظرهای رخ دهند، مثل تقسیم بر صفر یا دسترسی به ایندکس ناموجود. کنترل خطا یعنی برنامه را طوری بنویسیم که در مواجهه با این خطاها متوقف نشود و بتوانیم پیغام مناسب بدهیم یا رفتار جایگزین انجام دهیم.
۳۰ خطای مهم پایتون
در ادامه فهرستی از ۳۰ Exception مهم و توضیح کوتاه هرکدام آورده شده است:
- 1. ArithmeticError
- خطای عمومی محاسباتی (پایه برای دیگر خطاهای محاسباتی).
- 2. ZeroDivisionError
- وقتی عددی را بر صفر تقسیم کنیم.
- 3. OverflowError
- وقتی نتیجه محاسبات عددی از حد مجاز بزرگتر شود.
- 4. FloatingPointError
- خطاهای مربوط به عملیات اعداد اعشاری.
- 5. AssertionError
- وقتی شرط assert درست نباشد.
- 6. AttributeError
- دسترسی به متد یا ویژگیای که وجود ندارد.
- 7. EOFError
- وقتی ورودی input() به انتهای فایل برسد.
- 8. ImportError
- وقتی ماژولی پیدا نشود یا وارد نشود.
- 9. ModuleNotFoundError
- وقتی ماژول مشخص وجود نداشته باشد (زیرمجموعه ImportError).
- 10. IndexError
- وقتی ایندکس لیست یا تاپل خارج از محدوده باشد.
- 11. KeyError
- وقتی کلید دیکشنری وجود ندارد.
- 12. NameError
- وقتی متغیری تعریف نشده استفاده شود.
- 13. UnboundLocalError
- وقتی متغیر محلی قبل از مقداردهی استفاده شود.
- 14. TypeError
- وقتی نوع داده مناسب برای عملیات نباشد.
- 15. ValueError
- وقتی مقدار اشتباه ولی نوع درست داده شود (مثلاً int("abc")).
- 16. FileNotFoundError
- وقتی فایلی که باز میکنیم وجود ندارد.
- 17. IOError / OSError
- خطاهای ورودی/خروجی مثل باز کردن فایل یا نوشتن روی آن.
- 18. PermissionError
- وقتی دسترسی لازم برای فایل یا عملیات نداریم.
- 19. StopIteration
- وقتی Iterator به پایان میرسد و مقدار جدیدی نیست.
- 20. StopAsyncIteration
- نسخه Async برای Iteratorهای غیرهمزمان.
- 21. MemoryError
- وقتی حافظه کافی برای عملیات وجود ندارد.
- 22. RecursionError
- وقتی بازگشتی بیش از حد عمق داشته باشد.
- 23. NotImplementedError
- وقتی متدی هنوز در کلاس فرزند پیادهسازی نشده.
- 24. SyntaxError
- خطای نحوی (اشتباه در ساختار کد).
- 25. IndentationError
- خطای تورفتگی یا فاصله نادرست.
- 26. TabError
- استفاده همزمان از تب و فاصله در تورفتگی.
- 27. SystemError
- خطای داخلی پایتون (معمولاً ناشی از خود مفسر).
- 28. SystemExit
- وقتی برنامه با sys.exit() بسته میشود.
- 29. KeyboardInterrupt
- وقتی کاربر با Ctrl+C اجرای برنامه را متوقف میکند.
- 30. UnicodeError
- خطاهای مربوط به تبدیل یا کدگذاری یونیکد.
ساختار پایه try/except
بلوک try شامل کدی است که ممکن است باعث خطا شود و except مشخص میکند در صورت رخ دادن خطا چه کاری انجام شود.
try:
x = int(input("یک عدد وارد کنید: "))
result = 10 / x
except ZeroDivisionError:
print("خطا: نمیتوان عدد را بر صفر تقسیم کرد")
except ValueError:
print("خطا: لطفاً عدد معتبر وارد کنید")
میتوان چند except برای انواع مختلف خطاها داشت و برای هرکدام رفتار متفاوتی تعریف کرد.
اضافه کردن else
بلوک else فقط زمانی اجرا میشود که هیچ خطایی در بخش try رخ ندهد.
try:
x = int(input("یک عدد وارد کنید: "))
result = 10 / x
except ZeroDivisionError:
print("خطا: نمیتوان عدد را بر صفر تقسیم کرد")
except ValueError:
print("خطا: لطفاً عدد معتبر وارد کنید")
else:
print("کد بدون خطا اجرا شد، نتیجه:", result)
استفاده از finally
بلوک finally همیشه اجرا میشود، چه خطا رخ دهد چه نه. معمولاً برای آزادسازی منابع استفاده
میشود.
try:
f = open('data.txt')
data = f.read()
except FileNotFoundError:
print('فایل یافت نشد')
finally:
try:
f.close()
except:
pass
تعریف raise
با raise میتوانید عمداً یک استثنا ایجاد کنید تا هنگام مواجهه با شرایط نامعتبر، اجرای
برنامه کنترل شود.
def set_age(age):
if age < 0:
raise ValueError("age must be positive")
print(f"Age set to {age}")
توجه: raise بدون پارامتر میتواند خطای فعلی را دوباره بالا بفرستد (re-raise).
except Exception
این فرم همهٔ استثناهای مشتق از Exception را میگیرد و برای جلوگیری از کرش برنامه در زمان
خطاهای ناشناخته مفید است، اما نباید جایگزین مدیریت دقیق خطاها شود.
try:
x = int(input("یک عدد وارد کن: "))
print(10 / x)
except ZeroDivisionError:
print("نمیتوان بر صفر تقسیم کرد!")
except ValueError:
print("فقط عدد وارد کن!")
except Exception as e:
print("یه خطای غیرمنتظره:", e)
نکات مهم و Best Practices
- قبل از هر چیز، خطاها را به صورت مشخص هندل کنید (مثلاً ZeroDivisionError، FileNotFoundError).
- از
except Exceptionفقط به عنوان آخرین دفاع استفاده کنید. - در بلوکهای
exceptاطلاعات کافی برای دیباگ لاگ کنید (traceback یا پیام). اما از نمایش پیامهای داخلی حساس به کاربر خودداری کنید. - از
finallyبرای آزادسازی منابع استفاده کنید یا از context manager (with) بهره ببرید. - برای اعتبارسنجی ورودیها از
raiseاستفاده کنید تا خطاها زود و مشخص بروز کنند.
مثالهای کاربردی
مثال: خواندن امن فایل
try:
with open('config.json') as f:
cfg = f.read()
except FileNotFoundError:
cfg = '{}'
print('از تنظیمات پیشفرض استفاده شد')
مثال: هندل کردن تقسیم و گزارش لاگ
import logging
logging.basicConfig(level=logging.ERROR)
def safe_div(a, b):
try:
return a / b
except ZeroDivisionError as e:
logging.error('division by zero: %s', e)
return None
💡 تمرینها (20 سوال کاربردی از پایه تا پیشرفته)
سوال 1: تبدیل امن به عدد صحیح
تابعی بنویسید که یک رشته را به عدد صحیح تبدیل کند. اگر امکان تبدیل نبود، None برگرداند.
# ورودی: '123' -> خروجی: 123
# ورودی: 'abc' -> خروجی: None
def safe_int(s):
try:
return int(s)
except ValueError:
return None
print(safe_int('123')) # 123
print(safe_int('abc')) # None
سوال 2: تقسیم امن
تابعی بنویسید که دو عدد دریافت کند و حاصل تقسیم را برگرداند. اگر تقسیم بر صفر بود، پیغام "تقسیم بر صفر امکانپذیر نیست" چاپ کرده و None برگرداند.
# ورودی: 10, 2 -> خروجی: 5.0
# ورودی: 10, 0 -> خروجی: None
def safe_divide(a, b):
try:
return a / b
except ZeroDivisionError:
print("تقسیم بر صفر امکانپذیر نیست")
return None
print(safe_divide(10, 2)) # 5.0
print(safe_divide(10, 0)) # None
سوال 3: دسترسی امن به لیست
تابعی بنویسید که لیست و ایندکس دریافت کند و مقدار آن ایندکس را برگرداند. اگر ایندکس خارج از محدوده بود، "ایندکس نامعتبر" چاپ کند و None برگرداند.
# ورودی: [1,2,3], 1 -> خروجی: 2
# ورودی: [1,2,3], 10 -> خروجی: None
def safe_get(lst, index):
try:
return lst[index]
except IndexError:
print("ایندکس نامعتبر")
return None
print(safe_get([1,2,3], 1)) # 2
print(safe_get([1,2,3], 10)) # None
سوال 4: دسترسی امن به دیکشنری
تابعی بنویسید که دیکشنری و کلید دریافت کند و مقدار آن کلید را برگرداند. اگر کلید وجود نداشت، مقدار پیشفرض "NotFound" برگرداند.
# ورودی: {'a': 1}, 'a' -> خروجی: 1
# ورودی: {'a': 1}, 'b' -> خروجی: "NotFound"
def safe_dict_get(d, key):
try:
return d[key]
except KeyError:
return "NotFound"
data = {'name': 'Ali', 'age': 25}
print(safe_dict_get(data, 'name')) # Ali
print(safe_dict_get(data, 'address')) # NotFound
سوال 5: خواندن امن فایل
تابعی بنویسید که نام فایل را دریافت کند و محتوای آن را برگرداند. اگر فایل وجود نداشت یا خطای دسترسی رخ داد، پیغام مناسب چاپ کرده و رشته خالی برگرداند.
# ورودی: 'data.txt' (موجود) -> محتوای فایل
# ورودی: 'missing.txt' -> ""
def read_file_safe(filename):
try:
with open(filename, 'r', encoding='utf-8') as f:
return f.read()
except FileNotFoundError:
print(f"فایل {filename} یافت نشد")
return ""
except PermissionError:
print(f"دسترسی به فایل {filename} وجود ندارد")
return ""
except Exception as e:
print(f"خطای غیرمنتظره: {e}")
return ""
سوال 6: نوشتن امن در فایل
تابعی بنویسید که نام فایل و متن را دریافت کند و متن را در فایل بنویسد. در صورت بروز خطا، پیغام مناسب چاپ کرده و False برگرداند. در صورت موفقیت True برگرداند.
# ورودی: 'output.txt', 'Hello' -> True
# ورودی: '/root/file.txt', 'test' -> False (خطای دسترسی)
def write_file_safe(filename, content):
try:
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
return True
except PermissionError:
print("دسترسی برای نوشتن در فایل وجود ندارد")
return False
except Exception as e:
print(f"خطا در نوشتن فایل: {e}")
return False
write_file_safe('test.txt', 'Hello World')
سوال 7: محاسبه میانگین امن
تابعی بنویسید که لیستی از اعداد دریافت کند و میانگین آنها را برگرداند. اگر لیست خالی بود، پیغام "لیست خالی است" چاپ کرده و 0 برگرداند. مقادیر غیرعددی را نادیده بگیرد.
# ورودی: [1, 2, 3] -> خروجی: 2.0
# ورودی: [] -> خروجی: 0
# ورودی: [1, 'a', 3] -> خروجی: 2.0
def safe_average(numbers):
valid_numbers = []
for num in numbers:
try:
valid_numbers.append(float(num))
except (ValueError, TypeError):
continue
try:
return sum(valid_numbers) / len(valid_numbers)
except ZeroDivisionError:
print("لیست خالی است")
return 0
print(safe_average([1, 2, 3])) # 2.0
print(safe_average([])) # 0
print(safe_average([1, 'a', 3])) # 2.0
سوال 8: اعتبارسنجی سن
تابعی بنویسید که سن را دریافت کند. اگر سن منفی یا بیشتر از 150 بود، ValueError با پیغام مناسب raise کند. در غیر این صورت پیغام "سن معتبر است" برگرداند.
# ورودی: 25 -> "سن معتبر است"
# ورودی: -5 -> ValueError
# ورودی: 200 -> ValueError
def validate_age(age):
if age < 0:
raise ValueError("سن نمیتواند منفی باشد")
if age > 150:
raise ValueError("سن نمیتواند بیشتر از 150 باشد")
return "سن معتبر است"
try:
print(validate_age(25)) # سن معتبر است
print(validate_age(-5)) # ValueError
except ValueError as e:
print(f"خطا: {e}")
سوال 9: ماشینحساب امن
تابعی بنویسید که دو عدد و عملگر (+, -, *, /) دریافت کند و نتیجه را برگرداند. تمام خطاهای ممکن (تقسیم بر صفر، عملگر نامعتبر، ورودی غیرعددی) را مدیریت کنید.
# ورودی: 10, 2, '+' -> 12
# ورودی: 10, 0, '/' -> None
# ورودی: 10, 2, '%' -> None
def calculator(a, b, op):
try:
a, b = float(a), float(b)
if op == '+':
return a + b
elif op == '-':
return a - b
elif op == '*':
return a * b
elif op == '/':
return a / b
else:
raise ValueError("عملگر نامعتبر")
except ValueError as e:
print(f"خطا: {e}")
return None
except ZeroDivisionError:
print("خطا: تقسیم بر صفر")
return None
except Exception as e:
print(f"خطای غیرمنتظره: {e}")
return None
print(calculator(10, 2, '+')) # 12.0
print(calculator(10, 0, '/')) # None
سوال 10: پردازش لیست JSON
تابعی بنویسید که رشته JSON دریافت کند و آن را به لیست تبدیل کرده و تعداد عناصر را برگرداند. اگر JSON نامعتبر بود، پیغام خطا چاپ کرده و 0 برگرداند.
# ورودی: '[1,2,3]' -> 3
# ورودی: '{invalid}' -> 0
import json
def count_json_items(json_str):
try:
data = json.loads(json_str)
if isinstance(data, list):
return len(data)
else:
print("JSON باید لیست باشد")
return 0
except json.JSONDecodeError:
print("JSON نامعتبر است")
return 0
except Exception as e:
print(f"خطا: {e}")
return 0
print(count_json_items('[1,2,3]')) # 3
print(count_json_items('{invalid}')) # 0
سوال 11: اتصال به دیتابیس امن
تابعی بنویسید که به دیتابیس SQLite متصل شود، یک query ساده اجرا کند و نتیجه را برگرداند. از try/except/finally برای مدیریت اتصال استفاده کنید.
# تابع باید اتصال را در هر صورت ببندد
import sqlite3
def fetch_data_safe(db_name, query):
conn = None
try:
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
cursor.execute(query)
results = cursor.fetchall()
return results
except sqlite3.Error as e:
print(f"خطای دیتابیس: {e}")
return []
except Exception as e:
print(f"خطای غیرمنتظره: {e}")
return []
finally:
if conn:
conn.close()
print("اتصال بسته شد")
# مثال استفاده:
# data = fetch_data_safe('test.db', 'SELECT * FROM users')
سوال 12: پیادهسازی Retry Mechanism
تابعی بنویسید که یک تابع دیگر را دریافت کند و آن را حداکثر 3 بار تلاش کند اجرا کند. اگر در هر 3 بار خطا رخ داد، پیغام "عملیات ناموفق" برگرداند.
# تابع باید در صورت خطا، دوباره تلاش کند
def retry_function(func, *args, max_attempts=3):
for attempt in range(1, max_attempts + 1):
try:
result = func(*args)
print(f"موفق در تلاش {attempt}")
return result
except Exception as e:
print(f"تلاش {attempt} ناموفق: {e}")
if attempt == max_attempts:
print("عملیات ناموفق پس از 3 تلاش")
return None
# مثال:
def unstable_func():
import random
if random.random() < 0.7:
raise Exception("خطای تصادفی")
return "موفق"
retry_function(unstable_func)
سوال 13: سیستم لاگ خطاها
کلاسی بنویسید که خطاها را در یک فایل لاگ ذخیره کند. هر خطا باید شامل تاریخ، زمان، نوع خطا و پیام باشد. از try/except برای مدیریت خطاهای نوشتن در فایل استفاده کنید.
# کلاس ErrorLogger با متد log_error
from datetime import datetime
class ErrorLogger:
def __init__(self, log_file='errors.log'):
self.log_file = log_file
def log_error(self, error_type, message):
try:
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_entry = f"[{timestamp}] {error_type}: {message}\n"
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(log_entry)
print(f"خطا ثبت شد: {error_type}")
except PermissionError:
print("خطا: دسترسی به فایل لاگ وجود ندارد")
except Exception as e:
print(f"خطا در ثبت لاگ: {e}")
# استفاده:
logger = ErrorLogger()
try:
x = 10 / 0
except ZeroDivisionError as e:
logger.log_error('ZeroDivisionError', str(e))
سوال 14: Custom Exception برای بانک
کلاس Exception سفارشی به نام InsufficientFundsError بنویسید. سپس کلاس BankAccount بنویسید که متد withdraw داشته باشد و در صورت موجودی ناکافی این Exception را raise کند.
# موجودی: 1000، برداشت: 1500 -> InsufficientFundsError
class InsufficientFundsError(Exception):
def __init__(self, balance, amount):
self.balance = balance
self.amount = amount
super().__init__(f"موجودی ناکافی: دارایی={balance}، درخواست={amount}")
class BankAccount:
def __init__(self, balance=0):
self.balance = balance
def withdraw(self, amount):
if amount > self.balance:
raise InsufficientFundsError(self.balance, amount)
self.balance -= amount
return f"برداشت موفق. موجودی: {self.balance}"
# استفاده:
account = BankAccount(1000)
try:
print(account.withdraw(500)) # موفق
print(account.withdraw(1000)) # خطا
except InsufficientFundsError as e:
print(f"خطا: {e}")
سوال 15: مدیریت چندین فایل با Context Manager
تابعی بنویسید که محتوای چند فایل را بخواند، آنها را ترکیب کند و در فایل جدید بنویسد. تمام خطاهای ممکن (فایل ناموجود، دسترسی، کدگذاری) را مدیریت کنید.
# ورودی: ['file1.txt', 'file2.txt'], 'output.txt'
def merge_files(input_files, output_file):
combined_content = []
for filename in input_files:
try:
with open(filename, 'r', encoding='utf-8') as f:
content = f.read()
combined_content.append(f"=== {filename} ===\n{content}\n")
except FileNotFoundError:
print(f"فایل {filename} یافت نشد، نادیده گرفته شد")
except UnicodeDecodeError:
print(f"خطای کدگذاری در {filename}")
except Exception as e:
print(f"خطا در خواندن {filename}: {e}")
try:
with open(output_file, 'w', encoding='utf-8') as f:
f.write('\n'.join(combined_content))
print(f"فایلها با موفقیت در {output_file} ترکیب شدند")
return True
except Exception as e:
print(f"خطا در نوشتن فایل خروجی: {e}")
return False
# merge_files(['file1.txt', 'file2.txt'], 'merged.txt')
سوال 16: سیستم اعتبارسنجی فرم
کلاسی بنویسید که فرم ثبتنام را اعتبارسنجی کند. باید ایمیل، شماره تلفن و رمز عبور را بررسی کرده و برای هر خطا یک Exception مناسب raise کند.
# ایمیل نامعتبر -> InvalidEmailError
# رمز ضعیف -> WeakPasswordError
import re
class InvalidEmailError(Exception):
pass
class WeakPasswordError(Exception):
pass
class InvalidPhoneError(Exception):
pass
class FormValidator:
@staticmethod
def validate_email(email):
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
if not re.match(pattern, email):
raise InvalidEmailError("فرمت ایمیل نامعتبر است")
@staticmethod
def validate_password(password):
if len(password) < 8:
raise WeakPasswordError("رمز باید حداقل 8 کاراکتر باشد")
if not any(c.isdigit() for c in password):
raise WeakPasswordError("رمز باید شامل عدد باشد")
@staticmethod
def validate_phone(phone):
if not phone.isdigit() or len(phone) != 11:
raise InvalidPhoneError("شماره تلفن باید 11 رقم باشد")
def validate_form(self, email, password, phone):
try:
self.validate_email(email)
self.validate_password(password)
self.validate_phone(phone)
return "فرم معتبر است"
except (InvalidEmailError, WeakPasswordError, InvalidPhoneError) as e:
return f"خطای اعتبارسنجی: {e}"
# استفاده:
validator = FormValidator()
print(validator.validate_form('test@example.com', 'pass123', '09123456789'))
سوال 17: دانلود فایل با مدیریت خطا
تابعی بنویسید که URL یک فایل را دریافت کند و آن را دانلود کرده و ذخیره کند. خطاهای شبکه، timeout، و فایل ناموجود را مدیریت کنید (از کتابخانه requests استفاده کنید).
# تابع باید خطاهای مختلف شبکه را مدیریت کند
import requests
def download_file(url, filename):
try:
response = requests.get(url, timeout=10)
response.raise_for_status() # خطا برای کدهای 4xx/5xx
with open(filename, 'wb') as f:
f.write(response.content)
print(f"فایل با موفقیت دانلود شد: {filename}")
return True
except requests.exceptions.Timeout:
print("خطا: زمان اتصال به پایان رسید")
return False
except requests.exceptions.ConnectionError:
print("خطا: عدم اتصال به اینترنت")
return False
except requests.exceptions.HTTPError as e:
print(f"خطای HTTP: {e}")
return False
except PermissionError:
print("خطا: دسترسی برای ذخیره فایل وجود ندارد")
return False
except Exception as e:
print(f"خطای غیرمنتظره: {e}")
return False
# download_file('https://example.com/file.pdf', 'output.pdf')
سوال 18: سیستم کش با Timeout
کلاسی بنویسید که دادهها را با زمان انقضا ذخیره کند. اگر داده منقضی شده بود، Exception مناسبی raise کند. از خطاهای احتمالی در ذخیرهسازی محافظت کنید.
# کش با مدیریت زمان انقضا
import time
class CacheExpiredError(Exception):
pass
class TimedCache:
def __init__(self):
self.cache = {}
def set(self, key, value, ttl=60):
try:
expiry_time = time.time() + ttl
self.cache[key] = {'value': value, 'expiry': expiry_time}
print(f"کلید '{key}' با TTL {ttl} ثانیه ذخیره شد")
except Exception as e:
print(f"خطا در ذخیره کش: {e}")
def get(self, key):
try:
if key not in self.cache:
raise KeyError(f"کلید '{key}' در کش وجود ندارد")
item = self.cache[key]
if time.time() > item['expiry']:
del self.cache[key]
raise CacheExpiredError(f"کش '{key}' منقضی شده است")
return item['value']
except (KeyError, CacheExpiredError) as e:
print(f"خطا: {e}")
return None
# استفاده:
cache = TimedCache()
cache.set('user', {'name': 'Ali'}, ttl=5)
print(cache.get('user')) # {'name': 'Ali'}
time.sleep(6)
print(cache.get('user')) # None (منقضی شده)
سوال 19: تبدیل و پردازش CSV امن
تابعی بنویسید که فایل CSV بخواند، ستون مشخصی را فیلتر کند و نتیجه را در فایل JSON ذخیره کند. تمام خطاهای ممکن (فایل ناموجود، فرمت نادرست، ستون ناموجود) را مدیریت کنید.
# CSV → فیلتر → JSON با مدیریت کامل خطا
import csv
import json
def csv_to_json_filtered(csv_file, json_file, filter_column, filter_value):
try:
with open(csv_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
# بررسی وجود ستون
if filter_column not in reader.fieldnames:
raise ValueError(f"ستون '{filter_column}' در فایل CSV وجود ندارد")
# فیلتر کردن دادهها
filtered_data = [
row for row in reader
if row.get(filter_column) == filter_value
]
if not filtered_data:
print(f"هیچ دادهای با مقدار '{filter_value}' یافت نشد")
# ذخیره در JSON
with open(json_file, 'w', encoding='utf-8') as f:
json.dump(filtered_data, f, ensure_ascii=False, indent=2)
print(f"{len(filtered_data)} رکورد در {json_file} ذخیره شد")
return True
except FileNotFoundError:
print(f"فایل {csv_file} یافت نشد")
return False
except csv.Error as e:
print(f"خطای CSV: {e}")
return False
except ValueError as e:
print(f"خطا: {e}")
return False
except Exception as e:
print(f"خطای غیرمنتظره: {e}")
return False
# csv_to_json_filtered('users.csv', 'output.json', 'age', '25')
سوال 20: سیستم Transaction برای پایگاه داده
کلاسی بنویسید که چندین عملیات پایگاه داده را در یک Transaction اجرا کند. اگر هر عملیاتی با خطا مواجه شد، تمام تغییرات را Rollback کند. از try/except/else/finally استفاده کنید.
# Transaction با Rollback خودکار در صورت خطا
import sqlite3
class DatabaseTransaction:
def __init__(self, db_name):
self.db_name = db_name
self.conn = None
def execute_transaction(self, queries):
try:
self.conn = sqlite3.connect(self.db_name)
cursor = self.conn.cursor()
# اجرای کوئریها
for query in queries:
cursor.execute(query)
except sqlite3.Error as e:
print(f"خطای دیتابیس: {e}")
if self.conn:
self.conn.rollback()
print("تمام تغییرات Rollback شد")
return False
except Exception as e:
print(f"خطای غیرمنتظره: {e}")
if self.conn:
self.conn.rollback()
return False
else:
# اگر خطایی نبود، commit کن
self.conn.commit()
print("Transaction با موفقیت انجام شد")
return True
finally:
# همیشه اتصال را ببند
if self.conn:
self.conn.close()
print("اتصال دیتابیس بسته شد")
# استفاده:
# db = DatabaseTransaction('test.db')
# queries = [
# "INSERT INTO users (name, age) VALUES ('Ali', 25)",
# "INSERT INTO users (name, age) VALUES ('Sara', 30)"
# ]
# db.execute_transaction(queries)
دیدگاهتان را بنویسید