🐍 آموزش جامع Python — پست ۲۲۷
مدیریت نسخه‌های ساختار Database با Migration
در پروژه‌های واقعی، ساختار دیتابیس ممکن است به‌مرور تغییر کند؛ مثلاً ابتدا جدول کاربران فقط id و name داشته باشد، اما بعداً بخواهیم ستون email را هم اضافه کنیم.
به‌جای تغییر دستی دیتابیس در هر اجرا، می‌توانیم از روشی به نام Database Migration استفاده کنیم.
Migration یعنی ثبت و اجرای مرحله‌به‌مرحله تغییرات ساختار دیتابیس.
🔹 چرا Migration لازم است؟
فرض کنیم نسخه اول جدول چنین باشد:

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL
);

در نسخه بعدی می‌خواهیم ستون email را اضافه کنیم:

ALTER TABLE users
ADD COLUMN email TEXT;

اگر برنامه هر بار این دستور را بدون بررسی اجرا کند، اجرای دوباره آن باعث خطا می‌شود؛ چون ستون از قبل وجود دارد.
پس باید مشخص کنیم کدام تغییرات قبلاً اجرا شده‌اند.
🔹 ساخت جدول ثبت Migrationها
یک جدول برای نگهداری نسخه فعلی ساختار ایجاد می‌کنیم:

import sqlite3

connection = sqlite3.connect("app.db")

connection.execute("""
    CREATE TABLE IF NOT EXISTS schema_migrations (
        version INTEGER PRIMARY KEY,
        applied_at TEXT DEFAULT CURRENT_TIMESTAMP
    )
""")

connection.commit()
connection.close()

هر Migration یک شماره نسخه دارد.
🔹 اجرای Migration فقط یک بار
در مثال زیر، Migration شماره 1 ستون email را اضافه می‌کند:

import sqlite3

connection = sqlite3.connect("app.db")

connection.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL
    )
""")

connection.execute("""
    CREATE TABLE IF NOT EXISTS schema_migrations (
        version INTEGER PRIMARY KEY,
        applied_at TEXT DEFAULT CURRENT_TIMESTAMP
    )
""")

with connection:
    applied = connection.execute(
        "SELECT 1 FROM schema_migrations WHERE version = ?",
        (1,)
    ).fetchone()

    if applied is None:
        connection.execute(
            "ALTER TABLE users ADD COLUMN email TEXT"
        )

        connection.execute(
            "INSERT INTO schema_migrations (version) VALUES (?)",
            (1,)
        )

connection.close()

اگر Migration قبلاً اجرا شده باشد، دوباره اجرا نمی‌شود.
توجه: این مثال فرض می‌کند دیتابیس قبلی جدول users را بدون ستون email دارد و Migration شماره ۱ هنوز ثبت نشده است. برای یک سیستم Migration عمومی باید وضعیت واقعی ساختار نیز اعتبارسنجی شود.
🔹 ترتیب نسخه‌ها
در پروژه‌های بزرگ‌تر ممکن است چند تغییر داشته باشیم:

Migration 1 → ایجاد جدول users
Migration 2 → افزودن ستون email
Migration 3 → ایجاد Index برای email
Migration 4 → ایجاد جدول orders

هر نسخه باید به‌ترتیب اجرا و ثبت شود.
⚠️ نکات مهم
Migrationها را بعد از اجرا بی‌دلیل تغییر نده؛ ممکن است دیتابیس کاربران با نسخه قبلی ساخته شده باشد.
قبل از تغییرات مهم، Backup تهیه کن.
همه دستورهای SQLite الزاماً در یک تراکنش قابل استفاده نیستند؛ رفتار تراکنشی DDL را برای تغییرات موردنظر بررسی کن.
ثبت نسخه به‌تنهایی کافی نیست؛ خطاها و تفاوت ساختار واقعی دیتابیس نیز باید مدیریت شوند.
🧠 خلاصه

بررسی نسخه فعلی
       ↓
اجرای Migration جدید
       ↓
ثبت نسخه اجراشده
       ↓
ادامه برنامه

Migration کمک می‌کند ساختار دیتابیس در نسخه‌های مختلف برنامه به‌شکل قابل‌کنترل تغییر کند.
🎯 تمرین
برای جدول products یک Migration شماره 2 طراحی کن که ستون description TEXT را اضافه کند.
برنامه باید ابتدا بررسی کند آیا نسخه 2 اجرا شده است یا نه؛ اگر اجرا نشده بود، ستون را اضافه کند و نسخه را ثبت کند.
پست ۲۲۷ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۲۸
تست خودکار عملیات SQLite با unittest
تا اینجا یاد گرفتیم چطور با SQLite داده ذخیره کنیم، Query بنویسیم و ساختار Database را مدیریت کنیم.
حالا می‌خواهیم بررسی کنیم که کدهای Database واقعاً درست کار می‌کنند یا نه.
برای این کار می‌توانیم از ماژول استاندارد unittest در Python استفاده کنیم.
🔹 ساخت تابع مورد آزمایش
فایل database.py را بساز:

import sqlite3


def add_user(connection, name):
    connection.execute(
        "INSERT INTO users (name) VALUES (?)",
        (name,)
    )


def get_users(connection):
    return connection.execute(
        "SELECT name FROM users ORDER BY id"
    ).fetchall()

این توابع از یک اتصال آماده استفاده می‌کنند؛ بنابراین می‌توانیم آن‌ها را با یک دیتابیس آزمایشی تست کنیم.
🔹 نوشتن تست
فایل test_database.py:

import sqlite3
import unittest

from database import add_user, get_users


class TestDatabase(unittest.TestCase):

    def setUp(self):
        self.connection = sqlite3.connect(":memory:")

        self.connection.execute("""
            CREATE TABLE users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL
            )
        """)

    def tearDown(self):
        self.connection.close()

    def test_add_user(self):
        add_user(self.connection, "Ali")

        users = get_users(self.connection)

        self.assertEqual(users, [("Ali",)])

    def test_multiple_users(self):
        add_user(self.connection, "Sara")
        add_user(self.connection, "Reza")

        users = get_users(self.connection)

        self.assertEqual(
            users,
            [("Sara",), ("Reza",)]
        )


if __name__ == "__main__":
    unittest.main()

🔹 اجرای تست‌ها
در ترمینال، داخل پوشه پروژه اجرا کن:

python -m unittest -v

اگر هر دو تست موفق باشند، نتیجه‌ای مشابه این می‌بینی:

test_add_user ... ok
test_multiple_users ... ok

Ran 2 tests
OK

🔹 setUp() و tearDown() چه هستند؟
setUp() قبل از هر تست اجرا می‌شود و محیط آزمایشی را آماده می‌کند.
tearDown() بعد از هر تست اجرا می‌شود و منابع را آزاد می‌کند.
assertEqual() نتیجه واقعی را با نتیجه مورد انتظار مقایسه می‌کند.
چون از :memory: استفاده کردیم، اطلاعات تست وارد فایل دیتابیس اصلی نمی‌شوند.
⚠️ نکته مهم
هر تست در این مثال اتصال و دیتابیس حافظه‌ای مستقل خودش را دارد. بنابراین نتیجه یک تست روی تست دیگر اثر نمی‌گذارد.
همچنین اگر تستی شکست بخورد، پیام خطا کمک می‌کند بفهمیم نتیجه واقعی با انتظارمان چه تفاوتی دارد.
🧠 خلاصه

setUp()
   ↓
آماده‌سازی دیتابیس آزمایشی
   ↓
اجرای تست
   ↓
assertEqual()
   ↓
tearDown()

تست خودکار کمک می‌کند بعد از تغییر کد، مطمئن شویم قابلیت‌های قبلی همچنان درست کار می‌کنند.
🎯 تمرین
تابعی به نام delete_user(connection, user_id) بساز که یک کاربر را با شناسه حذف کند.
سپس یک تست با unittest بنویس که:
دو کاربر درج کند.
یکی را حذف کند.
بررسی کند فقط کاربر باقی‌مانده در نتیجه get_users() وجود داشته باشد.
پست ۲۲۸ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۲۹
مدیریت امن خطاهای تراکنش در SQLite
در پست قبل تست‌کردن عملیات Database را یاد گرفتیم. حالا می‌خواهیم یکی از مهم‌ترین اصول کار با دیتابیس را بررسی کنیم: اگر یکی از عملیات شکست خورد، تغییرات ناقص باقی نمانند.
برای این کار از تراکنش (Transaction) و rollback() استفاده می‌کنیم.
🔹 مشکل اجرای چند عملیات
فرض کنیم می‌خواهیم دو حساب بانکی را به‌روزرسانی کنیم. اگر مبلغ از حساب اول کم شود اما افزایش موجودی حساب دوم با خطا مواجه شود، اطلاعات ناسازگار خواهد شد.
تراکنش کمک می‌کند عملیات مرتبط را به‌صورت یک واحد مدیریت کنیم.
🔹 استفاده از with connection:

import sqlite3

connection = sqlite3.connect("bank.db")

connection.execute("""
    CREATE TABLE IF NOT EXISTS accounts (
        id INTEGER PRIMARY KEY,
        balance INTEGER NOT NULL
    )
