اتصال از پایتون و جنگو: psycopg 3 و django.contrib.postgres

از کد برنامه به دیتابیس، درست و امن

psycopg 3

psycopg 3 جانشین psycopg2 است: پشتیبانی بومی از async، COPY راحت، pool جدا و ارسال پارامتر سمت سرور. اگر pip به خاطر تحریم یا کندی شبکه به مشکل خورد، از یک میرور داخلی PyPI با گزینه‌ی -i استفاده کنید.

pip install "psycopg[binary]" psycopg_pool

import psycopg
from psycopg.rows import dict_row

DSN = "host=127.0.0.1 port=5432 dbname=carpet user=app_user password=... sslmode=prefer"

with psycopg.connect(DSN, row_factory=dict_row) as conn:
    # پارامتر همیشه با %s، هرگز با f-string (SQL injection)
    rows = conn.execute(
        "SELECT id, status FROM factory.orders WHERE customer_id = %s AND status = %s",
        (42, "weaving"),
    ).fetchall()

    # ورود انبوه با COPY: ده‌ها برابر سریع‌تر از INSERTهای تکی
    with conn.cursor() as cur:
        with cur.copy("COPY factory.customers (full_name, city) FROM STDIN") as copy:
            for name, city in [("زهرا کریمی", "کاشان"), ("علی نراقی", "آران و بیدگل")]:
                copy.write_row((name, city))
# خروج از with: COMMIT (یا ROLLBACK در صورت خطا) و بستن اتصال

تنظیمات جنگو

# settings.py
import os

INSTALLED_APPS += ["django.contrib.postgres"]

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",   # جنگو 4.2+ خودکار psycopg 3 را برمی‌دارد
        "NAME": "carpet",
        "USER": "app_user",
        "PASSWORD": os.environ["DB_PASSWORD"],
        "HOST": "127.0.0.1",
        "PORT": "5432",
        "CONN_MAX_AGE": 60,            # اتصال را ۶۰ ثانیه بین درخواست‌ها نگه دار
        "CONN_HEALTH_CHECKS": True,    # قبل از استفاده‌ی مجدد، سلامت اتصال را بسنج
        "OPTIONS": {
            "options": "-c search_path=factory,public -c statement_timeout=30000",
        },
    }
}

django.contrib.postgres در عمل

from django.contrib.postgres.fields import ArrayField
from django.contrib.postgres.indexes import GinIndex
from django.contrib.postgres.search import SearchVector, SearchQuery, SearchRank, TrigramSimilarity
from django.db import models

class Design(models.Model):
    code = models.CharField(max_length=20, primary_key=True)
    title = models.CharField(max_length=100)
    tags = ArrayField(models.CharField(max_length=30), default=list, blank=True)

    class Meta:
        indexes = [GinIndex(fields=["tags"], name="design_tags_gin")]

Design.objects.filter(tags__contains=["ابریشم"])          # tags @> ARRAY[...]
Design.objects.filter(tags__overlap=["افشان", "هریس"])     # tags && ARRAY[...]

q = SearchQuery("افشان", config="simple", search_type="websearch")
(Design.objects
   .annotate(rank=SearchRank(SearchVector("title", config="simple"), q))
   .filter(rank__gt=0).order_by("-rank"))

Design.objects.annotate(sim=TrigramSimilarity("title", "افشن")).filter(sim__gt=0.3).order_by("-sim")

TrigramSimilarity به اکستنشن pg_trgm نیاز دارد؛ آن را در یک migration با عملیات TrigramExtension() فعال کنید تا روی سرور تازه فراموش نشود.

نکته‌هایی که کمتر کسی می‌داند

• از جنگو 5.1 می‌توانید با "OPTIONS": {"pool": True} از pool داخلی psycopg استفاده کنید؛ در این حالت CONN_MAX_AGE باید 0 بماند و پشت PgBouncer هم به آن نیازی ندارید.
• CONN_MAX_AGE = None (اتصال دائمی) در کنار gunicorn با worker زیاد سریع به سقف max_connections می‌رسد؛ تعداد اتصال‌ها برابر workers × threads است.
• جست‌وجوی SearchVector در کوئری، tsvector را برای هر ردیف از نو می‌سازد؛ برای جدول بزرگ ستون tsvector ذخیره‌شده با ایندکس GIN (مثل فصل ۵) بسازید و با SearchVectorField به آن اشاره کنید.
• psycopg 3 پارامترها را سمت سرور می‌فرستد؛ پس %s فقط جای مقدار می‌نشیند، نه نام جدول یا ستون. برای شناسه‌ها از psycopg.sql.Identifier استفاده کنید.
• متد .iterator() در جنگو روی پستگرس server-side cursor می‌سازد و حافظه را برای میلیون‌ها ردیف ثابت نگه می‌دارد؛ اما پشت PgBouncer تراکنشی باید DISABLE_SERVER_SIDE_CURSORS را روشن کنید (فصل ۷).