sqlite3 در پایتون

۱. اتصال و مکان‌نما

'''Opening a connection to a SQLite database.

sqlite3 is part of the standard library - there is nothing to install.
connect() CREATES the file when it does not exist yet.
'''

import sqlite3


def main() -> None:
    # The file is created next to the program, in the current folder.
    conn = sqlite3.connect('database.db')

    # SQLite ignores the foreign keys unless we switch them on,
    # and it forgets the setting at every new connection.
    conn.execute('PRAGMA foreign_keys = ON')

    # The cursor is the object that runs the queries and holds the results.
    cursor = conn.cursor()

    # A first query to prove that everything works.
    cursor.execute('SELECT sqlite_version()')
    print('SQLite version:', cursor.fetchone()[0])

    # Always close what you opened.
    conn.close()

    # ':memory:' creates a database that only lives in the RAM - perfect
    # for trying things out.
    temp = sqlite3.connect(':memory:')
    temp.close()

اتصال: نمونه 01_connect.py

۲. ساختن ساختار جدول‌ها

SCHEMA = '''
    CREATE TABLE IF NOT EXISTS students (
        student_id INTEGER PRIMARY KEY AUTOINCREMENT,
        first_name TEXT NOT NULL,
        last_name  TEXT NOT NULL,
        email      TEXT UNIQUE NOT NULL
    );

    CREATE TABLE IF NOT EXISTS courses (
        course_id  INTEGER PRIMARY KEY AUTOINCREMENT,
        title      TEXT NOT NULL,
        teacher_id INTEGER,
        credit     INTEGER DEFAULT 3,
        FOREIGN KEY (teacher_id) REFERENCES teachers(teacher_id)
            ON DELETE SET NULL ON UPDATE CASCADE
    );

    CREATE TABLE IF NOT EXISTS enrollments (
        enrollment_id   INTEGER PRIMARY KEY AUTOINCREMENT,
        student_id      INTEGER NOT NULL,
        course_id       INTEGER NOT NULL,
        grade           REAL,
        enrollment_date TEXT DEFAULT (date('now')),
        FOREIGN KEY (student_id) REFERENCES students(student_id)
            ON DELETE CASCADE ON UPDATE CASCADE,
        UNIQUE (student_id, course_id)
    );
'''


def main() -> None:
    conn = sqlite3.connect('database.db')
    conn.execute('PRAGMA foreign_keys = ON')
    cursor = conn.cursor()

    cursor.executescript(SCHEMA)

    # Nothing is really written until commit() is called.
    conn.commit()
    conn.close()

ساختار: نمونه 02_create_tables.py

executescript چند دستور جداشده با نقطه‌ویرگول را با هم اجرا می‌کند و برای ساختن ساختار ایده‌آل است. برای پرس‌وجوهای عادی execute را به کار ببرید. عبارت IF NOT EXISTS باعث می‌شود اجرای دوباره برنامه خطا ندهد.

۳. درج داده با پارامتر

'''NEVER build the SQL with an f-string or with '+': always use '?' parameters.
The values are then sent apart from the query and can never become SQL code.
'''

# executemany() runs the same query once per tuple of the list.
students_data = [
    ('Ali', 'Rezaei', 'alireza@kadoosedu.ir'),
    ('Maryam', 'Ahmadi', 'maryam@kadoosedu.ir'),
    ('Sara', 'Karimi', 'sara@kadoosedu.ir'),
]
cursor.executemany(
    'INSERT OR IGNORE INTO students (first_name, last_name, email) '
    'VALUES (?, ?, ?)',
    students_data,
)
conn.commit()

# We must not guess the ids: we ask the database for them.
# NOTE: the comma in ('Mohammadi',) is what makes it a TUPLE.
cursor.execute('SELECT teacher_id FROM teachers WHERE last_name = ?',
               ('Mohammadi',))
teacher1 = cursor.fetchone()[0]

