کدهای آماده CRUD با SQLAlchemy در پایتون

رفیق برنامه‌نویس، اگه تا حالا دنبال راهی بودی که عملیات دیتابیسی CRUD (ساخت، خواندن، به‌روزرسانی، حذف) رو تو پروژه‌های پایتونیت با SQLAlchemy تمیز و سرراست پیاده‌سازی کنی، جای درستی اومدی. اینجا کلی کدهای آماده پایتون و اسنیپت‌های کاربردی منتظرته. فقط یه نگاه بنداز به فروشگاه ابزارهای برنامه‌نویسی ما تا ببینی چه گنجینه‌ای از اسنیپت‌های آماده اونجا داریم که کارت رو کلی راحت می‌کنه! همین الان یه سر بزن و کلی کدهای خفن رو کشف کن.

📞 تماس: 09202232789

تو دنیای توسعه وب و بک‌اند، کار با دیتابیس یه جزء جدانشدنیه. عملیات CRUD هم که دیگه الفبای کار با دیتابیس محسوب می‌شه. پایتون ابزارهای زیادی برای این کار داره، اما SQLAlchemy یکی از قدرتمندترین و انعطاف‌پذیرترین اوناست. این کتابخونه یه پکیج کامل برای ارتباط با انواع دیتابیس‌هاست، از SQLite ساده و فایل‌محور بگیر تا PostgreSQL و MySQL.

دلیل اینکه ما اینجا به سراغ SQLAlchemy اومدیم، توانایی بی‌نظیرشه در ایجاد یه لایه انتزاعی بین کد پایتون شما و دیتابیس. این یعنی دیگه لازم نیست نگران تفاوت‌های سینتکسی SQL برای دیتابیس‌های مختلف باشید. SQLAlchemy هم برای کوئری‌های خام SQL (SQL Expression Language) عالیه، هم برای مدل‌سازی شی‌گرا (ORM). پس اگه می‌خوای کد پایتون تمیز و قابل نگهداری داشته باشی، SQLAlchemy یه انتخاب هوشمندانه‌ست. این مقاله یه راهنمای جامع برای پیاده‌سازی عملیات CRUD با SQLAlchemy هستش و می‌تونی کلی کدهای آماده پایتونی برای پروژه‌هات از اینجا برداری.

SQLAlchemy یه کیت ابزار دیتابیس (Database Toolkit) و ORM (Object Relational Mapper) برای پایتونه که به توسعه‌دهنده‌ها این امکان رو می‌ده که با دیتابیس‌های رابطه‌ای به صورت شی‌گرا کار کنن. به جای اینکه مستقیماً کوئری‌های SQL بنویسی، با آبجکت‌های پایتونی کار می‌کنی و SQLAlchemy اونا رو به SQL تبدیل می‌کنه.

ارکستراسیون دیتابیس با ORM

ORM در SQLAlchemy یه نقشه‌برداری (mapping) بین آبجکت‌های پایتونی و ردیف‌های جدول تو دیتابیس ایجاد می‌کنه. این یعنی هر ردیف (رکورد) از جدول شما به یه آبجکت (instance) از یه کلاس پایتونی تبدیل می‌شه. این کار، مدیریت و تعامل با داده‌ها رو خیلی ساده‌تر و پایتونی‌تر می‌کنه. دیگه لازم نیست نگران جزئیات کم‌سطح دیتابیس باشی.

انعطاف‌پذیری و کنترل کامل

یکی از نقاط قوت SQLAlchemy اینه که هیچ‌وقت دست و پات رو نمی‌بنده. اگه لازم شد که کوئری‌های پیچیده SQL بنویسی که ORM نمی‌تونه به خوبی پوشش بده، می‌تونی به راحتی از SQL Expression Language یا حتی کوئری‌های خام SQL استفاده کنی. این انعطاف‌پذیری باعث می‌شه که SQLAlchemy برای پروژه‌های کوچیک تا بزرگ و پیچیده مناسب باشه.

نصب SQLAlchemy و ابزارهای لازم