""")

connection.execute(
    "INSERT OR IGNORE INTO accounts (id, balance) VALUES (?, ?)",
    (1, 1000)
)

connection.execute(
    "INSERT OR IGNORE INTO accounts (id, balance) VALUES (?, ?)",
    (2, 500)
)

connection.commit()

try:
    with connection:
        connection.execute(
            "UPDATE accounts SET balance = balance - ? WHERE id = ?",
            (100, 1)
        )

        connection.execute(
            "UPDATE accounts SET balance = balance + ? WHERE id = ?",
            (100, 2)
        )

except sqlite3.Error as error:
    print("Transaction failed:", error)

connection.close()

در این مثال، هر دو تغییر داخل یک تراکنش قرار می‌گیرند. اگر خطای مناسب رخ دهد، تراکنش بازگردانده می‌شود؛ در غیر این صورت تغییرات ثبت می‌شوند.
🔹 بررسی نتیجه
بعد از تراکنش می‌توانیم موجودی‌ها را بخوانیم:

with sqlite3.connect("bank.db") as connection:
    accounts = connection.execute("""
        SELECT id, balance
        FROM accounts
        ORDER BY id
    """).fetchall()

print(accounts)

اگر انتقال موفق باشد، موجودی حساب اول 900 و حساب دوم 600 خواهد بود.
🔹 چرا rollback() مهم است؟
فرض کنیم چند تغییر انجام شده، اما یکی از آن‌ها شکست می‌خورد.

شروع تراکنش
      ↓
کاهش موجودی حساب اول
      ↓
خطا در عملیات بعدی
      ↓
rollback()
      ↓
لغو تغییرات تأییدنشده

به این ترتیب، تغییرات ناقص به‌عنوان یک تراکنش موفق ثبت نمی‌شوند.
⚠️ نکته مهم
with connection: در صورت موفقیت تراکنش را تأیید می‌کند و در صورت خروج همراه با Exception، تغییرات تأییدنشده را بازمی‌گرداند.
این بلوک اتصال را نمی‌بندد؛ برای بستن اتصال باید جداگانه از close() یا مدیریت مناسب منابع استفاده کنیم.
خطاهایی که خودمان از نظر منطقی تشخیص نمی‌دهیم، لزوماً Exception ایجاد نمی‌کنند. مثلاً اگر UPDATE هیچ ردیفی را تغییر ندهد، باید تعداد ردیف‌های تحت تأثیر را بررسی کنیم.
🧠 خلاصه

Transaction
    ↓
چند عملیات مرتبط
    ↓
موفقیت → commit
خطا    → rollback

🎯 تمرین
برنامه‌ای بنویس که در یک تراکنش:
از موجودی حساب اول 200 واحد کم کند.
به موجودی حساب دوم 200 واحد اضافه کند.
اگر هر عملیاتی خطا داد، تغییرات تراکنش را لغو کند.
در پایان موجودی هر دو حساب را چاپ کند.
پست ۲۲۹ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۰
جلوگیری از SQL Injection در SQLite
وقتی برنامه اطلاعات را از کاربر دریافت می‌کند، باید هنگام اجرای Queryها مراقب باشیم.
یکی از آسیب‌پذیری‌های مهم در کار با دیتابیس، SQL Injection است؛ زمانی که ورودی کنترل‌نشده کاربر باعث تغییر ساختار دستور SQL شود.
🔹 روش ناامن
فرض کنیم می‌خواهیم کاربری را بر اساس نام پیدا کنیم:

username = input("Username: ")

query = f"""
    SELECT * FROM users
    WHERE username = '{username}'
