5  إطار بايثون لقواعد البيانات

يقال أن الخيمياء (Alchemy) تسعى لتحويل المعادن الرخيصة (مثل الرصاص) إلى معادن ثمينة (كالذهب).

حزمة SQLAlchemy هي أشهر إطار للتعامل مع قواعد البيانات في بايثون. وهي تحل مجموعة إشكالات استعرضناها في نهاية درس وصل التطبيق بقاعدة البيانات، حيث تبيَّن أن السايق (Driver) psycopg هو فقط واحدٌ من مجموعة أدوات نحتاج إليها للتعامل مع نظم قواعد البيانات بسهولة.

من بين هذه الحلول مثلاً إدارة مجمع اتصالات (Connection Pool) مستمرَّة محفوظة في الذاكرة يتم استعمالها وإعادتها للمجمع وعدم إغلاقها؛ لأن عملية إنشاء اتصال جديد عملية بطيئة في الغاية.

تكتسب الحزمة البرمجية لقب إطار (Framework) عندما تفرض طريقة معيَّنة في التطوير (كتابة النص البرمجي أو ترتيب خطوات للعمل). وهذا التأطير يتم عن طريق تعريف كلمات جديدة للغة، وسياقات توضَع فيها لتنتج معاني محددة بعناية في شبكة مفاهيم هذا الإطار.

تثبيت الحزمة

وطريقة التثبيت في الصفحة الرسمية تدلنا على طريقة تثبيت الحزمة:

uv add SQLAlchemy

المحرك

الجزء المحوري في الإطار هو المحرِّك (Engine) حيث يدير هو آلية إنشاء وحفظ الاتصالات وإعادتها واستعمالها مرة أخرى. وعادةً ما يكون المُحرِّك مُفردًا (Singleton) -أي: كائنًا فريدًا مشاعًا (global)- لكل قاعدة بيانات نريد التواصل معها.

فأولاً يجب إنشاء ملف قيَم متغيرات البيئة، على النحو التالي:

POSTGRES_USER=user123
POSTGRES_PASSWORD=secret123
POSTGRES_HOST=localhost
POSTGRES_DB=testdb

وفيما يلي نُنشئ المحرِّك ونمرر له معلومات الاتصال بعد قراءتها من ملف .env، ونؤكد على ضرورة عدم وضع هذه القيَم مباشرة في النص البرمجي للتطبيق لأنها تتغير بحسب البئية التي يعمل فيها البرنامج:

import os

from dotenv import load_dotenv

from sqlalchemy import URL
from sqlalchemy import create_engine

load_dotenv()  # defaults to loading .env from current directory

engine = create_engine(
    url=URL.create(
        "postgresql+psycopg", # Dialect + Driver
        username=os.getenv("POSTGRES_USER"),
        password=os.getenv("POSTGRES_PASSWORD"),
        host=os.getenv("POSTGRES_HOST"),
        database=os.getenv("POSTGRES_DB"),
    )
)

وننبه على اختلاف يسير هنا وهو أن أول معطى في إنشاء نص الاتصال (Connection String) هو: postgres+psycopg وهو شيئان يصفل بينهما علامة + ولم نرَ هذا الأمر مُسبقًا عند مكتبة psycopg، وإليك معنى كل واحد منهما:

أما postgres فهي اللهجة (Dialect) وتقبل كذلك (mysql أو sqlite أو mssql أو oracle ونحوها). وذلك لضبط آلية تحويل تعبيرات SQL التي سنكتبها في بايثون إلى بحسب نظام قاعدة البيانات المستعمل.

وأما psycopg فهو السائق (Driver) وتقبل أيضًا (mysqlclient أو pysqlite أو asyncpg أو pymysql أو pyodbc أو cx_oracle ونحوها). وذلك لضبط آلية تحويل الأوامِر المتوافقة مع DB-API إلى ما يقابلها من تفاصيل بحسب نظام قاعدة البيانات المستعمل.

وسبب الفصل هو أن النظام الواحد قد يساق بطرقتين؛ فمثلاً: نظام ولجهة postgres قد تساق بالقديم (psycopg2) أو بالجديد -الذي هو الإصدار الثالث، لكن من غير الرقم 3- وهو (psycopg).

تعريف مخطط البيانات (DDL)