اولین قدم، نصب SQLAlchemy هست. خیلی ساده با pip می‌تونید این کار رو انجام بدید:

pip install SQLAlchemy
        

اگه می‌خواید با دیتابیس‌های خاصی مثل PostgreSQL یا MySQL کار کنید، درایورهای مربوطه رو هم باید نصب کنید (مثلاً `psycopg2` برای PostgreSQL یا `mysql-connector-python` برای MySQL). ما اینجا از SQLite استفاده می‌کنیم که نیازی به درایور اضافه نداره چون پایتون خودش پشتیبانی داخلی ازش داره.

ساخت مدل دیتابیس

حالا باید مدل (Schema) دیتابیس رو تعریف کنیم. فرض کنید می‌خوایم یه دیتابیس برای مدیریت کاربران بسازیم که هر کاربر یه ID، نام، ایمیل و تاریخ ثبت‌نام داشته باشه.

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime

# Connection String
DATABASE_URL = "sqlite:///./test.db" # دیتابیس SQLite در همین مسیر پروژه

# ایجاد Engine برای ارتباط با دیتابیس
engine = create_engine(DATABASE_URL, echo=True) # echo=True برای دیدن کوئری‌های SQL تولید شده

# Base برای تعریف مدل‌های دیتابیس
Base = declarative_base()

# تعریف مدل User
class User(Base):
    __tablename__ = "users" # نام جدول در دیتابیس

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    registered_at = Column(DateTime, default=datetime.datetime.now)

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

# ایجاد جداول در دیتابیس
Base.metadata.create_all(engine)

# پیکربندی Session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# تابع کمکی برای دریافت Session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
        

تو این کد:

• `create_engine`: ارتباط با دیتابیس رو برقرار می‌کنه.
• `declarative_base`: پایه و اساس مدل‌های ORM ماست.
• `User`: کلاسی که مدل جدول `users` رو نشون می‌ده. هر خصوصیت (attribute) از این کلاس، یه ستون تو جدول دیتابیسه.
• `Base.metadata.create_all(engine)`: تمام جداولی که از `Base` ارث‌بری کردن رو تو دیتابیس می‌سازه.
• `sessionmaker`: یه “کارخانه” تولید سشن (Session) دیتابیسه. Session همون رابطی هست که ما برای تعامل با دیتابیس ازش استفاده می‌کنیم.
• `get_db`: یه تابع `generator` که برای مدیریت سشن تو یه بلاک `try…finally` استفاده می‌شه تا مطمئن بشیم سشن بعد از اتمام کار بسته می‌شه. (این الگو تو فریم‌ورک‌هایی مثل FastAPI خیلی رایجه).

حالا که مدل دیتابیس رو داریم و سشن آماده‌ست، بریم سراغ عملیات اصلی CRUD. اینا همون کدهای آماده و اسنیپت‌هایی هستن که می‌تونی تو پروژه‌هات مستقیم استفاده کنی.

C: ساخت (Create) رکورد جدید

برای اضافه کردن یه کاربر جدید، کافیه یه آبجکت از کلاس `User` بسازیم، به سشن اضافه‌اش کنیم و بعد `commit` کنیم تا تغییرات تو دیتابیس ذخیره بشن.

# فرض می‌کنیم 'db' یه Session فعال از 'get_db()' هست
def create_user(db, name: str, email: str):
    new_user = User(name=name, email=email)
    db.add(new_user)
    db.commit() # ذخیره تغییرات
    db.refresh(new_user) # به‌روزرسانی آبجکت برای دریافت ID تولید شده
    return new_user

# مثال استفاده:
# for db_session in get_db():
#     user1 = create_user(db_session, "Ali Ahmadi", "ali.ahmadi@example.com")
#     print(f"کاربر جدید ساخته شد: {user1}")
#     user2 = create_user(db_session, "Sara Karimi", "sara.karimi@example.com")
#     print(f"کاربر جدید ساخته شد: {user2}")
        

R: خواندن (Read) رکوردها

برای بازیابی اطلاعات، از متد `query()` سشن استفاده می‌کنیم. می‌تونیم بر اساس ID، ایمیل یا هر فیلد دیگه‌ای فیلتر کنیم.