"""

cursor.execute(query)

در این روش، ورودی کاربر مستقیماً وارد دستور SQL می‌شود و ممکن است ساختار Query را تغییر دهد.
برای ساخت Query از f-string یا اتصال رشته‌ها با ورودی کاربر استفاده نکن.
🔹 روش امن با پارامترها
در sqlite3 از Placeholder استفاده می‌کنیم:

import sqlite3

connection = sqlite3.connect("users.db")

username = input("Username: ")

cursor = connection.execute(
    "SELECT id, username FROM users WHERE username = ?",
    (username,)
)

print(cursor.fetchall())

connection.close()

علامت ? محل مقدار ورودی را مشخص می‌کند.
این بخش:

(username,)

یک Tuple تک‌عضوی است؛ به همین دلیل بعد از username کاما قرار دارد.
SQLite مقدار ورودی را به‌عنوان داده پردازش می‌کند، نه بخشی از ساختار SQL.
🔹 پارامترگذاری در INSERT
همین روش برای ثبت اطلاعات نیز کاربرد دارد:

connection.execute(
    "INSERT INTO users (username, age) VALUES (?, ?)",
    ("Ali", 25)
)

connection.commit()

هر مقدار در Placeholder مربوط به خودش قرار می‌گیرد.
🔹 پارامترگذاری در UPDATE

connection.execute(
    "UPDATE users SET age = ? WHERE username = ?",
    (26, "Ali")
)

connection.commit()

ترتیب مقادیر باید با ترتیب Placeholderها مطابقت داشته باشد.
⚠️ نکته مهم
پارامترها برای مقدارها هستند، نه نام جدول یا ستون.
مثلاً نمی‌توانیم نام ستون را با ? جایگزین کنیم:

cursor.execute(
    "SELECT ? FROM users",
    ("username",)
)

این دستور ستون username را انتخاب نمی‌کند؛ بلکه مقدار ثابت "username" را برای هر ردیف برمی‌گرداند.
اگر نام ستون باید پویا باشد، آن را از یک فهرست مجاز و از پیش تعریف‌شده انتخاب کن؛ نه مستقیماً از ورودی کاربر.
🧠 خلاصه

ناامن:
f"SELECT ... WHERE name = '{name}'"

امن:
"SELECT ... WHERE name = ?", (name,)

قانون اصلی: برای انتقال داده به Queryهای SQL از Parameterized Query استفاده کن.
🎯 تمرین
جدول products شامل ستون‌های id، name و price است.
کدی بنویس که نام محصول را از کاربر بگیرد و محصول را با استفاده از ? جست‌وجو کند.
شرط مهم: ورودی کاربر نباید با f-string وارد دستور SQL شود.
پست ۲۳۰ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۱
مدیریت هم‌زمانی در SQLite با timeout
وقتی چند برنامه یا اتصال هم‌زمان به یک دیتابیس SQLite دسترسی دارند، ممکن است یکی از آن‌ها در حال نوشتن باشد و اتصال دیگر نتواند بلافاصله تغییرات خود را ثبت کند.
در این شرایط ممکن است خطای database is locked رخ دهد.
یکی از راه‌های مدیریت این وضعیت، استفاده از پارامتر timeout است.
🔹 استفاده از timeout

import sqlite3

connection = sqlite3.connect(
    "app.db",
    timeout=10
)

در این مثال، SQLite در صورت قفل بودن دیتابیس تا حدود ۱۰ ثانیه برای آزاد شدن آن منتظر می‌ماند.
اگر دیتابیس در این مدت قابل دسترسی نشود، ممکن است خطای مربوط به قفل بودن دیتابیس رخ دهد.
🔹 مدیریت خطا
می‌توانیم خطای احتمالی را مدیریت کنیم:

import sqlite3

try:
    with sqlite3.connect("app.db", timeout=5) as connection:
        connection.execute(
            "UPDATE products SET price = ? WHERE id = ?",
            (120, 1)
        )

except sqlite3.OperationalError as error:
    print("Database operation failed:", error)

در اینجا:
timeout=5 برای منتظر ماندن جهت آزاد شدن قفل استفاده می‌شود.
with connection: تراکنش را مدیریت می‌کند.
OperationalError می‌تواند خطاهایی مانند قفل بودن دیتابیس را دریافت کند.
توجه کن که OperationalError فقط مخصوص قفل دیتابیس نیست و ممکن است دلایل دیگری هم داشته باشد.
🔹 آیا timeout همه مشکلات را حل می‌کند؟
خیر. اگر یک اتصال برای مدت طولانی قفل نوشتن را نگه دارد، افزایش timeout فقط باعث می‌شود اتصال دیگر مدت بیشتری منتظر بماند.
برای کاهش احتمال این مشکل:
تراکنش‌ها را کوتاه نگه دار.
داخل تراکنش عملیات غیرضروری انجام نده.
هنگام استفاده از چند اتصال، خطاهای احتمالی را مدیریت کن.
برای سناریوهای نوشتن هم‌زمان زیاد، محدودیت‌های SQLite را در طراحی برنامه در نظر بگیر.
⚠️ نکته مهم
SQLite چند خواننده هم‌زمان را پشتیبانی می‌کند، اما نوشتن هم‌زمان محدودیت دارد؛ در یک لحظه فقط یک تراکنش می‌تواند نویسنده باشد.
پارامتر timeout امکان انتظار برای آزاد شدن قفل را فراهم می‌کند، اما جایگزین طراحی درست تراکنش‌ها نیست.
🧠 خلاصه

چند اتصال به دیتابیس
        ↓
قفل بودن دیتابیس
        ↓
timeout
        ↓
انتظار برای آزاد شدن قفل
        ↓
موفقیت یا بروز خطا

🎯 تمرین
برنامه‌ای بنویس که:
به دیتابیس app.db با timeout=8 متصل شود.
یک دستور SELECT اجرا کند.
خطاهای sqlite3.OperationalError را مدیریت کند.
در صورت موفقیت، نتایج را چاپ کند.
پست ۲۳۱ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۲
کنترل شروع تراکنش با BEGIN IMMEDIATE
در پست قبل با timeout آشنا شدیم.
حالا می‌خواهیم روش دیگری برای مدیریت تراکنش‌ها در SQLite یاد بگیریم: BEGIN IMMEDIATE.
این دستور زمانی مفید است که می‌خواهیم یک تراکنش را برای انجام عملیات نوشتن آغاز کنیم.
🔹 تفاوت BEGIN و BEGIN IMMEDIATE
در SQLite، دستور معمولی:

BEGIN;

یک تراکنش را با حالت پیش‌فرض DEFERRED آغاز می‌کند. در این حالت، قفل نوشتن لزوماً از همان ابتدا گرفته نمی‌شود.
اما:

BEGIN IMMEDIATE;

تراکنش را با درخواست دسترسی نوشتن آغاز می‌کند.
اگر اتصال دیگری در حال نوشتن باشد، ممکن است دستور منتظر بماند یا پس از پایان timeout با خطا مواجه شود.
🔹 مثال با Python

import sqlite3

connection = sqlite3.connect(
    "app.db",
    timeout=5,
    isolation_level=None
)

try:
    connection.execute("BEGIN IMMEDIATE")

    connection.execute(
        "UPDATE products SET price = ? WHERE id = ?",
        (150, 1)
    )

    connection.execute("COMMIT")
    print("Transaction completed")

except sqlite3.Error as error:
    if connection.in_transaction:
        connection.execute("ROLLBACK")

    print("Transaction failed:", error)

finally:
    connection.close()

در این مثال:
isolation_level=None مدیریت ضمنی تراکنش را غیرفعال می‌کند تا شروع تراکنش را صریحاً کنترل کنیم.
BEGIN IMMEDIATE تراکنش را با درخواست قفل نوشتن آغاز می‌کند.
COMMIT تغییرات را ثبت می‌کند.
ROLLBACK تغییرات تأییدنشده را لغو می‌کند.
connection.in_transaction بررسی می‌کند که آیا تراکنشی فعال است یا نه.
🔹 چرا از BEGIN IMMEDIATE استفاده کنیم؟
فرض کن ابتدا موجودی یک حساب را می‌خوانیم و سپس می‌خواهیم آن را تغییر دهیم.
اگر چند اتصال هم‌زمان همین کار را انجام دهند، ممکن است بین خواندن و نوشتن تداخل ایجاد شود.
با شروع تراکنش به روش IMMEDIATE، درخواست نوشتن از ابتدای تراکنش مشخص می‌شود و احتمال مواجهه با بعضی خطاهای ارتقای تراکنش کاهش می‌یابد.
البته این روش جایگزین طراحی صحیح عملیات و بررسی داده‌ها نیست.
⚠️ نکته مهم
از BEGIN IMMEDIATE برای هر Query ساده استفاده نکن. این روش برای تراکنش‌هایی مناسب‌تر است که می‌دانیم قرار است عملیات نوشتن انجام دهند.
همچنین هنگام استفاده از آن، تراکنش را کوتاه نگه دار؛ چون تا پایان تراکنش، اتصال‌های دیگر ممکن است نتوانند بنویسند.
🧠 خلاصه

BEGIN
  → شروع تراکنش با حالت DEFERRED

BEGIN IMMEDIATE
  → شروع تراکنش با درخواست دسترسی نوشتن

COMMIT
  → ثبت تغییرات

ROLLBACK
  → لغو تغییرات تأییدنشده

🎯 تمرین
کدی بنویس که:
به app.db متصل شود.
با BEGIN IMMEDIATE تراکنش را آغاز کند.
قیمت محصولی با id = 1 را تغییر دهد.
در صورت موفقیت، تغییرات را ثبت کند.
در صورت بروز خطا، تراکنش فعال را لغو کند.
در پایان اتصال را ببندد.
پست ۲۳۲ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۳
کنترل تراکنش با SAVEPOINT در SQLite
در پست‌های قبل یاد گرفتیم با COMMIT تغییرات را ثبت و با ROLLBACK تغییرات تأییدنشده را لغو کنیم.
اما گاهی یک تراکنش شامل چند مرحله است و می‌خواهیم فقط بخشی از تغییرات را برگردانیم، بدون اینکه کل تراکنش لغو شود.
برای این کار از SAVEPOINT استفاده می‌کنیم.
🔹 ساخت Savepoint
با SAVEPOINT یک نقطه بازگشت داخل تراکنش ایجاد می‌کنیم:

import sqlite3

connection = sqlite3.connect("app.db")

try:
    connection.execute("BEGIN")

    connection.execute(
        "UPDATE products SET price = ? WHERE id = ?",
        (100, 1)
    )

    connection.execute("SAVEPOINT second_step")

    connection.execute(
        "UPDATE products SET price = ? WHERE id = ?",
        (200, 2)
    )

    connection.execute("RELEASE SAVEPOINT second_step")
    connection.commit()

except sqlite3.Error:
    connection.rollback()
    raise

finally:
    connection.close()

در این مثال، ابتدا قیمت محصول اول تغییر می‌کند و سپس یک Savepoint ایجاد می‌شود.
🔹 بازگشت به Savepoint
فرض کنیم می‌خواهیم تغییر دوم را لغو کنیم، اما تغییر اول باقی بماند:

import sqlite3

connection = sqlite3.connect("app.db")

try:
    connection.execute("BEGIN")

    connection.execute(
        "UPDATE products SET price = ? WHERE id = ?",
        (100, 1)
    )

    connection.execute("SAVEPOINT second_step")

    connection.execute(
        "UPDATE products SET price = ? WHERE id = ?",
        (200, 2)
    )

    connection.execute("ROLLBACK TO SAVEPOINT second_step")

    connection.execute("RELEASE SAVEPOINT second_step")

    connection.commit()

except sqlite3.Error:
    connection.rollback()
    raise

finally:
    connection.close()

اینجا تغییر قیمت محصول دوم لغو می‌شود، اما تغییر محصول اول می‌تواند ثبت شود.
🔹 تفاوت سه دستور مهم
دستور
کاربرد
SAVEPOINT
ایجاد نقطه بازگشت
ROLLBACK TO SAVEPOINT
بازگشت به نقطه مشخص
RELEASE SAVEPOINT
آزاد کردن Savepoint
نکته: ROLLBACK TO SAVEPOINT تراکنش اصلی را تمام نمی‌کند؛ تراکنش همچنان فعال است.
همچنین RELEASE SAVEPOINT به‌تنهایی تغییرات را به‌صورت نهایی ثبت نمی‌کند؛ برای ثبت تراکنش اصلی باید از COMMIT استفاده کنیم.
⚠️ نکته مهم
اگر خطایی رخ دهد و بخواهیم کل تراکنش را لغو کنیم، می‌توانیم از:

connection.rollback()

استفاده کنیم.
اما اگر فقط بخشی از عملیات باید لغو شود، SAVEPOINT کنترل دقیق‌تری فراهم می‌کند.
🧠 خلاصه

BEGIN
  ↓
عملیات اول
  ↓
SAVEPOINT
  ↓
عملیات دوم
  ↓
ROLLBACK TO SAVEPOINT
  ↓
ادامه تراکنش
  ↓
COMMIT

🎯 تمرین
برنامه‌ای بنویس که:
یک تراکنش آغاز کند.
قیمت محصول اول را تغییر دهد.
یک Savepoint ایجاد کند.
قیمت محصول دوم را تغییر دهد.
به Savepoint برگردد تا تغییر دوم لغو شود.
تراکنش را ثبت کند.
پست ۲۳۳ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۴
اجرای هم‌زمان چند Query با executescript()
تا اینجا بیشتر از execute() برای اجرای یک دستور SQL استفاده کردیم.
اما اگر بخواهیم چند دستور SQL را پشت سر هم اجرا کنیم، می‌توانیم از executescript() استفاده کنیم.
🔹 استفاده ساده

import sqlite3

connection = sqlite3.connect("app.db")

connection.executescript("""
    CREATE TABLE IF NOT EXISTS categories (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL
    );

    INSERT INTO categories (name)
    VALUES ('Programming');

    INSERT INTO categories (name)
    VALUES ('Design');