تعريف البيانات يكون عن طريق البيانات الوصفية (Metadata)؛ وهو عادةً ما يكون مُفردًا (Singleton) -أي: كائنًا فريدًا مشاعًا-.

ويتم استيراد مفردات التعريف على النحو التالي:

from sqlalchemy.schema import MetaData, Table, Column
from sqlalchemy.types import Integer, String, Date

metadata = MetaData()

نعرف الجدول الأول: جدول المستخدمين:

users_table = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(100)),
    Column("birth_date", Date)
)

ثم سيتم تحويل هذا التعريف لاحقًا إلى هذا النص بلغة SQL:

CREATE TABLE users (
    id   INTEGER NOT NULL,
    name VARCHAR(100),
    birth_date DATE,
    PRIMARY KEY (id)
);

وإليك مرجع جميع أنواع SQL الممثلة في SQLAlchemy لمعرفة ما يقابل كل نوع؛ كمعرفة أن String تحوَّل إلى VARCHAR مثلاً.

القيود

لتحديد القيود (Constraints) مثل أن يوجَد لكل عنوان المستخدم الذي ينسَب إليه هذا العنوان، فإننا نستعمل المفتاح الأجنبي (ForeignKey) في تعريف جدول العناوين بهذا الشكل:

from sqlalchemy.schema import ForeignKey

user_emails_table = Table(
    "user_emails",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("user_id", ForeignKey("users.id"), nullable=False),
    Column("email", String, nullable=False),
)

وإذا نظرنا إلى ترجمتها في SQL فيما بعد فستكون بهذا الشكل:

CREATE TABLE user_emails (
    id      INTEGER NOT NULL,
    user_id INTEGER NOT NULL,
    email   VARCHAR NOT NULL,
    PRIMARY KEY (id),
    FOREIGN KEY(user_id) REFERENCES users (id)
);

تطبيق التعريفات على قاعدة البيانات

لا يتم إنشاء الجداول إلا حين نستعمل طريقة create_all من كائن البيانات الوصفية (Metadata)، وعكسه طريقة drop_all، وكلاهما يتطلَّب المحرِّك:

metadata.create_all(engine)
# metadata.drop_all(engine)

تنبيه: لدينا خلل كبير لم نعالجه بعد. فهذه الطريقة (create_all و drop_all) مدمِّرة للبيانات الموجودة! مما يجعلها صالحة للاستعمال في التجرِبة فقط. أما في تطوير البرمجيات في الواقع فعلينا استعمال آلية دقيقة في تتبع تطوُّر مخطط البيانات بحيث نستطيع التعديل من غير أن نفقد شيئًا من البيانات. وهذه الآلية معروفة باسم: الترحيل (Migrations)، وسيأتي الكلام عنها في الدرس القادم (متابعة تطوارات المخطط).

عكس التعريف الموجود في قاعدة البيانات إلى بايثون

ماذا لو كنت أعمل على مشروع تم تعريف البيانات فيه وأصبحت قاعدة البيانات مخططة، وأريد الآن أن أقرأ هذا التخطيط وأحوِّله إلى بايثون؟ هذه العملية تسمى العكس (Reflection).

العمل مع البيانات

بعد أن رأينا تعريف الجداوِل وإنشاءها في قاعدة البيانات بالفعل. ننتقل إلى البيانات ونرى كيف يمكننا إطار SQLAlchemy من التعامل معها.

أولاً لاحظ أن العبارات تأتي من حزمة واحدة وهكذا تستورد:

from sqlalchemy.sql.expression import insert, select, update, delete

ثم يمكن إنشاء الجُمَل في بايثون على النحو التالي:

stmt = (
    insert(users_table)
    .values(
        name="Adam Banana",
        birth_date=datetime.date(1990, 1, 1)
    )
    .returning(users_table.c.id)
)

وهذه الجملة ستحوَّل إلى شيئين: الأمر بلغة SQL والمعطيات بجانبها:

INSERT INTO users (name, birth_date)
VALUES (:name, :birth_date) RETURNING users.id;

{
  'name': 'Adam Banana',
  'birth_date': datetime.date(1990, 1, 1)
}

والآن نريد أن نرسل هذه الجملة كأوامِر لنظام قاعدة البيانات:

with engine.connect() as conn:
    result_cursor = conn.execute(stmt)
    conn.commit()
    row = result_cursor.fetchone()
    user_id = row[0]
    print(user_id)

وإليك شرح هذه القطعة:

1. نأخذ اتصالاً من المحرِّك (connect as conn)
2. ونرسل من خلاله الجملة execute(stmt)، ثم نحصل على النتائج (result_cursor) الذي يطبِّق بروتوكول Cursor المعروف في DB-API
3. فنعرف أنه يمكننا من خلاله سحب صف واحد (fetchone)
4. ثم نأخذ أول نتيجة في هذا الصف ([0]row) لنحصل على user_id

الآن سنجمع هذه الخطوات ونضعها في دالة بايثون:

def add_user(name: str, birth_date: datetime.date):
    stmt = (
        insert(users_table)
        .values(
            name=name,
            birth_date=birth_date
        )
        .returning(users_table.c.id)
    )
    with engine.connect() as conn:
        result = conn.execute(stmt)
        conn.commit()
        row = result.fetchone()
        user_id = row[0]
        return user_id

ثم يكون استدعاؤها بهذا الشكل:

user_id = add_user(
    name="Adam Banana",
    birth_date=datetime.date(1990, 1, 1)
)
print(user_id)

إضافة مجموعة بيانات

نريد أن نعيِّن مجموعة عنواين البريد الإلكتروني للمستخدم. وذلك يعني حذف كل القديم وإضافة القائمة الجديدة. تأمل المثال التالي وسنشرحه:

def set_user_emails(user_id: int, emails: list[str]):
    with engine.connect() as conn:
        conn.execute(
            delete(emails_table)
            .where(emails_table.c.user_id == user_id)
        )
        result_cursor = conn.execute(
            insert(emails_table).values([
                { "email": email, "user_id": user_id }
                for email in emails
            ])
        )
        conn.commit()
        return result_cursor.rowcount

عرفنا دالة في بايثون، تأخذ معرِّف المستخدم، وقائمة بالعناوين. ونحن نعلم أن العلاقة واحدة من طرف جدول المستخدم (users_table) ومتعددة من طرف جدول عناوين البريد الإكلتروني (user_emails_table)، وعليه كتبنا:

أولاً حذف القديم؛ وذلك بجملة الحذف (delete) حيث يتم تعيين الجدوَل المحذوف منه، ثم في جملة الشرط (where) يتم تحدد الشرط بالمسار (emails_table.c.user_id) حيث الحرف (c) هنا واسطة تتضمن أسماء الأعمدة الموجودة في الجدوَل؛ وهذا يعني أن بايثون تساعدنا في الإملاء إن أخطأنا.

إضافة الجديد، وربط كل صف بمعرِّف المتسخدم نفسه؛ وقد استعملنا جملة الإنشاء المختصر (comprehension) لقائمة مكوَّنة من قواميس (يعبَّر عنها هكذا: list[dict])؛ في كل قاموس منها اسم العمود وما يقابله من قيمة.

ثم نستدعيها هكذا دون الحاجة لمعرفة التفاصيل (كيفية الإضافة)، وقد حصلنا على معرِّف المستخدم user_id من النص البرمجي السابق:

emails = [
    "xyz@example.com",
    "abc@example.com",
    "def@example.com",
]
add_user_emails(user_id, emails)

استعلام الاختيار

تأمل المثال التالي الذي نستعمل فيه جملة select ونشير إلى الأعمدة بمسار يبدأ من اسم الجدوَل ثم حرف (c) ثم اسم العمود (وبهذا ستضمن لنا بايثون صحة الإملاء):

def get_user_emails(user_id: int):
    with engine.connect() as conn:
        result_cursor = conn.execute(
            select(emails_table.c.email)
            .where(emails_table.c.user_id == user_id)
        )
        emails = [row[0] for row in result_cursor]
        return emails

وإليك شرح الزائد على ما تقدَّم، وهو السطر الأخير: فلأن النتيجة (result_cursor) تطبق آلية Cursor من البروتوكول DB-API؛ فإننا نعرف أنه يمكننا كرُّها باستعمال حلقة (for) على هذا النحو، وكل كرَّة ستنتج لنا صفًّا كاملاً لكننا نعلم أن فيه قيمة واحدة: email ولذلك نختار القيمة الأولى ([0]row) من كل صف من النتائج.

ثم نستديعها كأي دالة في بايثون على هذا النحو:

emails = get_user_emails(user_id)
print(emails)

استعلام التلخيص

