عمليات CRUD

عمليات CRUD — CRUD Operations

CRUD هي اختصار للعمليات الأربع الأساسية في أي تطبيق يتعامل مع بيانات:

الحرف	الاسم	SQL	المعنى
C	Create	`INSERT`	إنشاء سجل جديد
R	Read	`SELECT`	قراءة سجل أو أكثر
U	Update	`UPDATE`	تعديل سجل موجود
D	Delete	`DELETE`	حذف سجل

في هذا الدرس نبني طبقة CRUD كاملة لجدول contacts (جهات الاتصال). كل خطوة هي تحدٍّ تطبّق فيه ما تعلمته.

بنية المشروع

سنبني وظيفة create_table أولاً، ثم وظائف CRUD بالترتيب. الهدف النهائي: نظام بسيط لإدارة جهات الاتصال.

contacts
├── id      INTEGER PRIMARY KEY AUTOINCREMENT
├── name    TEXT NOT NULL
├── phone   TEXT
└── email   TEXT

الخطوة 0 — إعداد الاتصال المشترك

في التطبيقات الحقيقية تُنشئ الاتصال مرة واحدة وتُمرّره للوظائف. هذا النمط يُوضّح المسؤوليات ويُسهّل الاختبار.

main.go

import sqlite3

def create_connection():
    """إنشاء اتصال بقاعدة البيانات — Create database connection"""
    conn = sqlite3.connect(':memory:')
    conn.row_factory = sqlite3.Row  # الوصول بالاسم — access by column name
    return conn

def create_table(conn):
    """إنشاء جدول جهات الاتصال — Create contacts table"""
    conn.execute('''
        CREATE TABLE IF NOT EXISTS contacts (
            id    INTEGER PRIMARY KEY AUTOINCREMENT,
            name  TEXT    NOT NULL,
            phone TEXT,
            email TEXT
        )
    ''')
    conn.commit()

# اختبار الإعداد — Test setup
conn = create_connection()
create_table(conn)
print("الاتصال يعمل")
print("الجدول: contacts (id, name, phone, email)")
conn.close()

Output:

لاحظ CREATE TABLE IF NOT EXISTS — هذا التعبير يُنشئ الجدول فقط إن لم يكن موجوداً، مما يجعل الكود آمناً للتشغيل المتكرر.

الخطوة 1 — Create (الإدراج)

تحدي — Challenge

import sqlite3

def create_connection():
    conn = sqlite3.connect(':memory:')
    conn.row_factory = sqlite3.Row
    return conn

def create_table(conn):
    conn.execute('''
        CREATE TABLE IF NOT EXISTS contacts (
            id    INTEGER PRIMARY KEY AUTOINCREMENT,
            name  TEXT NOT NULL,
            phone TEXT,
            email TEXT
        )
    ''')
    conn.commit()

def create_contact(conn, name, phone=None, email=None):
    """إدراج جهة اتصال — Insert a contact"""
    # استخدم معاملات موضعية دائماً — always use parameterized queries
    cursor = conn.execute(
        "INSERT INTO contacts (name, phone, email) VALUES (?, ?, ?)",
        (name, phone, email)
    )
    conn.commit()
    return cursor.lastrowid  # معرّف السجل الجديد — new record ID

conn = create_connection()
create_table(conn)

# أدرج جهتَي اتصال واطبع معرّفيهما
id1 = create_contact(conn, "أحمد العلي", "0501234567", "ahmed@example.com")
id2 = create_contact(conn, "فاطمة الزهراء", "0559876543")

print(f"تم إدراج جهة اتصال برقم: {id1}")
print(f"تم إدراج جهة اتصال برقم: {id2}")

conn.close()

cursor.lastrowid يُعيد الـID الذي خصّصته SQLite للسجل المُدرج للتو. هذا مفيد لربط السجل بكيانات أخرى أو إعادته للمستخدم.

الخطوة 2 — Read (القراءة الجماعية والفردية)

القراءة لها وجهان: جلب جميع السجلات، وجلب سجل واحد بمعرّفه.

تحدي — Challenge

import sqlite3

def create_connection():
    conn = sqlite3.connect(':memory:')
    conn.row_factory = sqlite3.Row
    return conn

def setup(conn):
    conn.execute('''
        CREATE TABLE IF NOT EXISTS contacts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL, phone TEXT, email TEXT
        )
    ''')
    conn.executemany(
        "INSERT INTO contacts (name, phone, email) VALUES (?, ?, ?)",
        [
            ("أحمد العلي",    "0501234567", "ahmed@example.com"),
            ("فاطمة الزهراء", "0559876543", None),
            ("خالد المطيري",  "0521111222", "khalid@example.com"),
        ]
    )
    conn.commit()

def get_all_contacts(conn):
    """جلب جميع جهات الاتصال — Fetch all contacts"""
    cursor = conn.execute("SELECT id, name, phone, email FROM contacts ORDER BY id")
    return cursor.fetchall()