""")

connection.commit()
connection.close()

در این مثال، چند دستور SQL در یک رشته قرار گرفته‌اند و با executescript() اجرا می‌شوند.
🔹 بررسی نتیجه
بعد از اجرای دستورات می‌توانیم اطلاعات را بخوانیم:

import sqlite3

with sqlite3.connect("app.db") as connection:
    categories = connection.execute(
        "SELECT id, name FROM categories"
    ).fetchall()

    print(categories)

خروجی در اجرای اول، با فرض خالی بودن جدول:

[(1, 'Programming'), (2, 'Design')]

اگر کد درج اطلاعات را دوباره اجرا کنیم، دو ردیف دیگر اضافه می‌شوند؛ چون در مثال بالا از شرط جلوگیری از درج تکراری استفاده نکرده‌ایم.
🔹 تفاوت execute() و executescript()
متد
کاربرد
execute()
اجرای یک دستور SQL
executemany()
اجرای یک دستور با چند مجموعه مقدار
executescript()
اجرای چند دستور SQL در یک اسکریپت
مثلاً برای ثبت چند کاربر با یک دستور تکرارشونده، معمولاً executemany() مناسب‌تر است؛ اما برای ساخت جدول و اجرای چند دستور متفاوت، executescript() کاربرد دارد.
⚠️ نکته مهم درباره تراکنش‌ها
رفتار تراکنش در executescript() به نسخه Python و شیوه مدیریت تراکنش بستگی دارد. در حالت مدیریت تراکنش قدیمی، این متد می‌تواند تراکنش باز قبلی را پیش از اجرای اسکریپت Commit کند.
اگر به اتمی بودن اجرای چند دستور نیاز داری، مرز تراکنش را صریحاً در طراحی اسکریپت مشخص کن و رفتار آن را در حالت اتصال مورد استفاده‌ات بشناس.
همچنین ورودی کاربر را مستقیماً داخل رشته اسکریپت قرار نده؛ executescript() برای پارامترگذاری مقادیر مثل execute() مناسب نیست. برای مقادیر پویا از Query پارامتری استفاده کن.
🧠 خلاصه

execute()      → یک دستور SQL
executemany()  → یک دستور با چند مجموعه مقدار
executescript() → چند دستور SQL متفاوت

🎯 تمرین
با استفاده از executescript():
جدولی به نام courses بساز که ستون‌های id و title داشته باشد.
دو درس با نام‌های Python و SQLite به آن اضافه کن.
با SELECT تمام درس‌ها را بخوان و چاپ کن.
پست ۲۳۴ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۵
استفاده از sqlite3 با contextlib.closing
در پست‌های قبل یاد گرفتیم اتصال دیتابیس را با close() ببندیم و از with connection: برای مدیریت تراکنش استفاده کنیم.
اما یک نکته مهم وجود دارد:
with connection: اتصال SQLite را نمی‌بندد؛ فقط تراکنش را مدیریت می‌کند.
برای بستن خودکار اتصال می‌توانیم از closing() در ماژول contextlib استفاده کنیم.
🔹 استفاده از closing()

import sqlite3
from contextlib import closing

with closing(sqlite3.connect("app.db")) as connection:
    connection.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL
        )
    """)

    connection.commit()

وقتی از بلوک with خارج شویم، connection.close() به‌صورت خودکار فراخوانی می‌شود.
🔹 مدیریت هم‌زمان اتصال و تراکنش
برای مدیریت هر دو مورد می‌توانیم closing() را با with connection: ترکیب کنیم:

import sqlite3
from contextlib import closing

with closing(sqlite3.connect("app.db")) as connection:
    with connection:
        connection.execute(
            "INSERT INTO users (name) VALUES (?)",
            ("Ali",)
        )

در این مثال:
with connection: تراکنش را Commit یا Rollback می‌کند.
closing(connection) اتصال را هنگام خروج می‌بندد.
🔹 چرا ترتیب مهم است؟
ابتدا باید تراکنش تمام شود و سپس اتصال بسته شود.
در کد بالا، با خروج از بلوک داخلی، تراکنش مدیریت می‌شود؛ سپس با خروج از بلوک خارجی، اتصال بسته خواهد شد.

شروع اتصال
    ↓
اجرای Queryها
    ↓
مدیریت تراکنش
    ↓
بستن اتصال

⚠️ نکته مهم
اگر عملیات داخل بلوک داخلی با Exception مواجه شود، تراکنش Rollback می‌شود. سپس بلوک خارجی نیز خارج شده و اتصال بسته می‌شود.
همچنین closing() به‌تنهایی تراکنش را مدیریت نمی‌کند؛ فقط متد close() را هنگام خروج فراخوانی می‌کند.
🧠 خلاصه

with connection:
    → مدیریت تراکنش

closing(connection):
    → بستن خودکار اتصال

🎯 تمرین
برنامه‌ای بنویس که:
به دیتابیس school.db متصل شود.
جدولی به نام students بسازد.
نام یک دانش‌آموز را با Query پارامتری ثبت کند.
تراکنش را مدیریت کند.
اتصال را با closing() به‌صورت خودکار ببندد.
پست ۲۳۵ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۶
ساخت لایه دسترسی به داده با Repository Pattern در Python
در پروژه‌های کوچک ممکن است تمام Queryهای SQLite را داخل یک فایل بنویسیم؛ اما با بزرگ‌تر شدن پروژه، نگهداری و تست کد دشوارتر می‌شود.
یکی از روش‌های سازمان‌دهی کد، Repository Pattern است.
در این الگو، عملیات مربوط به دریافت و ذخیره اطلاعات را در یک کلاس جداگانه قرار می‌دهیم.
🔹 ساختار پروژه

project/
├── database.py
├── user_repository.py
└── main.py

database.py مسئول ایجاد اتصال است.
user_repository.py عملیات مربوط به کاربران را مدیریت می‌کند.
main.py منطق اصلی برنامه را اجرا می‌کند.
🔹 فایل database.py

import sqlite3


def get_connection():
    connection = sqlite3.connect("app.db")
    connection.row_factory = sqlite3.Row
    return connection

با row_factory می‌توانیم مقادیر هر ردیف را با نام ستون نیز بخوانیم.
🔹 فایل user_repository.py