وأما الدوال، ومنها كلمات جمل الاختصار فيتم استيرادها على النحو التالي:

from sqlalchemy.sql.functions import count, sum, min, max, mode

واستعمالها نحو هذا المثال، حيث نريد معرفة عدد العنواين لكل من المتسخدمين على حدة:

def count_emails_per_user():
    with engine.connect() as conn:
        result_cursor = conn.execute(
            select(
                emails_table.c.user_id,
                count(emails_table.c.email)
            )
            .group_by(emails_table.c.user_id)
        )
        return result_cursor.fetchall()

ثم استدعاؤها هكذا:

print(count_emails_per_user())

النزول إلى كتابة نص SQL خام

من مميزات SQLAlchemy أنها تسمح بخرق التجريد والنزول إلى مستوى كتابة SQL يدويًّا حين يتطلَّب الأمر ذلك.

استبدال كلي

الطريقة الأولى: استبدال كُلِّي باستعمال text ثم ربط المتغيرات بقيمها:

from sqlalchemy import select, text

sql_text = text(
    "SELECT name, salary"
    "FROM employees"
    "WHERE department = :var_name"
      "AND salary > :var_salary"
)

parameters = {
    "var_name": "Sales",
    "var_salary": 50000
}

with engine.connect() as conn:
    result = conn.execute(sql_text, parameters)
    for row in result:
        print(f"Name: {row.name}, Salary: {row.salary}")

قد تتساءل لماذا لا نكتب قيمة المتغيرات مباشرة؟ والجواب عن ذلك يكمن في أن تمرير المعطيات الآتية من نظام خارجي قد يعرضنا لاخراق حقن SQL المسمى (SQL Injection). لذلك يتم الفصل بين القالب النصي (text) وبين المتغيرات parameters.

وفائدة أخرى هي الاقتصاد المؤدي للسرعة. فقد يُحفَظُ القالب إن تكرر كثيرًا إرسال نفس الجُملة، وبالتالي لا نحتاج إلا لإرسال إشارة إليها في ذاكرة خادم قاعدة البيانات، وبذلك نوفِّر في موارد الشبكة وموارد المعالجة، ونجعل النتيجة أسرع.

استبدال جزئي

الطريقة الثانية: استبدال جُزئي باستعمال literal_column، وذلك على النحو التالي:

from sqlalchemy import select, literal_column

stmt = (
    select(
        employees_table.c.name,
        literal_column("salary * 1.15").label("new_salary")
    )
    .order_by(literal_column("new_salary").desc())
)
print(stmt)

لتظهر نتيجة الجملة عند الطباعة:

SELECT
    employees.name,
    salary * 1.15 AS new_salary
FROM employees
ORDER BY new_salary DESC

6 خلاصة

إنّ SQLAlchemy يرفع العمل مع قواعد البيانات من وحل السائقين إلى طبقةٍ أرقى؛ فيجعل المحرّك أصل التعامل، يتولّى صوغ نصّ الاتصال، وضبط اللهجة، وإدارة مجمّع الاتصالات. ثم تُحدَّد البنية عبر Metadata وجداولٍ ذات أعمدة وقيود، فتُترجم إلى DDL عند الطلب. وبعد تثبيت المخطّط تُنشأ عبارات الإدراج والاختيار والتعديل بوسائط بايثونية تُحوَّل إلى SQL منقَّح، ويُستعاد الناتج على هيئة صفوفٍ تمضي كما يمضي المؤشّر في DB-API. ولمن أراد النفاذ إلى العمق، فالإطار يفتح باب SQL الخام بلا تجريد، مع حفظٍ للأمن عبر الفصل بين القالب والمعطيات.

السؤال التالي: ما هي الطريقة السليمة لتطوير مخطط قاعدة البيانات بحيث نتجنَّب خسارة البيانات أو تعارض المخطط التجريبي مع المخطط الحقيقي؟

---

ملحق

في هذا الملحق نُدرِج الأمثلة التي رأيناها في درس لغة البيانات لأنها تستعرض خصائص مختلفة من SQL.

نبدأ أولاً بتعريف الجداول:

from sqlalchemy.schema import MetaData, Table, Column, ForeignKey
from sqlalchemy.types import DateTime, Integer, String

metadata = MetaData()

products_table = Table(
    "products",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("price", Integer),
)

orders_table = Table(
    "orders",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("order_datetime", DateTime),
    Column("customer_id", Integer, ForeignKey("customers.id")),
)