# خواندن همه کاربران
def get_all_users(db, skip: int = 0, limit: int = 100):
    return db.query(User).offset(skip).limit(limit).all()

# خواندن کاربر بر اساس ID
def get_user_by_id(db, user_id: int):
    return db.query(User).filter(User.id == user_id).first() # .first() یا .one_or_none()

# خواندن کاربر بر اساس ایمیل
def get_user_by_email(db, email: str):
    return db.query(User).filter(User.email == email).first()

# مثال استفاده:
# for db_session in get_db():
#     all_users = get_all_users(db_session)
#     print(f"nهمه کاربران: {all_users}")
#
#     user_id_1 = get_user_by_id(db_session, 1)
#     print(f"کاربر با ID 1: {user_id_1}")
#
#     user_email = get_user_by_email(db_session, "ali.ahmadi@example.com")
#     print(f"کاربر با ایمیل Ali: {user_email}")
        

U: به‌روزرسانی (Update) رکوردها

برای به‌روزرسانی یه رکورد، اول باید اون رکورد رو از دیتابیس بخونیم، تغییرات لازم رو روی آبجکت پایتونی اعمال کنیم و دوباره `commit` کنیم.

def update_user(db, user_id: int, new_name: str = None, new_email: str = None):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        if new_name:
            user.name = new_name
        if new_email:
            user.email = new_email
        db.commit()
        db.refresh(user)
    return user

# مثال استفاده:
# for db_session in get_db():
#     updated_user = update_user(db_session, 1, new_name="Ali Reza Ahmadi")
#     print(f"nکاربر به‌روزرسانی شد: {updated_user}")
#     updated_user2 = update_user(db_session, 2, new_email="sara.k@example.com")
#     print(f"کاربر به‌روزرسانی شد: {updated_user2}")
        

D: حذف (Delete) رکوردها

برای حذف یه رکورد، اون رو از دیتابیس پیدا می‌کنیم، به سشن دستور `delete()` رو می‌دیم و در نهایت `commit` می‌کنیم.

