SQLAlchemy - دیتابیس در دستان شماست

تو پروژه هایی که نیاز کمی به ارتباط با دیتابیس (MySQL) داشتن از کتابخونه PyMySQL استفاده می‌کردم برای پروژه های کوچیک خیلی کار راه بنداز بود تا اینکه در آخرین پروژم بکل داستان عوض شد.

تو این پروژه بحث ارتباط با دیتابیس و ارسال و دریافت اطلاعات و موجودیت های سیستم بسیار پر رنگ بود دیگه نمیتونستم از PyMySQL استفاده کنم چون PyMySQL یک wrapper ساده بود که دستورات خام SQL رو اجرا میکرد ولی من یه ORM لازم داشتم با امکانات زیاد تا کار رو برام آسون تر کنه

تو ذهنم بود که از یه ORM استفاده کنم که

کارایی و سرعت سیستم رو به صورت مشخصی کم نکنه
داکیومنت خوبی داشته
کامیونیتی فعالی داشته باشه تا اگه مشکلی پیش اومد بتونم سریع حلش کنم
سریع بشه باهاش کار کرد و مدت زمان زیادی رو برای یادگرفتن از من نگیره

اگه از حق نگذریم SQLAlchemy همه اینا رو داره

به هر حال من SQLAlchemy رو انتخاب کردم با اینکه میشه سریع یک پروژه رو با SQLAlchemy شروع کرد ولی متوجه شدن بعضی مفاهیم اون خیلی از من وقت گرفت من در این نوشته میخوام از اول تا آخر کارهایی که تو این پروژه با SQLAlchemy انجام دادم رو به شما توضیح بدم تا کامل مفاهیم رو متوجه بشید.

آموزش SQLAlchemy

اول با دستور زیر SQLAlchemy رو نصب کنید

pip3 install sqlalchemy

1	pip3 install sqlalchemy

اولین قدم ارتباط با دیتابیس هست. برای ارتباط با هر دیتابیسی نیاز به یک url با فرمت مخصوص همون دیتابیس هست الان که دیتابیس ما MySQL از کد زیر استفاده میکنیم

from sqlalchemy import create_engine

db_engine = create_engine(
    "mysql+pymysql://{}:{}@{}/{}?charset=utf8mb4".format(
        DB_USER, DB_PASSWORD, DB_SERVER, DB_NAME
    ),
    pool_size=10,
    max_overflow=5,
    pool_timeout=30,
    echo=False,
)

from sqlalchemy import create_engine