class UserRepository:

    def __init__(self, connection):
        self.connection = connection

    def create_table(self):
        self.connection.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL
            )
        """)

    def add_user(self, name):
        cursor = self.connection.execute(
            "INSERT INTO users (name) VALUES (?)",
            (name,)
        )
        return cursor.lastrowid

    def get_all(self):
        return self.connection.execute(
            "SELECT id, name FROM users ORDER BY id"
        ).fetchall()

این کلاس مسئول اجرای Queryهای مربوط به کاربران است؛ اما خودش اتصال دیتابیس را ایجاد نمی‌کند.
🔹 فایل main.py

from contextlib import closing
from database import get_connection
from user_repository import UserRepository


with closing(get_connection()) as connection:
    with connection:
        users = UserRepository(connection)

        users.create_table()
        user_id = users.add_user("Ali")

        print("New user ID:", user_id)

        for user in users.get_all():
            print(user["id"], user["name"])

خروجی نمونه:

New user ID: 1
1 Ali

اگر دیتابیس از قبل کاربر داشته باشد، شناسه جدید ممکن است عدد دیگری باشد.
🔹 چرا این روش مفید است؟
Queryها از منطق اصلی برنامه جدا می‌شوند.
کد قابل‌خواندن‌تر و نگهداری آن ساده‌تر می‌شود.
می‌توانیم عملیات دیتابیس را مستقل‌تر تست کنیم.
در صورت تغییر ساختار ذخیره‌سازی، تغییرات در یک بخش متمرکز می‌شوند.
⚠️ نکته مهم
Repository Pattern به‌خودی‌خود تراکنش یا اتصال را مدیریت نمی‌کند. در مثال ما، main.py مسئول مدیریت تراکنش و بستن اتصال است.
همچنین در پروژه‌های بزرگ‌تر می‌توانیم اعتبارسنجی ورودی‌ها، مدیریت خطاها و عملیات بیشتری مانند get_by_id() و delete_user() را به Repository اضافه کنیم.
🧠 خلاصه

main.py
   ↓
UserRepository
   ↓
SQLite

Repository عملیات دسترسی به داده را متمرکز می‌کند و منطق اصلی برنامه را از جزئیات Queryها جدا نگه می‌دارد.
🎯 تمرین
کلاس UserRepository را گسترش بده و متدی به نام get_by_id(user_id) بساز که کاربر موردنظر را با شناسه پیدا کند.
اگر کاربری با آن شناسه وجود نداشت، باید None برگردانده شود.
پست ۲۳۶ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۷
پیاده‌سازی CRUD با Repository Pattern در SQLite
در پست قبل با Repository Pattern آشنا شدیم و عملیات دریافت و ذخیره کاربران را داخل یک کلاس قرار دادیم.
حالا می‌خواهیم چهار عملیات اصلی مدیریت داده را کامل کنیم که به آن‌ها CRUD می‌گویند.
Create: ایجاد رکورد
Read: خواندن رکورد
Update: ویرایش رکورد
Delete: حذف رکورد
🔹 تکمیل کلاس UserRepository
در فایل user_repository.py:

class UserRepository:

    def __init__(self, connection):
        self.connection = connection

    def create_table(self):
        self.connection.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL
            )
        """)

    # Create
    def add_user(self, name):
        cursor = self.connection.execute(
            "INSERT INTO users (name) VALUES (?)",
            (name,)
        )
        return cursor.lastrowid

    # Read
    def get_by_id(self, user_id):
        return self.connection.execute(
            "SELECT id, name FROM users WHERE id = ?",
            (user_id,)
        ).fetchone()

    def get_all(self):
        return self.connection.execute(
            "SELECT id, name FROM users ORDER BY id"
        ).fetchall()

    # Update
    def update_user(self, user_id, new_name):
        cursor = self.connection.execute(
            "UPDATE users SET name = ? WHERE id = ?",
            (new_name, user_id)
        )
        return cursor.rowcount > 0

    # Delete
    def delete_user(self, user_id):
        cursor = self.connection.execute(
            "DELETE FROM users WHERE id = ?",
            (user_id,)
        )
        return cursor.rowcount > 0

🔹 استفاده از عملیات CRUD
در فایل main.py:

from contextlib import closing
from database import get_connection
from user_repository import UserRepository


with closing(get_connection()) as connection:
    with connection:
        users = UserRepository(connection)

        users.create_table()

        user_id = users.add_user("Ali")
        print("Created:", user_id)

        user = users.get_by_id(user_id)
        print("Read:", dict(user) if user else None)

        updated = users.update_user(user_id, "Reza")
        print("Updated:", updated)

        deleted = users.delete_user(user_id)
        print("Deleted:", deleted)

خروجی نمونه:

Created: 1
Read: {'id': 1, 'name': 'Ali'}
Updated: True
Deleted: True

شناسه ممکن است در دیتابیس موجود شما متفاوت باشد.
🔹 چرا متدهای Update و Delete مقدار bool برمی‌گردانند؟
در این دو متد:

return cursor.rowcount > 0

اگر حداقل یک ردیف تغییر کرده یا حذف شده باشد، نتیجه True است؛ در غیر این صورت False خواهد بود.
به این ترتیب، برنامه می‌تواند تشخیص دهد آیا رکورد موردنظر پیدا شده و عملیات روی آن انجام شده است یا خیر.
⚠️ نکته مهم
تمام Queryها از Placeholder یعنی ? استفاده می‌کنند. این کار از ترکیب مستقیم ورودی کاربر با رشته SQL جلوگیری می‌کند.
همچنین with connection: تراکنش را مدیریت می‌کند و closing() اتصال را پس از خروج از بلوک می‌بندد.
🧠 خلاصه

Create → add_user()
Read   → get_by_id(), get_all()
Update → update_user()
Delete → delete_user()

این چهار عملیات پایه بسیاری از برنامه‌های واقعی مانند مدیریت کاربران، محصولات و سفارش‌ها هستند.
🎯 تمرین
متدی به نام rename_user(user_id, new_name) طراحی کن که:
کاربر را بر اساس شناسه پیدا کند.
نام او را تغییر دهد.
مشخص کند آیا رکوردی تغییر کرده است یا خیر.
می‌توانی از متد update_user() استفاده مجدد کنی، بدون اینکه Query تکراری بنویسی.
پست ۲۳۷ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۸
اعتبارسنجی ورودی‌ها در Repository
در پست قبل عملیات CRUD را پیاده‌سازی کردیم.
اما یک Repository خوب نباید هر داده‌ای را بدون بررسی وارد دیتابیس کند. برای مثال، نام کاربر نباید خالی یا فقط شامل فاصله باشد.
در این پست یاد می‌گیریم چگونه پیش از اجرای Query، ورودی‌ها را اعتبارسنجی کنیم.
🔹 اعتبارسنجی نام کاربر
در فایل user_repository.py، متد add_user() را به این شکل تغییر بده:

def add_user(self, name):
    if not isinstance(name, str):
        raise TypeError("Name must be a string")

    name = name.strip()

    if not name:
        raise ValueError("Name cannot be empty")

    cursor = self.connection.execute(
        "INSERT INTO users (name) VALUES (?)",
        (name,)
    )

    return cursor.lastrowid

در این کد:
isinstance() نوع داده را بررسی می‌کند.
strip() فاصله‌های ابتدا و انتهای نام را حذف می‌کند.
اگر نام خالی باشد، ValueError ایجاد می‌شود.
فقط پس از اعتبارسنجی، دستور SQL اجرا می‌شود.
🔹 اعتبارسنجی شناسه
برای متدهای ویرایش و حذف نیز می‌توانیم شناسه را بررسی کنیم:

def validate_user_id(self, user_id):
    if isinstance(user_id, bool) or not isinstance(user_id, int):
        raise TypeError("User ID must be an integer")

    if user_id <= 0:
        raise ValueError("User ID must be positive")

سپس در متد ویرایش:

def update_user(self, user_id, new_name):
    self.validate_user_id(user_id)

    if not isinstance(new_name, str):
        raise TypeError("Name must be a string")

    new_name = new_name.strip()

    if not new_name:
        raise ValueError("Name cannot be empty")

    cursor = self.connection.execute(
        "UPDATE users SET name = ? WHERE id = ?",
        (new_name, user_id)
    )

    return cursor.rowcount > 0

نکته: در Python، مقدار bool زیرنوع int است؛ بنابراین برای رد کردن True و False باید آن‌ها را جداگانه بررسی کنیم.
🔹 مدیریت خطا هنگام استفاده
در فایل main.py:

try:
    user_id = users.add_user("   ")

except (TypeError, ValueError) as error:
    print("Invalid input:", error)

خروجی:

Invalid input: Name cannot be empty

با مدیریت خطا، برنامه می‌تواند به‌جای متوقف شدن ناگهانی، پیام مناسبی نمایش دهد.
⚠️ نکته مهم
اعتبارسنجی در Repository به کیفیت داده‌ها کمک می‌کند؛ اما جایگزین محدودیت‌های دیتابیس مانند NOT NULL و CHECK نیست.
بهتر است از هر دو استفاده کنیم:
اعتبارسنجی Python برای ارائه خطاهای واضح.
محدودیت‌های SQLite برای محافظت از یکپارچگی داده‌ها.
🧠 خلاصه

ورودی
  ↓
بررسی نوع
  ↓
بررسی مقدار
  ↓
اجرای Query پارامتری
  ↓
ثبت یا دریافت نتیجه