def delete_user(db, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
        return True # حذف موفق
    return False # کاربر یافت نشد

# مثال استفاده:
# for db_session in get_db():
#     delete_status = delete_user(db_session, 2)
#     print(f"nحذف کاربر با ID 2 موفقیت‌آمیز بود؟ {delete_status}")
#
#     all_users_after_delete = get_all_users(db_session)
#     print(f"همه کاربران بعد از حذف: {all_users_after_delete}")
        

وقتی یه پروژه بزرگ‌تر می‌شه، بهتره کدهات رو ساختاردهی کنی تا هم قابل نگهداری‌تر باشن و هم تیم‌های دیگه راحت‌تر باهاشون کار کنن.

جدا کردن منطق دیتابیس

بهتره کدهای مربوط به دیتابیس (مدل‌ها، توابع CRUD) رو تو فایل‌های جداگونه‌ای نگه داری. مثلاً:

• `database.py`: شامل `engine`, `Base`, `SessionLocal`, `get_db`.
• `models.py`: شامل کلاس‌های مدل دیتابیس مثل `User`.
• `crud.py`: شامل توابع CRUD مثل `create_user`, `get_user_by_id`, `update_user`, `delete_user`.
• `main.py` یا `app.py`: شامل منطق اصلی برنامه و فراخوانی توابع CRUD.

مثال یک پروژه ساده

اینجوری می‌تونی یه ساختار ساده داشته باشی:

.
├── app
│   ├── __init__.py
│   ├── database.py
│   ├── models.py
│   ├── crud.py
│   └── main.py
└── requirements.txt
        

مدیریت Session

مدیریت درست Session تو SQLAlchemy خیلی مهمه. هر Session یه ارتباط با دیتابیس رو نشون می‌ده. اگه Session‌ها رو باز بذاری، منابع دیتابیس رو هدر می‌دی و ممکنه به مشکلاتی مثل Deadlock بخوری.

• بستن Session: همیشه بعد از اتمام کار، `db.close()` رو فراخوانی کن. الگوی `try…finally` (که تو تابع `get_db` دیدیم) بهترین راهه.
• Scope Session: برای برنامه‌های وب، معمولاً از `scoped_session` استفاده می‌شه تا هر درخواست وب (Request) یه Session مخصوص به خودش داشته باشه و بعد از اتمام درخواست، Session بسته بشه.

هندل کردن خطاها (Error Handling)

دیتابیس می‌تونه خطا بده (مثلاً تکراری بودن ایمیل برای فیلد `unique`). باید این خطاها رو با `try…except` مدیریت کنی. اگه خطایی رخ داد، حتماً `db.rollback()` رو فراخوانی کن تا تغییرات ناقص لغو بشن.

from sqlalchemy.exc import IntegrityError

def create_user_safe(db, name: str, email: str):
    try:
        new_user = User(name=name, email=email)
        db.add(new_user)
        db.commit()
        db.refresh(new_user)
        return new_user
    except IntegrityError:
        db.rollback() # لغو تغییرات در صورت خطا
        print(f"خطا: ایمیل '{email}' قبلاً ثبت شده است.")
        return None
        

Pagination و فیلترینگ پیشرفته

برای برنامه‌هایی که با حجم زیادی از داده‌ها سر و کار دارن، `pagination` (صفحه‌بندی) و فیلترینگ پیشرفته حیاتیه.

• Pagination: با `offset()` و `limit()` (که تو تابع `get_all_users` دیدیم) می‌تونی صفحات رو مدیریت کنی.
• فیلترینگ: از `filter_by()` برای فیلترهای ساده و از `filter()` برای فیلترهای پیچیده‌تر با استفاده از عملگرهای منطقی (مثل `and_`, `or_`) استفاده کن.

بعضی وقت‌ها ممکنه به مشکلاتی بربخوری. اینجا چندتا از رایج‌ترینشون رو با راه حل آوردم.

مشکل ۱: Connection String نامعتبر

اگه دیتابیس وصل نمی‌شه یا خطایی مثل `sqlalchemy.exc.NoSuchModuleError` می‌گیری، ممکنه `Connection String` اشتباه باشه یا درایور دیتابیس رو نصب نکرده باشی.

مشکل ۲: تغییرات ذخیره نمی‌شوند

بعد از اضافه کردن یا به‌روزرسانی یه رکورد، می‌بینی که تغییرات تو دیتابیس اعمال نشدن.

مشکل ۳: Orphaned Sessions / Session is closed

اگه بعد از یه عملیات دیتابیسی، دوباره سعی کنی با همون Session کار کنی و خطای `Session is closed` یا `This Session’s transaction has been rolled back` بگیری.

خب رفیق، تا اینجا دیدیم که SQLAlchemy چطور می‌تونه کار با دیتابیس و عملیات CRUD رو تو پروژه‌های پایتونیت به یه تجربه شیرین و منظم تبدیل کنه. با استفاده از ORM قدرتمندش، می‌تونی با آبجکت‌های پایتونی کار کنی و از پیچیدگی‌های SQL دور بمونی، در حالی که همیشه کنترل کامل روی دیتابیس رو داری.

کدهای آماده و اسنیپت‌هایی که تو این مقاله دیدی، یه شروع عالی هستن برای اینکه بتونی خیلی سریع عملیات CRUD رو تو هر پروژه پایتونی با SQLAlchemy پیاده‌سازی کنی. مدیریت درست Session، هندل کردن خطاها و بهینه‌سازی کوئری‌ها، نکاتی هستن که اگه رعایت کنی، پروژه‌های پایتونیت حسابی حرفه‌ای می‌شن.

امیدوارم این مقاله برات مفید بوده باشه و بتونی با کمک این کدهای آماده CRUD با SQLAlchemy، کارهات رو سریع‌تر و تمیزتر پیش ببری. فراموش نکن که همیشه می‌تونی به فروشگاه ابزارهای برنامه‌نویسی ما سر بزنی و کلی اسنیپت و کد آماده دیگه پیدا کنی. موفق باشی!