# A single insert, and the id it produced.
cursor.execute(
    'INSERT OR IGNORE INTO students (first_name, last_name, email) '
    'VALUES (?, ?, ?)',
    ('Omid', 'Shabani', 'omid@kadoosedu.ir'),
)
print('The new student got the id', cursor.lastrowid)
conn.commit()

درج: نمونه 03_insert_data.py

شرح مثال

• علامت سؤال جای مقدارهاست و مقدارها در یک تاپل به عنوان آرگومان دوم می‌آیند.
• دام کلاسیک: ('Mohammadi',) با ویرگول انتهایی. بدون آن یک رشته ساده است نه تاپل، و خطا می‌گیرید. به خاطر دارید از فصل چهاردهم.
• executemany همان پرس‌وجو را برای هر تاپل لیست اجرا می‌کند و بسیار سریع‌تر از حلقه است.
• lastrowid شناسه‌ای را می‌دهد که پایگاه داده در همین مرحله تولید کرده است. برای درج والد و بعد فرزندانش حیاتی است.
• INSERT OR IGNORE اگر محدودیت یکتایی نقض شود، بی‌سروصدا رد می‌شود. برای داده نمونه خوب است، ولی در برنامه واقعی معمولاً می‌خواهید خطا را ببینید.
• شناسه‌ها را حدس نزنید. آنها را از پایگاه داده بپرسید، وگرنه کدتان به ترتیب درج وابسته می‌شود.

۴. خواندن داده

'''
    fetchone()   -> one row, or None
    fetchall()   -> every remaining row, in a list
    fetchmany(n) -> n rows
    looping over the cursor -> one row at a time, without loading everything
'''

def show_table(table_name: str) -> None:
    '''Print the content of one table.

    The table NAME cannot be a '?' parameter (parameters are only for values),
    so this function must only ever be called with a name WE choose, never with
    something typed by a user.
    '''
    conn = sqlite3.connect('database.db')
    cursor = conn.cursor()

    cursor.execute(f'SELECT * FROM {table_name}')

    # cursor.description holds the column names.
    columns = [column[0] for column in cursor.description]
    print(' | '.join(columns))

    for row in cursor:               # looping is the memory-friendly way
        print(row)

    conn.close()

خواندن: نمونه 04_select_all.py

۵. تغییر و حذف

'''cursor.rowcount tells us how many rows were really affected - use it to know
whether the WHERE matched anything.
'''

cursor.execute(
    'UPDATE students SET email = ? WHERE email = ?',
    ('new_alireza@kadoosedu.ir', 'alireza@kadoosedu.ir'),
)
print('rows updated:', cursor.rowcount)
conn.commit()

cursor.execute(
    'DELETE FROM enrollments WHERE grade < ?',
    (16,),
)
print('rows deleted:', cursor.rowcount)
conn.commit()

تغییر و حذف: نمونه 05_update_delete.py

صفت rowcount می‌گوید چند سطر واقعاً تحت تأثیر قرار گرفته‌اند. اگر صفر بود یعنی شرط شما با هیچ سطری جور در نیامده است، و این معمولاً چیزی است که باید به کاربر بگویید.

۶. پیوند از داخل پایتون

QUERY = '''
    SELECT s.first_name || ' ' || s.last_name AS student,
           c.title                            AS course,
           e.grade                            AS grade
    FROM enrollments e
    JOIN students s ON e.student_id = s.student_id
    JOIN courses  c ON e.course_id  = c.course_id
    ORDER BY student, course
'''

پرس‌وجوی پیوندی: نمونه 06_join_query.py

نکته مهم این است که هیچ چیز تازه‌ای نیست. SQL دقیقاً همان SQL فصل قبل است؛ فقط شکل اجرایش عوض شده. رشته سه‌گیومه‌ای، یک پرس‌وجوی بلند را خوانا نگه می‌دارد.

۷. پیدا کردن یک رکورد

'''fetchone() returns ONE row, or None when nothing matched.

Forgetting to test for None is the most common bug of this session:
"TypeError: 'NoneType' object is not subscriptable".
'''