🎯 تمرین
متد delete_user() را طوری تغییر بده که:
شناسه باید از نوع int باشد.
مقدارهای True و False پذیرفته نشوند.
شناسه باید بزرگ‌تر از صفر باشد.
در صورت معتبر بودن، رکورد حذف شود.
پست ۲۳۸ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۳۹
مدیریت خطاهای سفارشی در پروژه‌های Python
در پست قبل یاد گرفتیم ورودی‌های Repository را اعتبارسنجی کنیم.
حالا می‌خواهیم به‌جای استفاده از خطاهای عمومی در تمام بخش‌های برنامه، Exceptionهای اختصاصی برای پروژه خودمان بسازیم.
این روش باعث می‌شود مدیریت خطاها دقیق‌تر و کد خواناتر شود.
🔹 ساخت Exception اختصاصی
در Python می‌توانیم با ارث‌بری از Exception، خطای مخصوص پروژه بسازیم:

class UserError(Exception):
    pass


class InvalidUserError(UserError):
    pass


class UserNotFoundError(UserError):
    pass

در اینجا:
UserError خطای پایه مربوط به کاربران است.
InvalidUserError برای ورودی نامعتبر است.
UserNotFoundError برای کاربر پیدا نشده است.
چون دو خطای آخر از UserError ارث‌بری می‌کنند، می‌توانیم همه آن‌ها را با گرفتن UserError مدیریت کنیم.
🔹 استفاده در Repository
در فایل user_repository.py:

class UserRepository:

    def __init__(self, connection):
        self.connection = connection

    def add_user(self, name):
        if not isinstance(name, str):
            raise InvalidUserError("Name must be a string")

        name = name.strip()

        if not name:
            raise InvalidUserError("Name cannot be empty")

        cursor = self.connection.execute(
            "INSERT INTO users (name) VALUES (?)",
            (name,)
        )

        return cursor.lastrowid

    def get_by_id(self, user_id):
        cursor = self.connection.execute(
            "SELECT id, name FROM users WHERE id = ?",
            (user_id,)
        )

        user = cursor.fetchone()

        if user is None:
            raise UserNotFoundError(
                f"User with ID {user_id} was not found"
            )

        return user

در این مثال فرض شده کلاس‌های Exception در همین فایل تعریف شده‌اند.
🔹 مدیریت خطا در برنامه اصلی
در main.py:

try:
    user = users.get_by_id(999)

except UserNotFoundError as error:
    print("User error:", error)

except UserError as error:
    print("Invalid operation:", error)

اگر کاربر با شناسه 999 وجود نداشته باشد، پیام مناسب نمایش داده می‌شود.
🔹 چرا ترتیب except مهم است؟
فرض کنیم:

class UserError(Exception):
    pass


class UserNotFoundError(UserError):
    pass

اگر ابتدا بنویسیم:

except UserError:
    ...
except UserNotFoundError:
    ...

خطای UserNotFoundError ابتدا توسط UserError گرفته می‌شود؛ بنابراین بخش دوم اجرا نخواهد شد.
به همین دلیل، ابتدا خطاهای اختصاصی‌تر و سپس خطاهای عمومی‌تر را بنویس.
⚠️ نکته مهم
Exception اختصاصی را برای خطاهایی بساز که معنای مشخصی در پروژه دارند. لازم نیست برای هر شرط ساده یک کلاس جدید ایجاد کنیم.
همچنین خطاهای دیتابیس، مانند sqlite3.IntegrityError، همچنان ممکن است رخ دهند و باید متناسب با نیاز پروژه مدیریت شوند.
🧠 خلاصه

Exception
    ↓
UserError
    ├── InvalidUserError
    └── UserNotFoundError

با Exceptionهای اختصاصی می‌توانیم خطاهای پروژه را بهتر دسته‌بندی و مدیریت کنیم.
🎯 تمرین
یک Exception به نام DuplicateUserError بساز که از UserError ارث‌بری کند.
سپس آن را در Repository به کار ببر تا اگر نام کاربری تکراری بود، خطای مشخصی ایجاد شود.
پست ۲۳۹ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۴۰
ثبت خطاها با ماژول logging
در پست قبل با Exceptionهای اختصاصی آشنا شدیم.
حالا می‌خواهیم یاد بگیریم چگونه اتفاقات مهم برنامه و خطاها را ثبت کنیم تا هنگام عیب‌یابی بتوانیم آن‌ها را بررسی کنیم.
برای این کار از ماژول استاندارد logging استفاده می‌کنیم.
🔹 اولین Log

import logging

logging.basicConfig(level=logging.INFO)

logging.info("Application started")
logging.warning("Low disk space")
logging.error("Database connection failed")

خروجی نمونه:

INFO:root:Application started
WARNING:root:Low disk space
ERROR:root:Database connection failed

سطح‌های مهم Log عبارت‌اند از:
سطح
کاربرد
DEBUG
اطلاعات جزئی برای عیب‌یابی
INFO
رویدادهای عادی برنامه
WARNING
وضعیت غیرمنتظره اما قابل ادامه
ERROR
خطا در یک عملیات
CRITICAL
مشکل بسیار جدی
🔹 ثبت Exception
فرض کنیم هنگام اتصال به دیتابیس خطایی رخ می‌دهد:

import logging
import sqlite3

logging.basicConfig(level=logging.INFO)

try:
    connection = sqlite3.connect("app.db")
    connection.execute("SELECT * FROM missing_table")

except sqlite3.Error:
    logging.exception("Database operation failed")

finally:
    if "connection" in locals():
        connection.close()

متد logging.exception() پیام خطا را همراه با اطلاعات Exception و Traceback ثبت می‌کند.
این متد را معمولاً داخل بلوک except استفاده می‌کنیم.
🔹 ذخیره Log در فایل
برای ثبت گزارش در فایل:

import logging

logging.basicConfig(
    filename="app.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    encoding="utf-8"
)

logging.info("Application started")
logging.error("Operation failed")

اکنون پیام‌ها در فایل app.log ذخیره می‌شوند.
گزینه encoding="utf-8" برای ثبت درست متن فارسی مفید است.
🔹 چرا print() کافی نیست؟
print() برای نمایش خروجی مناسب است، اما logging امکانات بیشتری دارد:
سطح‌بندی پیام‌ها
ثبت زمان رویداد
ذخیره گزارش در فایل
ثبت Traceback خطاها
کنترل میزان جزئیات گزارش
در برنامه‌های واقعی، Logging به عیب‌یابی و نگهداری بهتر کمک می‌کند.
⚠️ نکته مهم
logging.basicConfig() معمولاً فقط زمانی تنظیمات اصلی را اعمال می‌کند که Root Logger از قبل Handler نداشته باشد.
همچنین رمز عبور، کلید API و اطلاعات حساس کاربران را در Log ثبت نکن؛ فایل‌های گزارش نیز باید از نظر دسترسی محافظت شوند.
🧠 خلاصه

DEBUG
INFO
WARNING
ERROR
CRITICAL

برای ثبت Exception همراه با Traceback:

logging.exception("Operation failed")

و برای ذخیره گزارش در فایل:

logging.basicConfig(filename="app.log")

🎯 تمرین
برنامه‌ای بنویس که:
گزارش‌ها را در فایل database.log ذخیره کند.
شروع برنامه را با INFO ثبت کند.
یک خطای آزمایشی ValueError ایجاد کند.
خطا را با logging.exception() ثبت کند.
اجرای برنامه را با یک پیام INFO گزارش دهد.
پست ۲۴۰ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پروژه عملی شماره ۱
ساخت سیستم مدیریت کاربران با SQLite
در این پروژه قرار است یک برنامه نسبتاً واقعی بسازیم که کاربران را در دیتابیس ذخیره کند، اطلاعاتشان را مدیریت کند و خطاها را ثبت کند.
🎯 هدف پروژه: ترکیب مفاهیمی که تا اینجا یاد گرفته‌ایم در یک برنامه کاربردی.
🔹 امکانات برنامه
برنامه باید بتواند:
کاربر جدید ثبت کند.
تمام کاربران را نمایش دهد.
کاربر را با شناسه پیدا کند.
اطلاعات کاربر را ویرایش کند.
کاربر را حذف کند.
ورودی‌ها را اعتبارسنجی کند.
خطاها را مدیریت و در فایل ثبت کند.
اطلاعات را در SQLite نگهداری کند تا با بستن برنامه از بین نروند.
🔹 ساختار پروژه

user_manager/
│
├── main.py
├── database.py
├── user_repository.py
├── exceptions.py
└── app.log

وظیفه هر فایل:
main.py — منوی برنامه و دریافت ورودی
database.py — ایجاد اتصال به دیتابیس
user_repository.py — عملیات CRUD
exceptions.py — تعریف خطاهای اختصاصی
app.log — ذخیره گزارش خطاها و رویدادها
🔹 مرحله اول: طراحی جدول
جدول users باید این ستون‌ها را داشته باشد:
ستون
نوع
محدودیت
id
INTEGER
کلید اصلی
username
TEXT
اجباری و یکتا
age
INTEGER
بین ۱ تا ۱۲۰
از محدودیت‌های PRIMARY KEY، NOT NULL، UNIQUE و CHECK استفاده کن.
🔹 مرحله دوم: پیاده‌سازی Repository
در کلاس UserRepository این متدها را پیاده‌سازی کن:

create_table()
add_user(username, age)
get_all()
get_by_id(user_id)
update_user(user_id, username, age)
delete_user(user_id)

قوانین:
نام کاربری نباید خالی باشد.
نام کاربری تکراری نباید ثبت شود.
سن باید عدد صحیح بین ۱ تا ۱۲۰ باشد.
شناسه باید عدد صحیح مثبت باشد.
برای تمام Queryهای دارای ورودی از Placeholder یعنی ? استفاده کن.
🔹 مرحله سوم: مدیریت خطا
در فایل exceptions.py دست‌کم این خطاها را تعریف کن:

class UserError(Exception):
    pass


class InvalidUserError(UserError):
    pass


class UserNotFoundError(UserError):
    pass


class DuplicateUserError(UserError):
    pass

خطاها باید متناسب با شرایط ایجاد شوند و در main.py به شکل مناسب مدیریت شوند.
برای خطای تکراری بودن نام کاربری می‌توانی از sqlite3.IntegrityError استفاده کنی؛ اما فقط زمانی آن را به DuplicateUserError تبدیل کن که علت نقض محدودیت واقعاً تکراری بودن نام باشد.
🔹 مرحله چهارم: ثبت گزارش
با logging رویدادهای مهم را در app.log ثبت کن:

INFO     User created
INFO     User updated
INFO     User deleted
WARNING  Invalid input
ERROR    Database operation failed

اطلاعات حساس را در گزارش‌ها ذخیره نکن.
🔹 مرحله پنجم: منوی برنامه
منوی زیر را پیاده‌سازی کن:

===== User Manager =====

1. Add user
2. Show all users
3. Find user
4. Update user
5. Delete user
0. Exit

برنامه باید تا زمانی که کاربر گزینه خروج را انتخاب نکرده، به اجرای منو ادامه دهد.
⚠️ قوانین مهم پروژه
کد را بین فایل‌های مشخص‌شده تقسیم کن.
از try/except برای مدیریت خطاهای مورد انتظار استفاده کن.
اتصال دیتابیس را به‌درستی ببند.
تراکنش‌ها را مدیریت کن تا تغییرات ناقص ثبت نشوند.
از تکرار بی‌دلیل Queryها جلوگیری کن.
اعتبارسنجی Python را در کنار محدودیت‌های SQLite انجام بده.
🎯 معیار تکمیل پروژه
پروژه زمانی کامل است که بتوانی:
دو کاربر متفاوت ثبت کنی.
کاربر تکراری را رد کنی.
اطلاعات یک کاربر را تغییر بدهی.
کاربر مشخصی را حذف کنی.
شناسه نامعتبر را مدیریت کنی.
برنامه را ببندی و دوباره اجرا کنی، بدون اینکه اطلاعات قبلی از بین بروند.
گزارش خطاها را در app.log مشاهده کنی.
پست پروژه‌ای از دوره جامع Python 🐍

هیچ صحنه ای توی دنیا
اندازه این صحنه که چیزی در حال لود شدن یا هر چیز دیگه ای روی اعصابم راه نمیره

📢 اطلاعیه درباره روند انتشار آموزش‌های Python 🐍
دوستان عزیز، همراهان دوره جامع Python،
با توجه به حجم مطالبی که تا اینجا در دوره منتشر شده، تصمیم گرفتیم فعلاً سرعت انتشار آموزش‌های جدید را کمی کاهش دهیم تا فرصت کافی برای مطالعه، مرور و تمرین مطالب قبلی داشته باشید.
🎯 هدف ما فقط انتشار مطالب بیشتر نیست؛ هدف این است که مطالب را به‌خوبی یاد بگیرید و بتوانید در پروژه‌های واقعی از آن‌ها استفاده کنید.
پیشنهاد می‌کنیم در این مدت:
🔹 پست‌های قبلی را با دقت مرور کنید.
🔹 مثال‌های کد را خودتان اجرا کنید.
🔹 تمرین‌های هر بخش را حل کنید.
🔹 سعی کنید بدون نگاه کردن به آموزش، کدها را دوباره بنویسید.
🔹 اگر بخشی را خوب متوجه نشده‌اید، به همان قسمت برگردید و دوباره مطالعه کنید.
⏳ این کاهش سرعت موقتی است و به معنی توقف دوره نیست. آموزش‌ها ادامه خواهند داشت؛ فقط می‌خواهیم فرصت بیشتری برای یادگیری عمیق‌تر و تثبیت مطالب داشته باشید.
از همراهی و حمایت شما سپاسگزاریم. ❤️
آرام‌تر پیش رفتن، اما عمیق‌تر یاد گرفتن؛ اصل مهم این مسیر است.
🐍 آموزش جامع Python

🔥 یه کانال پیدا کردم که از بچه‌های همین کانال خودمونه! 😎

داخلش هم چیزای خوب و کاربردی پیدا میشه، هم یه سری چیزای عجیب و باحال 🤯

مثلاً تو یکی از پست‌هاش نوشته: «آموزش داشتن دو تا سیستم‌عامل برای کامپیوتر» 😳
مگه همچین چیزی داریم؟! 😅

شاید داشته باشیم… 😏
خواستید یه سر بزنید، ضرر نمی‌کنید 👇
https://t.me/whyitinfo

هر چی تو عمرم ندیده بودم اینجا دیدم 😁
🐍 آموزش جامع Python

🐍 آموزش جامع Python — پست ۲۴۱
ساخت تست‌های یکپارچه برای Repository در SQLite
در پست‌های قبلی یاد گرفتیم چطور با Repository Pattern عملیات دیتابیس را از منطق اصلی برنامه جدا کنیم و با unittest تست بنویسیم.
حالا می‌خواهیم بررسی کنیم که آیا چند بخش برنامه، وقتی با هم کار می‌کنند، درست عمل می‌کنند یا نه. به این روش Integration Testing می‌گوییم.
🔹 تفاوت تست واحد و تست یکپارچه
Unit Test: یک بخش کوچک از برنامه را به‌صورت مستقل بررسی می‌کند.
Integration Test: تعامل چند بخش را با یکدیگر بررسی می‌کند؛ مثلاً Repository و دیتابیس SQLite.
🔹 مثال عملی
فرض کنیم فایل user_repository.py شامل کلاس زیر است:

class UserRepository:

    def __init__(self, connection):
        self.connection = connection

    def create_table(self):
        self.connection.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL
            )
        """)

    def add_user(self, name):
        cursor = self.connection.execute(
            "INSERT INTO users (name) VALUES (?)",
            (name,)
        )
        return cursor.lastrowid

    def get_by_id(self, user_id):
        return self.connection.execute(
            "SELECT id, name FROM users WHERE id = ?",
            (user_id,)
        ).fetchone()

حالا یک فایل به نام test_user_repository.py بساز:

import sqlite3
import unittest

from user_repository import UserRepository


class TestUserRepository(unittest.TestCase):

    def setUp(self):
        self.connection = sqlite3.connect(":memory:")
        self.connection.row_factory = sqlite3.Row

        self.repository = UserRepository(self.connection)
        self.repository.create_table()

    def tearDown(self):
        self.connection.close()

    def test_add_and_get_user(self):
        user_id = self.repository.add_user("Ali")

        user = self.repository.get_by_id(user_id)

        self.assertIsNotNone(user)
        self.assertEqual(user["id"], user_id)
        self.assertEqual(user["name"], "Ali")


if __name__ == "__main__":
    unittest.main()

🔹 این تست چه کاری انجام می‌دهد؟
۱. یک دیتابیس موقت در حافظه ایجاد می‌کند.
۲. جدول کاربران را می‌سازد.
۳. کاربر جدیدی اضافه می‌کند.
۴. همان کاربر را با شناسه‌اش می‌خواند.
۵. بررسی می‌کند که شناسه و نام بازیابی‌شده صحیح باشند.
۶. پس از پایان تست، اتصال را می‌بندد.
استفاده از :memory: باعث می‌شود تست به دیتابیس واقعی برنامه وابسته نباشد و اطلاعات آزمایشی در فایل اصلی ذخیره نشوند.
🔹 اجرای تست
در ترمینال، از پوشه پروژه اجرا کن:

python -m unittest -v

اگر تست موفق باشد، نتیجه‌ای مشابه زیر مشاهده می‌کنی:

test_add_and_get_user ... ok

Ran 1 test
OK

⚠️ نکته مهم
این تست فقط تعامل Repository با SQLite را بررسی می‌کند. اگر بعداً منطق اصلی برنامه، اعتبارسنجی یا مدیریت خطاها را نیز اضافه کردی، بهتر است برای آن‌ها تست‌های جداگانه بنویسی.
🧠 خلاصه
در تست یکپارچه، به‌جای بررسی صرف یک تابع، چند بخش واقعی برنامه را کنار هم آزمایش می‌کنیم.
🎯 تمرین
یک تست جدید به نام test_missing_user بنویس که شناسه‌ای ناموجود را به get_by_id() بدهد و بررسی کند که نتیجه None است.
پست ۲۴۱ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۴۲
استفاده از unittest.mock برای شبیه‌سازی وابستگی‌ها
در پست قبل تست یکپارچه برای Repository و SQLite نوشتیم.
اما گاهی می‌خواهیم یک بخش از برنامه را بدون اجرای وابستگی‌های واقعی آن آزمایش کنیم. برای این کار می‌توانیم از ماژول استاندارد unittest.mock استفاده کنیم.
به این روش، شبیه‌سازی وابستگی‌ها یا Mocking می‌گوییم.
🔹 چرا Mock لازم است؟
فرض کنید تابعی داریم که برای دریافت اطلاعات به یک سرویس خارجی متصل می‌شود. هنگام تست، شاید نخواهیم واقعاً به آن سرویس متصل شویم.
با Mock می‌توانیم نتیجه مورد انتظار را شبیه‌سازی کنیم و رفتار تابع خودمان را مستقل آزمایش کنیم.
🔹 اولین Mock با Mock

from unittest.mock import Mock

service = Mock()

service.get_username.return_value = "Ali"

result = service.get_username()

print(result)

خروجی:

Ali

در این مثال:
Mock() یک شیء شبیه‌سازی‌شده می‌سازد.
return_value نتیجه‌ای را تعیین می‌کند که فراخوانی متد برمی‌گرداند.
بدون نیاز به پیاده‌سازی واقعی سرویس، می‌توانیم رفتار آن را آزمایش کنیم.
🔹 تست با unittest
فرض کنیم تابع زیر در فایل user_service.py قرار دارد:

def get_display_name(repository, user_id):
    user = repository.get_by_id(user_id)

    if user is None:
        return "Unknown user"

    return user["name"]

این تابع به یک Repository وابسته است، اما برای تست آن نیازی به دیتابیس واقعی نداریم.
فایل test_user_service.py:

import unittest
from unittest.mock import Mock

from user_service import get_display_name


class TestUserService(unittest.TestCase):

    def test_existing_user(self):
        repository = Mock()
        repository.get_by_id.return_value = {
            "id": 1,
            "name": "Ali"
        }

        result = get_display_name(repository, 1)

        self.assertEqual(result, "Ali")
        repository.get_by_id.assert_called_once_with(1)

    def test_missing_user(self):
        repository = Mock()
        repository.get_by_id.return_value = None

        result = get_display_name(repository, 99)

        self.assertEqual(result, "Unknown user")


if __name__ == "__main__":
    unittest.main()

🔹 متد assert_called_once_with()
این متد بررسی می‌کند که یک متد دقیقاً یک بار و با آرگومان‌های مشخص فراخوانی شده باشد:

repository.get_by_id.assert_called_once_with(1)

اگر متد اصلاً فراخوانی نشده باشد، چند بار اجرا شده باشد یا آرگومان دیگری دریافت کرده باشد، تست شکست می‌خورد.
⚠️ نکته مهم
Mock جایگزین تمام تست‌های واقعی نیست. در این مثال بررسی می‌کنیم که منطق get_display_name() درست کار می‌کند؛ اما صحت Queryهای SQLite را بررسی نمی‌کنیم.
بنابراین:
برای منطق مستقل، از Unit Test و Mock استفاده کن.
برای بررسی تعامل واقعی با SQLite، از Integration Test استفاده کن.
🧠 خلاصه
Mock به ما اجازه می‌دهد وابستگی‌های یک تابع را شبیه‌سازی کنیم و رفتار خود تابع را بدون اجرای واقعی آن وابستگی‌ها آزمایش کنیم.
🎯 تمرین
تابعی به نام delete_user_if_exists(repository, user_id) بنویس که:
با repository.get_by_id(user_id) وجود کاربر را بررسی کند.
اگر کاربر وجود داشت، repository.delete_user(user_id) را فراخوانی کند و True برگرداند.
اگر کاربر وجود نداشت، هیچ حذفی انجام ندهد و False برگرداند.
سپس با Mock برای هر دو حالت تست بنویس.
پست ۲۴۲ از دوره جامع Python 🐍

🐍 آموزش جامع Python — پست ۲۴۳
تست‌کردن خطاها با assertRaises
در پست‌های قبلی با Exceptionهای اختصاصی و تست‌نویسی با unittest آشنا شدیم.
اما تست موفقیت‌آمیز بودن یک عملیات کافی نیست؛ گاهی باید مطمئن شویم برنامه هنگام دریافت ورودی نامعتبر، خطای درست را ایجاد می‌کند.
برای این کار از assertRaises() استفاده می‌کنیم.
🔹 مثال ساده

import unittest


def divide(a, b):
    if b == 0:
        raise ValueError("Divisor cannot be zero")

    return a / b


class TestDivide(unittest.TestCase):

    def test_divide_by_zero(self):
        with self.assertRaises(ValueError):
            divide(10, 0)


if __name__ == "__main__":
    unittest.main()

در این مثال، تست زمانی موفق است که فراخوانی divide(10, 0) خطای ValueError ایجاد کند.
اگر تابع هیچ خطایی ایجاد نکند، تست شکست می‌خورد.
🔹 بررسی متن خطا
گاهی فقط نوع خطا کافی نیست و می‌خواهیم پیام آن را هم بررسی کنیم:

def test_error_message(self):
    with self.assertRaises(ValueError) as context:
        divide(10, 0)

    self.assertEqual(
        str(context.exception),
        "Divisor cannot be zero"
    )

متغیر context اطلاعات Exception رخ‌داده را در اختیارمان قرار می‌دهد.
ویژگی context.exception خود شیء خطاست و با str() می‌توانیم پیام آن را بخوانیم.
🔹 کاربرد در پروژه SQLite
فرض کنیم تابعی برای اعتبارسنجی سن کاربران داریم:

def validate_age(age):
    if not isinstance(age, int) or isinstance(age, bool):
        raise TypeError("Age must be an integer")

    if not 1 <= age <= 120:
        raise ValueError("Age must be between 1 and 120")

    return age

حالا می‌توانیم حالت‌های نامعتبر را تست کنیم:

class TestAgeValidation(unittest.TestCase):

    def test_invalid_type(self):
        with self.assertRaises(TypeError):
            validate_age("twenty")

    def test_invalid_range(self):
        with self.assertRaises(ValueError):
            validate_age(150)

    def test_boolean_is_rejected(self):
        with self.assertRaises(TypeError):
            validate_age(True)

در این تست‌ها بررسی می‌کنیم که ورودی نامعتبر با نوع خطای مورد انتظار رد شود.
🔹 تفاوت assertRaises و try/except
try/except برای مدیریت خطا در اجرای عادی برنامه استفاده می‌شود.
assertRaises بررسی می‌کند که هنگام تست، یک عملیات خطای مشخصی ایجاد کند.
⚠️ نکته مهم
نوع خطای مورد انتظار را دقیق انتخاب کن. اگر انتظار ValueError داری اما برنامه به‌جای آن TypeError ایجاد کند، تست نباید موفق محسوب شود.
همچنین فقط همان دستوری را داخل بلوک assertRaises قرار بده که انتظار داری خطا ایجاد کند؛ در غیر این صورت ممکن است خطای بخش دیگری از کد باعث قبولی اشتباه تست شود.
🧠 خلاصه

with self.assertRaises(ValueError):
    some_function()

این الگو برای تست اعتبارسنجی ورودی‌ها، خطاهای اختصاصی و رفتارهای نامعتبر در پروژه‌های Python کاربرد زیادی دارد.
🎯 تمرین
برای تابع validate_age() سه تست بنویس:
۱. سن 25 باید پذیرفته شود.
۲. سن 0 باید ValueError ایجاد کند.
۳. مقدار "25" باید TypeError ایجاد کند.
پست ۲۴۳ از دوره جامع Python 🐍