def get_contact_by_id(conn, contact_id):
    """جلب جهة اتصال بالمعرّف — Fetch contact by ID"""
    cursor = conn.execute(
        "SELECT id, name, phone, email FROM contacts WHERE id = ?",
        (contact_id,)
    )
    return cursor.fetchone()  # None إن لم يوجد — None if not found

conn = create_connection()
setup(conn)

# اطبع جميع جهات الاتصال
all_contacts = get_all_contacts(conn)
print("جميع جهات الاتصال:")
for c in all_contacts:
    print(f"  {c['id']}: {c['name']} ({c['phone']})")

# اجلب جهة اتصال محددة
contact = get_contact_by_id(conn, 2)
print(f"جهة الاتصال 2: {contact['name']}")

conn.close()

الخطوة 3 — Update (التحديث)

UPDATE يُعدّل سجلاً موجوداً. ضروري استخدام WHERE id = ? لتحديد السجل المستهدف — بدونه تُعدّل جميع السجلات في الجدول.

main.go

import sqlite3

def create_connection():
    conn = sqlite3.connect(':memory:')
    conn.row_factory = sqlite3.Row
    return conn

def setup(conn):
    conn.execute('''
        CREATE TABLE IF NOT EXISTS contacts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL, phone TEXT, email TEXT
        )
    ''')
    conn.executemany(
        "INSERT INTO contacts (name, phone) VALUES (?, ?)",
        [("أحمد", "0501234567"), ("فاطمة", "0559876543")]
    )
    conn.commit()

def update_contact(conn, contact_id, name=None, phone=None, email=None):
    """تحديث جهة اتصال — Update a contact"""
    # نجمع الحقول المُراد تحديثها فقط — only update provided fields
    fields = []
    values = []
    if name is not None:
        fields.append("name = ?")
        values.append(name)
    if phone is not None:
        fields.append("phone = ?")
        values.append(phone)
    if email is not None:
        fields.append("email = ?")
        values.append(email)

    if not fields:
        return 0  # لا شيء للتحديث — nothing to update

    values.append(contact_id)  # للـ WHERE — for WHERE clause
    sql = f"UPDATE contacts SET {', '.join(fields)} WHERE id = ?"
    cursor = conn.execute(sql, values)
    conn.commit()
    return cursor.rowcount  # عدد الصفوف المتأثرة — rows affected

conn = create_connection()
setup(conn)

# قبل التحديث — before update
row = conn.execute("SELECT name, phone FROM contacts WHERE id = 1").fetchone()
print(f"قبل: {row['name']} | {row['phone']}")

# تحديث الهاتف فقط — update phone only
affected = update_contact(conn, 1, phone="0509999999")
print(f"صفوف متأثرة: {affected}")

# بعد التحديث — after update
row = conn.execute("SELECT name, phone FROM contacts WHERE id = 1").fetchone()
print(f"بعد: {row['name']} | {row['phone']}")

conn.close()

Output:

الخطوة 4 — Delete (الحذف)

تحدي — Challenge

import sqlite3

def create_connection():
    conn = sqlite3.connect(':memory:')
    conn.row_factory = sqlite3.Row
    return conn

def setup(conn):
    conn.execute('''
        CREATE TABLE IF NOT EXISTS contacts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL, phone TEXT, email TEXT
        )
    ''')
    conn.executemany(
        "INSERT INTO contacts (name, phone) VALUES (?, ?)",
        [("أحمد", "050"), ("فاطمة", "055"), ("خالد", "052")]
    )
    conn.commit()

def delete_contact(conn, contact_id):
    """حذف جهة اتصال بالمعرّف — Delete contact by ID"""
    cursor = conn.execute(
        "DELETE FROM contacts WHERE id = ?",
        (contact_id,)
    )
    conn.commit()
    return cursor.rowcount  # 1 إن نجح، 0 إن لم يوجد — 1 if found, 0 if not

def count_contacts(conn):
    """عدد جهات الاتصال — Count contacts"""
    row = conn.execute("SELECT COUNT(*) FROM contacts").fetchone()
    return row[0]

conn = create_connection()
setup(conn)

print(f"قبل الحذف: {count_contacts(conn)} جهات اتصال")

# احذف فاطمة (ID: 2)
delete_contact(conn, 2)

print(f"بعد الحذف: {count_contacts(conn)} جهات اتصال")

# اطبع المتبقية
rows = conn.execute("SELECT name FROM contacts ORDER BY id").fetchall()
print("المتبقية:")
for r in rows:
    print(f"  {r['name']}")

conn.close()

الخطوة 5 — الطبقة الكاملة

الآن نجمع كل شيء في واجهة موحدة. هذا النمط يُشبه Repository Pattern الذي ستتعلمه في درس SQLAlchemy:

main.go

import sqlite3