def find_student(email: str) -> tuple | None:
    'Return the row of a student, or None.'
    conn = sqlite3.connect('database.db')
    cursor = conn.cursor()

    cursor.execute('SELECT * FROM students WHERE email = ?', (email,))
    student = cursor.fetchone()

    conn.close()
    return student


def main() -> None:
    for email in ('zahra@kadoosedu.ir', 'nobody@kadoosedu.ir'):
        student = find_student(email)

        if student:
            print('Student found:', student)
        else:
            print(f'No student with the email {email}.')

جست‌وجوی یک رکورد: نمونه 08_find_one.py

این پرتکرارترین باگ کل فصل است: fetchone() اگر چیزی پیدا نکند None می‌دهد، و بعد student[0] برنامه را با TypeError می‌کشد. همیشه بررسی کنید.

۸. تراکنش

تراکنش مجموعه‌ای از عملیات است که باید به صورت یک واحد انجام شوند؛ یا همه عملیات با موفقیت انجام می‌شوند یا هیچ‌کدام نباید اثر دائمی داشته باشند. مثال کلاسیک آن انتقال وجه است: اگر مبلغ از یک حساب کم شود اما به حساب مقصد اضافه نشود، یکپارچگی داده‌ها از بین می‌رود.

try:
    cursor.execute(
        'INSERT INTO students (first_name, last_name, email) '
        'VALUES (?, ?, ?)',
        ('Sara', 'Hosseini', 'sara2@kadoosedu.ir'),
    )
    new_student_id = cursor.lastrowid

    cursor.execute('SELECT course_id FROM courses WHERE title = ?',
                   ('Python Programming',))
    row = cursor.fetchone()
    if row is None:
        raise sqlite3.Error('the course does not exist')

    cursor.execute(
        'INSERT INTO enrollments (student_id, course_id, grade) '
        'VALUES (?, ?, ?)',
        (new_student_id, row[0], 20),
    )

    # commit() makes everything permanent, in one single step.
    conn.commit()
    print('Transaction done.')

except sqlite3.Error as e:
    # rollback() undoes everything since the last commit.
    conn.rollback()
    print('Rollback happened:', e)

finally:
    conn.close()

تراکنش: نمونه 09_transaction.py

شرح مثال

• ساختار همیشه یکی است: try با چند دستور و یک commit در انتها، except با rollback، و finally با close.
• rollback() همه چیز را به حالت آخرین commit برمی‌گرداند.
• نکته ظریف: with sqlite3.connect(...) در موفقیت commit و در خطا rollback می‌کند، ولی اتصال را نمی‌بندد. این برخلاف انتظار است و باید به خاطر داشته باشید.
• در فصل سی و پنجم می‌بینید که نبود تراکنش در نسخه اصلی برنامه کتابخانه، باعث می‌شد کتاب همزمان هم امانت داده شود و هم موجود بماند.

۹. پشتیبان‌گیری

'''conn.backup() copies a database safely, even while it is being used -
much better than copying the file with the operating system.
'''

def main() -> None:
    source = sqlite3.connect('database.db')
    backup = sqlite3.connect('backup.db')

    try:
        source.backup(backup)
        print('backup.db was written.')
    finally:
        # The connections are closed even if the backup failed.
        backup.close()
        source.close()

پشتیبان‌گیری: نمونه 10_backup.py

کپی کردن فایل پایگاه داده با دستور سیستم‌عامل خطرناک است: اگر همان لحظه برنامه‌ای در حال نوشتن باشد، کپی ناقص و خراب می‌شود. متد backup این را درست انجام می‌دهد.

۱۰. سطر به شکل دیکشنری

'''By default a row is a TUPLE, so we read it with row[0], row[1]...
That is short but unreadable, and it breaks when a column is added.

sqlite3.Row lets us read a row by COLUMN NAME.
'''