db_engine = create_engine(

"mysql+pymysql://{}:{}@{}/{}?charset=utf8mb4".format(

DB_USER, DB_PASSWORD, DB_SERVER, DB_NAME

pool_size=10,

max_overflow=5,

pool_timeout=30,

echo=False,

)

اشتباه نکنید با این کد SQLAlchemy به دیتابیس وصل نمیشه در واقع SQLAlchemy تنها زمانی که لازم باشه به دیتابیس وصل میشه یعنی زمانی که شما یه چیزی رو از دیتابیس درخواست کنید و یا تغییری رو اعمال کنید

این تابع چند تا پارامتر دیگه هم قبول میکنه poolClass, pool_size, max_overflow, pool_timeout,echo

echo

اگه مقدار این پارامتر رو True قرار بدید هر کوئری که به دیتابیس بفرستید تو خروجی براتون به نمایش در میاد و اگه debug بزارید علاوه بر query خود نتابج رو هم نشون میده

autocommit

تمام عملیات ها در SQLAlchemy در قالب transaction انجام میشه پس باید برای اعمال تغییراتی که هر کوئری در دیتابیس ایجاد کرده دستور commit رو اجرا کنید اگه نخواید commit رو انجام بدید و به این صورت باشه که بعد از هر کوئری، تغییرات به صورت اتوماتیک commit بشن باید مقدار این آرگومان رو برابر True قرار بدید با این کار SQLAlchemy دستوراتی که باعث تغییر در دیتابیس میشن رو شناسایی و بعد از اون commit رو اجرا کنه (پیشنهاد میکنم مقدار این آرگومان رو همون False بزارید)

قبل از اینکه بقیه پارامتر ها رو توضیح بدم میخوام درباره Connection Pooling صحبت کنم.

Connection Pooling در SQLAlchemy

روال عادی برنامه ها اینطوری هست که برنامه بعد از شروع شدن، یک اتصال (Conenction) به دیتابیس ایجاد میکنه، درخواست ها رو به دیتابیس میفرسته و بعد اتصال رو میبنده حالا شما یک برنامه مثل یک وب سایت رو در نظر بگیرید و با هر برنامه دیگه ای که درخواست های زیادی بهش ارسال میشه، تو این برنامه در هر درخواست یک اتصال باید ایجاد و بعد از اتمام درخواست اتصال باید بسته بشه متوجه مشکل میشید؟

ایجاد اتصال به دیتابیس یک حرکت زمان بر هست در هر بار باید نام کاربری و رمز عبور چک بشه و یک سوکت ایحاد بشه و برای بسته شدن هم این سوکت حذف بشه و برای درخواست بعدی دوباره همین مراحل تکرار میشه؟ راه حل بهتری نیست؟

Connection Pooling

Connection Pooling یک تکنیک در سطح برنامه و همچنین دیتابیس هست که شما اول pool_size یعنی اندازه اتصال هایی که باید در pool قرار بگیرند رو تعریف میکنید بعد برنامه شما به ازای هر درخواست یک اتصال ایجاد میکنه و تا زمانی که تعداد اتصال ها به اندازه pool_size نرسیده این کار رو انجام میده و هیچ اتصالی رو نمیبنده اما به محض اینکه تعداد اتصال ها به pool_size رسید برای درخواست های جدید از اتصال های بیکار داخل pool استفاده میکنه و اتصال جدید به دیتابیس ایجاد نمیشه.

SQLAlchemy یک پارامتر دیگه هم داره به نام max_overflow یعنی حداکثر تعداد اتصال هایی که بیشتر از اندازه pool_size میتونن داخل pool قرار بگیرن حالا اگه تعداد کل اتصال های داخل pool مشغول باشن SQLAlchemy به اندازه max_overflow هم درخواست جدید قبول میکنه و اتصال جدید میسازه.

ببینم فرق max_overflow و pool_size رو متوجه شدید؟ اگه برای مثال pool_size=5 باشه تا زمانی که تعداد اتصال ها در pool به ۵ نرسیده برنامه میاد و اتصال جدید ایجاد میکنه اما به محض اینکه به ۵ برسه از اتصال های داخل pool استفاده میکنه و اتصال جدیدی ایجاد نمیکنه اما اگه همه اتصال های داخل pool مشغول بودن چی؟

حالا برنامه میاد برای درخواست های بیشتر از pool_size تا زمانی که بیشتر از max_overflow نشدن اتصال جدید ایجاد میکنه اما این اتصال ها دیگه به pool بر نمیگردن و به محض تموم شدن کار برنامه با اونها بسته میشن.

یعنی حداکثر اتصال هایی که برنامه میتونه بسازه به اندازه pool_size + max_overflow هست و اگه این اندازه اتصال ایجاد شده باشه و درخواست جدید بیاد، اونموقع این درخواست جدید باید به اندازه pool_timeout صبر کنه تا یکی از اتصال ها کارش تموم بشه و اگه بیشتر از این مقدار صبر کرد و اتصالی گیرش نیومد درخواست دراپ و لغو میشه.

شاید کار شما جوری باشه که نخواهید درخواست های جدید صبر کنن تا یک اتصال آزاد بشه یعنی کارتون خیلی حیاتی باشه و هر ثانیه برای برنامتون اهمیت داشته باشه اما تا اونجا که برای من پیش اومده برای ۹۹ درصد برنامه هایی که تعداد درخواست های زیادی دارن استفاده از این تکنیک کارایی برنامه رو به شدت افزایش میده و همچنین دیگه با خطای Too Many Connections هم مواجه نمیشید.

برای غیر فعال کردن Connection Pooling باید poolClass=Nullpool قرار بدید.

یک مفهوم خیلی مهم در این کتابخونه Session هست

Session ها در SQLAlchemy

خوب Session چی هست و چیکار میکنه؟

Session یک حافظه موقتی هست که وقتی به دیتابیس ارسال میشه تغییرات شما رو اعمال میکنه در واقع میتونیم بگیم هم transaction هست هم نیست. بزارید نحوه ایجادش رو بگم بعد کامل نحوه عملکردش رو توضیح میدم

from sqlalchemy.orm import sessionmaker

db_session_handler = sessionmaker(bind=db_engine)
dbs = db_session_handler()

from sqlalchemy.orm import sessionmaker

db_session_handler = sessionmaker(bind=db_engine)

dbs = db_session_handler()

هر زمان که یک session ایجاد کنید یک اتصال جداگانه به سرور دیتابیس ایجاد و به اون session اختصاص داده میشه

اگه تو Session شما یک query اجرا کنید چه select چه چیز دیگه ای یک transaction رو شروع کردید و اگه در Session هیج query اجرا نکنید و فقط یک رکورد ایحاد کنید و با تابع add اون رو به Session اضافه کنید اون موقع هم یک transaction درست کردید این ۲ تا قانون کلی درباره Session هست.

حالا چرا مهم هست که بدونید کی یک transaction ایجاد میشه؟

وقتی یک transaction ایجاد میشه اگه isolation_level برابر Repeatable Read باشه (که پیش فرضش همین هست) تا زمانی که commit نشه هر تغییری در دیتابیس ایجاد بشه اون transaction نمیتونه اون تغییر رو ببینه.

این دو نمونه از نحوه ویرایش و اضافه کردن رکورد به دیتابیس هست.

item = dbs.query(Item).get(1)  # transaction started
item.name = "new"
dbs.commit()  # changes apply to database

item = dbs.query(Item).get(1) # transaction started

item.name = "new"

dbs.commit() # changes apply to database

Item کلاس مدلمون هست که پایینتر توضیح میدم

و نوع دوم

new_item = Item()
new_item.name = "item2"
dbs.add(new_item)  # transaction started
dbs.commit()  # changes apply to database

new_item = Item()

new_item.name = "item2"

dbs.add(new_item) # transaction started

dbs.commit() # changes apply to database

وقتی شما commit میکنید چند تا اتفاق میفته:

۱. تمام تغییراتی که توسط شما ایجاد شده در دیتابیس نوشته میشه

۲. تمام متغیر های قبلی expire میشن

برای مثال در کد اول وقتی commit کردید اول اطلاعات رکورد با ای دی ۱ به روز میشه بعد متغیر item منقضی میشه و وقتی دوباره item.name رو اجرا کنید SQLAlchemy یک دستور select به دیتابیس میفرسته تا (dbs.query(Item).get(1 اجرا بشه بعد مقدار name رو به شما نشون میده

خوب حالا اون Item چی هست؟ اون درواقع کلاس جدول متناظر در دیتابیس هست که به این صورت تعریف میشه

from sqlalchemy import Column, String, Text, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False, unique=True)
    bio = Column(Text, nullable=False, unique=True)
    number = Column(Integer)

from sqlalchemy import Column, String, Text, Integer

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):

__tablename__ = "items"

id = Column(Integer, primary_key=True)

name = Column(String(100), nullable=False, unique=True)

bio = Column(Text, nullable=False, unique=True)

number = Column(Integer)

یکی از فیلدها حتما باید primary key باشه

برای دیدن انواع داده های ستون ها اینجا رو ببنید.

برای ساخت این جدول هم میتونید تو کنسول پایتون این فایل به همراه فایلی که اتصال رو تعریف کردید رو ایمپورت کنید و کد زیر رو اجرا کنید

from item import Item, Base


Base.metadata.create_all(db_engine)
# or
Item.__table__.create(bind=db_engine)

from item import Item, Base

Base.metadata.create_all(db_engine)

# or

Item.__table__.create(bind=db_engine)

برای حذف کردن کل جدول های تعریف شده هم

Base.metadata.drop_all(bind=db_engine)

1	Base.metadata.drop_all(bind=db_engine)

برای اجرای انواع عملیات در دیتابیس من یه سری کد نمونه میزارم که فکر میکنم نیازی به توضیح نداره

from sqlalchemy import asc

# select * from items where id=1
dbs.query(Item).get(1)

# select * from items where name ='test' and number < 1000;
dbs.query(item).filter(Item.name == "test", Item.number < 10)
dbs.query(item).filter(Item.name == "test").filter(Item.number < 10)
dbs.query(item).filter((Item.name == "test") & (Item.number < 10))
dbs.query(item).filter_by(name = "test", number < 10)


# select * from items where number < 10 or id > 1000;
dbs.query(item).filter((Item.number < 10) | (Item.id > 1000)).order_by(asc(Item.id))

# select * from items where name like '%test%'
dbs.query(Item).filter(Item.name.like("%test%"))

# select id, name from items order by id asc
dbs.query(Item.id, Item.name).order_by(Item.id.asc())

# update items set number=0 where id=1
item = dbs.query(Item).get(1)
item.number = 0
dbs.add(item)
dbs.commit()

# update items set number=0 where name is not Null
dbs.query(Item).filter(Item.name != None).update({"number": 0})
dbs.commit()

# delete from items where id=1
dbs.query(Item).filter(Item.id == 1).delete()
dbs.commit()
# or
item = dbs.query(Item).get(1)
dbs.delete(item)
dbs.commit()

from sqlalchemy import asc

# select * from items where id=1

dbs.query(Item).get(1)

# select * from items where name ='test' and number < 1000;

dbs.query(item).filter(Item.name == "test", Item.number < 10)

dbs.query(item).filter(Item.name == "test").filter(Item.number < 10)

dbs.query(item).filter((Item.name == "test") & (Item.number < 10))

dbs.query(item).filter_by(name = "test", number < 10)

# select * from items where number < 10 or id > 1000;

dbs.query(item).filter((Item.number < 10) | (Item.id > 1000)).order_by(asc(Item.id))

# select * from items where name like '%test%'

dbs.query(Item).filter(Item.name.like("%test%"))

# select id, name from items order by id asc

dbs.query(Item.id, Item.name).order_by(Item.id.asc())

# update items set number=0 where id=1

item = dbs.query(Item).get(1)

item.number = 0

dbs.add(item)

dbs.commit()

# update items set number=0 where name is not Null

dbs.query(Item).filter(Item.name != None).update({"number": 0})

dbs.commit()

# delete from items where id=1

dbs.query(Item).filter(Item.id == 1).delete()

dbs.commit()

# or

item = dbs.query(Item).get(1)

dbs.delete(item)

dbs.commit()

اینجا هم میتونید نمونه های بیشتری ار دستورات SQLAlchemy رو ببینید.

وقتی از دو تابع filter و filter_by استفاده میکنید هیچ چیزی اجرا نمیشه یعنی شما اگه کد زیر رو پرینت بگیرید میبیند که تنها یک دستور sql هست نه چیز دیگه ای

result = dbs.query(Item).filter(name == 'test')

1	result = dbs.query(Item).filter(name == 'test')

دستور SQL زمانی اجرا میشه که یکی از متدهای all, first یا one رو اجرا کنید و یا متغیر result رو تو حلقه for بزارید

و توضیح این متدها

all

همه رکوردهای کوئری رو بر میگردونه و اگه چیزی پیدا نکنه یکی لیست خالی

first

به انتهای دستور limit 1 اضافه میکنه و اگه باز نتایج خالی باشه None بر میگردونه

one

با این متد به SQLALchemy میگید که نتیجه فقط و فقط باید یک کورد باشه و همون رکورد رو برگردونه اگه این دستور بیشتر از یک نتیجه داشته باشه خطا نشون داده میشه

چند مثال پائین دقیقا مثل هم هستند

result = dbs.query(Item)
for row in result:
    ...


result = dbs.query(Item).all()
for row in result:
    ...


result = list(dbs.query(Item))
for row in result:
    ...

result = dbs.query(Item)

for row in result:

...

result = dbs.query(Item).all()

for row in result:

...

result = list(dbs.query(Item))

for row in result:

...

برای مثال بیشتر اینجا رو ببینید.

چطوری رکورد های زیادی رو دریافت کنیم؟

اول پیشنهاد میکنم نوشته قبلیم ۷۰۰ میلیون رکورد MySQL را به راحتی آب خوردن با PHP پردازش کنید و اجرای دستورات خام SQL در SQLAlchemy قسمت چطوری رکورد های زیادی رو دریافت کنیم؟ که در همین رابطه هست رو بخونید.

وقتی نتایج رو با all دریافت میکنید و یا به طور مستقیم مقدار (dbs.query(Item رو تو حلقه استفاده میکنید (کد اول در مثال بالا) کل نتایج یکجا از دیتابیس دریافت میشه و در حافظه قرار میگیره و نتایج buffer میشن، حالا مواقعی هست که نتایج خیلی زیاد هست برای مثال بیشتر از ۱۰ هزار رکورد، وقتی همچین نتایجی رو میخواید پردازش کنید ممکن هست با مشکل حافظه مواجه بشید برای حل این مشکل باید از unbuffered queries استفاده کنید تا نتایج رو قسمت قسمت از دیتابیس دریافت کنید کد زیر همین کار رو میکنه و نتایج رو به صورت stream از MySQL میگیره.

result = dbs.query(Item)
for row in result.yield_per(1000):
    ...

result = dbs.query(Item)

for row in result.yield_per(1000):

...

در کد بالا در هر پیمایش شما یک رکورد رو میگیرید ولی کل نتایج در قسمت های ۱۰۰۰ تایی از دیتابیس دریافت میشه تا حافظه سرور پایتون یکباره پر نشه.

در واقع yield_per متغیر execution_options مربوط به create_engine رو برابر True قرار میده (به همراه یک تغییر دیگه)

مثال های بیشتر رو اینجا میتونید ببینید.

نکات مهم

در SQLAlchemy اگه مقدار فیلدی رو برابر None بزارید در دیتابیس برابر Null قرار میگیره
باید دقت کنید False برای هر نوع داده ای مقدار برابری برنمیگردونه برای مثال اگه فیلدی نوعش int باشه و شما مقدارش رو False قرار بدید SQLAlchemy در دیتابیس مقدارش رو ۰ میکنه و اگه نوعش string باشه Null میشه.
برای بستن اتصال هم این نکته مهم رو داشته باشید

برای اینکه Session رو ببندید کافیه متد close رو صدا بزنید با اینکار اگه Connection Pooling فعال باشه این اتصال به pool برمیگرده در غیر این صورت اتصال بسته میشه و منابع سخت افزاری آزاد میشه

پس برای بستن اتصال کد زیر رو اجرا می‌کنیم

dbs.close()

1	dbs.close()

برای اینکه کل اتصال های دیتابیس رو ببینید و منابع سخت افزاری رو آزاد کنید کد زیر رو اجرا کنید

db_engine.dispose()

1	db_engine.dispose()

زمانی این کد رو اجرا کنید که کل برنامتون کارش تموم شده باشه و بهتره تو هندلر atexit بزارید

تمام عملیات ها در SQLAlchemy در قالب transaction انجام میشه حتی وقتی autocommit=True باشه SQLAlchemy قبل و بعد از هر دستور transaction رو شروع و commit میکنه.

آموزش های بیشتری هم در سایت خود SQLAlchemy وجود داره که میتونید مطالعه کنید.

همچنین اگه حجم رکوردهاتون زیاد هست اینجا میتونید مقایسه های انجام شده رو ببینید

برای اجرای دستورات خام SQL در SQLAlchemy این نوشته رو بخونید.

آموزش SQLAlchemy

Connection Pooling در SQLAlchemy

Connection Pooling

Session ها در SQLAlchemy

چطوری رکورد های زیادی رو دریافت کنیم؟

نکات مهم

دیدگاهتان را بنویسید لغو پاسخ