class ContactsDB:
    """طبقة CRUD لجهات الاتصال — CRUD layer for contacts"""

    def __init__(self):
        self.conn = sqlite3.connect(':memory:')
        self.conn.row_factory = sqlite3.Row
        self._create_table()

    def _create_table(self):
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS contacts (
                id    INTEGER PRIMARY KEY AUTOINCREMENT,
                name  TEXT NOT NULL,
                phone TEXT,
                email TEXT
            )
        ''')
        self.conn.commit()

    def create(self, name, phone=None, email=None):
        """إدراج — Create"""
        cur = self.conn.execute(
            "INSERT INTO contacts (name, phone, email) VALUES (?, ?, ?)",
            (name, phone, email)
        )
        self.conn.commit()
        return cur.lastrowid

    def get_all(self):
        """قراءة الكل — Read all"""
        return self.conn.execute(
            "SELECT id, name, phone, email FROM contacts ORDER BY id"
        ).fetchall()

    def get_by_id(self, contact_id):
        """قراءة واحد — Read one"""
        return self.conn.execute(
            "SELECT id, name, phone, email FROM contacts WHERE id = ?",
            (contact_id,)
        ).fetchone()

    def update(self, contact_id, name):
        """تحديث الاسم — Update name"""
        cur = self.conn.execute(
            "UPDATE contacts SET name = ? WHERE id = ?",
            (name, contact_id)
        )
        self.conn.commit()
        return cur.rowcount

    def delete(self, contact_id):
        """حذف — Delete"""
        cur = self.conn.execute(
            "DELETE FROM contacts WHERE id = ?",
            (contact_id,)
        )
        self.conn.commit()
        return cur.rowcount

    def close(self):
        self.conn.close()


# تجربة الطبقة الكاملة — Test the full layer
db = ContactsDB()

# C — إنشاء
id1 = db.create("أحمد", "0501234567", "ahmed@test.com")
id2 = db.create("سلمى", "0559876543")
id3 = db.create("ماجد", "0521234567", "majed@test.com")
print(f"أُدرج 3 جهات اتصال (IDs: {id1}, {id2}, {id3})")

# R — قراءة
all_c = db.get_all()
print(f"\nالكل ({len(all_c)} جهات):")
for c in all_c:
    print(f"  #{c['id']}: {c['name']}")

# R — قراءة واحد
one = db.get_by_id(2)
print(f"\nجهة #2: {one['name']} | {one['phone']}")

# U — تحديث
db.update(2, "سلمى الأحمد")
one = db.get_by_id(2)
print(f"\nبعد التحديث: {one['name']}")

# D — حذف
db.delete(3)
all_c = db.get_all()
print(f"\nبعد الحذف: {len(all_c)} جهات اتصال")

db.close()

Output:

نقاط مهمة تذكّرها

1. conn.commit() بعد كل تعديل: بدون commit التغييرات مؤقتة في الذاكرة ولا تُحفظ على القرص (أو في :memory: تُفقد عند إغلاق الاتصال قبل الأوان).

2. ORDER BY في كل SELECT: النتائج بدون ORDER BY لها ترتيب غير محدد — صحيح في السيناريو الحالي لكن خطر في التطبيقات الحقيقية.

3. rowcount للتحقق: بعد UPDATE أو DELETE تحقق من cursor.rowcount. إذا كانت 0 فالسجل لم يُوجد، وهذا يساعدك تُعيد خطأ واضحاً للمستخدم.

4. المعاملات الموضعية أبداً: كل ? تحمي من SQL Injection. لا استثناء حتى لو كانت القيمة رقماً أو قيمة داخلية.

تحدي — Challenge

import sqlite3

# أنشئ ContactsDB بسيطة وجرّب:
# 1. أدرج 4 جهات اتصال: سعد/ليلى/طارق/هند
# 2. اطبع عددها
# 3. حدّث اسم ليلى (ID=2) إلى "ليلى محمد"
# 4. احذف هند (ID=4)
# 5. اطبع العدد النهائي

conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
conn.execute('''
    CREATE TABLE contacts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL
    )
''')
conn.executemany(
    "INSERT INTO contacts (name) VALUES (?)",
    [("سعد",), ("ليلى",), ("طارق",), ("هند",)]
)
conn.commit()

# 1. اطبع الإجمالي — print total
total = conn.execute("SELECT COUNT(*) FROM contacts").fetchone()[0]
print(f"الإجمالي: {total}")

# 2. حدّث ليلى — update ليلى
conn.execute("UPDATE contacts SET name = ? WHERE id = ?", ("ليلى محمد", 2))
conn.commit()
row = conn.execute("SELECT name FROM contacts WHERE id = 2").fetchone()
print(f"بعد التحديث: {row['name']}")

# 3. احذف هند — delete هند
conn.execute("DELETE FROM contacts WHERE id = ?", (4,))
conn.commit()
remaining = conn.execute("SELECT COUNT(*) FROM contacts").fetchone()[0]
print(f"بعد الحذف: {remaining}")

conn.close()