# Default: tuples.
cursor.execute('SELECT student_id, first_name, email FROM students')
for row in cursor.fetchall()[:3]:
    print(row, '->', row[1])

# With sqlite3.Row: access by name (and still by index).
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('SELECT student_id, first_name, email FROM students')
for row in cursor.fetchall()[:3]:
    print(row['first_name'], row['email'], '| id =', row['student_id'])

# A row can be turned into a real dictionary, ready for JSON.
cursor.execute('SELECT * FROM students LIMIT 2')
students = [dict(row) for row in cursor.fetchall()]
print(students)

row_factory: نمونه 12_row_factory.py

۱۱. تزریق SQL در عمل

def unsafe_search(cursor: sqlite3.Cursor, name: str) -> list:
    'NEVER DO THIS: the value is glued into the SQL text.'
    query = f"SELECT * FROM students WHERE first_name = '{name}'"
    print('  SQL sent:', query)
    cursor.execute(query)
    return cursor.fetchall()


def safe_search(cursor: sqlite3.Cursor, name: str) -> list:
    'The right way: the value travels apart from the query.'
    cursor.execute('SELECT * FROM students WHERE first_name = ?', (name,))
    return cursor.fetchall()

حمله و دفاع: نمونه 13_sql_injection.py

cursor.execute('... WHERE name = ?' % name)          # wrong
cursor.execute(f'... WHERE name = {name}')           # wrong
cursor.execute('... WHERE name = ?', (name,))        # right

سه شکل، فقط یکی درست

۱۲. برنامه کامل

'''A complete little application on top of sqlite3.'''

import sqlite3

DB_NAME = 'students.db'

SCHEMA = '''
    CREATE TABLE IF NOT EXISTS students (
        student_id INTEGER PRIMARY KEY AUTOINCREMENT,
        first_name TEXT NOT NULL,
        last_name  TEXT NOT NULL,
        email      TEXT UNIQUE NOT NULL
    );
'''


def get_connection() -> sqlite3.Connection:
    '''Return a ready-to-use connection.

    Every function of the program goes through here, so the PRAGMA and the
    row_factory are never forgotten.
    '''
    conn = sqlite3.connect(DB_NAME)
    conn.execute('PRAGMA foreign_keys = ON')
    conn.row_factory = sqlite3.Row
    return conn


def add_student(first_name: str, last_name: str, email: str) -> bool:
    'Insert one student. Return False when the email already exists.'
    try:
        with get_connection() as conn:
            conn.execute(
                'INSERT INTO students (first_name, last_name, email) '
                'VALUES (?, ?, ?)',
                (first_name, last_name, email),
            )
        return True
    except sqlite3.IntegrityError:
        # Raised by the UNIQUE constraint on the email.
        return False


def list_students() -> list[sqlite3.Row]:
    'Return every student, sorted by last name.'
    conn = get_connection()
    try:
        return conn.execute(
            'SELECT * FROM students ORDER BY last_name'
        ).fetchall()
    finally:
        conn.close()

برنامه کامل: نمونه 14_students_app.py

الگوهایی که باید از این برنامه یاد بگیرید

• تابع کمکی اتصال. همه توابع برنامه از یک جا اتصال می‌گیرند، پس PRAGMA و row_factory هرگز فراموش نمی‌شوند. این مهم‌ترین درس این فصل است.
• تابع‌ها بولی برمی‌گردانند، نه پیام. add_student فقط می‌گوید موفق شد یا نه؛ تصمیم درباره پیام با لایه رابط کاربر است. به همین دلیل همین توابع را در فصل سی و پنجم می‌شود پشت یک رابط گرافیکی گذاشت.
• گرفتن IntegrityError. نقض محدودیت یکتایی یک اتفاق عادی است، نه باگ. باید مدیریتش کرد، نه اینکه برنامه بترکد.
• finally: conn.close(). چون with در sqlite3 اتصال را نمی‌بندد، بستنش را خودتان تضمین کنید.

خلاصه فصل