المثال الأول: الشروط

الاستعلام: اختر الزبائن من دولة “ألمانيا”، والذين أعمارهم إما تحت 40 وإما فوق 60:

SELECT name FROM Customers
WHERE Country = 'Germany'
  AND (Age < 40 OR Age > 60);

وما يقابلها في بايثون:

from sqlalchemy import select, or_

stmt = (
    select(
        customers_table.c.name
    )
    .where(customers_table.c.country == 'Germany')
    .where(or_(
        customers_table.c.age < 40,
        customers_table.c.age > 60
        )
    )
)
print(stmt)

لاحظ أننا نضطر لوضع شرطة سفلية في كلمة _or لأنها من غيرها محجوزة ضمن كلمات لغة بايثون، وكذلك الأمر في _and لكننا لم نحتج إلى استعمالها، إذ تكرار where مكافئ لها.

SELECT customers.name
FROM customers
WHERE customers.country = :country_1
  AND (customers.age < :age_1 OR customers.age > :age_2)

المثال الثاني: الترتيب والحد

SELECT * FROM Products
ORDER BY Price DESC
LIMIT 5;

stmt = (
    select(products_table)
    .order_by(products_table.c.price.desc())
    .limit(5)
)
print(stmt)

نتيجة الطباعة:

SELECT products.id, products.name, products.price
FROM products
ORDER BY products.price DESC
LIMIT :param_1

المثال الثالث: نفي العضوية

الاستعلام: أريد الزبائن الذين ليسوا من دولة كذا وكذا وكذا:

SELECT * FROM customers
WHERE country NOT IN ('Germany', 'France', 'UK');

block_list = ['Germany', 'France', 'UK']
stmt = (
    select(customers_table)
    .where(customers_table.c.country.notin_(block_list))
)

المثال الرابع: دمج الجداول

استعلام: أريد أسماء المنتجات وأسعارها ومعرِّف الطلب من الطلبات التي حصلت في النصف الأول من السنة:

SELECT orders.id, products.name, products.price
FROM orders JOIN products
  ON orders.id = products.id
WHERE orders.order_datetime BETWEEN '1996-01-01' AND '1996-06-01';
  ;

stmt = (
    select(
        orders_table.c.id,
        products_table.c.name,
        products_table.c.price
    )
    .join(products_table, orders_table.c.id == products_table.c.id)
    .where(orders_table.c.order_datetime.between('1996-01-01', '1996-06-01'))
)

المثال الخامس: الترشيح قبل وبعد التجميع

وهذا المثال يحتاج إلى جدوَل الطلبة، فننشئه على هذا النحو:

students_table = Table(
    "students",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("country", String),
    Column("score", Integer),
)

جملة الاستعلام: متوسط نقاط الطلبة الناجحين (فوق 60 درجة) بحسب الدولة (يرشح من ذلك الدول التي يتجاوز متوسطها 90):

SELECT country, (SUM(score) / COUNT(id)) AS avg
FROM students
WHERE score >= 60
GROUP BY country
HAVING avg_score_per_country > 90
ORDER BY avg_score_per_country DESC;

وهذا مقابله في بايثون:

total_score = func.sum(students_table.c.score).label('total_score')

count_score = func.count(students_table.c.score).label('count_score')

avg_score_per_country = (total_score / count_score).label('avg')

stmt = (
    select(students_table.c.country, avg_score_per_country)
    .where(students_table.c.score >= 60)
    .group_by(students_table.c.country)
    .having(avg_score_per_country > 90)
    .order_by(avg_score_per_country.desc())
)

المثال السادس: جملة الاختيار المركبة

تعريف الجداول في بايثون:

departments_table = Table(
    "departments",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
)

employees_table = Table(
    "employees",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("dep_id", Integer, ForeignKey("departments.id")),
    Column("salary", Integer),
)

جملة الاستعلام: أريد أسماء الأقسام التي لديها موظفين رواتبهم تتجاوز التسعة آلاف.

SELECT name FROM Departments
WHERE id IN (
    SELECT dep_id FROM Employees
    WHERE salary > 9000);

وفي بايثون:

stmt = (
    select(departments_table.c.name)
    .where(departments_table.c.id.in_(
        select(employees_table.c.dep_id)
        .where(employees_table.c.salary > 9000)
    ))
)