ORM — SQLAlchemy দিয়ে ডেটাবেস

১ · সমস্যা: ব্যবসার কোডে ছড়ানো SQL — ORM কী করে

পর্ব ২৮-এ আপনি sqlite3 দিয়ে SQL লিখে ডেটাবেসে ডেটা রেখেছেন, পড়েছেন, transaction দিয়ে নিরাপদ রেখেছেন। পর্ব ৫১-এ FastAPI দিয়ে একটি দোকানের API বানিয়েছেন। এবার দুটি জোড়া লাগাতে গেলে একটি অস্বস্তি টের পাবেন। ধরুন একটি অনলাইন দোকানে ক্রেতা আর অর্ডার আছে। প্রায় প্রতিটি ফাংশনে এমন কোড জমতে থাকে:

def customer_summary(conn, customer_id):
    row = conn.execute("SELECT id, name, district FROM customers WHERE id = ?",
                       (customer_id,)).fetchone()
    orders = conn.execute("SELECT amount FROM orders WHERE customer_id = ?",
                          (customer_id,)).fetchall()
    return {"name": row[1], "district": row[2],      # row[1] মানে কী? মনে রাখতে হয়
            "total": sum(r[0] for r in orders)}

কোডটি ঠিকই কাজ করে, কিন্তু তিনটি সমস্যা লুকিয়ে আছে। প্রথমত, ব্যবসার কথা আর ডেটাবেসের খুঁটিনাটি মিশে গেছে — “ক্রেতার মোট কেনাকাটা” বোঝার আগে পাঠককে টেবিলের নাম, কলামের ক্রম আর row[1]-এর অর্থ মনে রাখতে হয়। দ্বিতীয়ত, একই SQL বারবার লেখা হয়: পঞ্চাশটি ফাংশনে SELECT ... FROM customers, আর একদিন একটি কলাম যোগ হলে পঞ্চাশটি জায়গায় খুঁজে বদলাতে হয়। তৃতীয়ত, দুই জগতের আকার আলাদা: Python-এ একজন ক্রেতা একটি object, যার ভেতরে তাঁর অর্ডারের একটি list; ডেটাবেসে ক্রেতা একটি টেবিলের একটি সারি, আর অর্ডার আরেক টেবিলে, শুধু একটি সংখ্যা (customer_id) দিয়ে জোড়া।

এই তৃতীয় সমস্যাটির একটি নাম আছে: object-relational impedance mismatch. নিচের টেবিলে দেখুন, দুই জগৎ একই জিনিসকে কতটা ভিন্নভাবে দেখে:

একটি ORM (Object-Relational Mapper) এই দুই জগতের মাঝের সেতু। আপনি class লেখেন, object বানান, attribute বদলান; ORM সেই কাজগুলোকে SQL-এ অনুবাদ করে ডেটাবেসে পাঠায়, আর ফেরত আসা সারিগুলোকে আবার object বানিয়ে দেয়।

ORM ভেতরে কী করে, তা বোঝার সবচেয়ে ভালো উপায় নিজে একটি ছোট সংস্করণ বানানো। নিচের প্রোগ্রামে একটি dataclass (পর্ব ৩৪) object থেকে INSERT বানানো হয়েছে, আর SELECT-এর সারি থেকে object. set_trace_callback দিয়ে sqlite3-কে বলেছি, প্রতিটি SQL পাঠানোর আগে ছাপিয়ে দিতে:

import sqlite3
from dataclasses import astuple, dataclass, fields

@dataclass
class Customer:
    name: str
    district: str

def save(conn, obj):                              # object → INSERT
    table, cols = type(obj).__name__.lower() + "s", [f.name for f in fields(obj)]
    marks = ", ".join("?" * len(cols))
    conn.execute(f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({marks})", astuple(obj))
def find(conn, cls, **where):                     # SELECT → objects
    cols = ", ".join(f.name for f in fields(cls))
    cond = " AND ".join(f"{key} = ?" for key in where)
    sql = f"SELECT {cols} FROM {cls.__name__.lower()}s WHERE {cond}"
    return [cls(*row) for row in conn.execute(sql, tuple(where.values()))]
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, district TEXT)")
conn.set_trace_callback(lambda sql: print("SQL>", sql))
for c in [Customer("রহিম", "ঢাকা"), Customer("করিম", "খুলনা")]: save(conn, c)
print(find(conn, Customer, district="ঢাকা"))

২ · SQLAlchemy-র মানচিত্র — engine, connection URL, Core আর ORM

Python-এ সবচেয়ে প্রচলিত ORM হলো SQLAlchemy. Flask আর FastAPI-র প্রজেক্টে এটিই সাধারণ পছন্দ, আর অনেক ডেটা টুলও ভেতরে এটি ব্যবহার করে (যেমন pandas-এর read_sql একটি SQLAlchemy engine নিতে পারে)। এটি ব্যবহার করা যায় SQLite, PostgreSQL, MySQL, Oracle, SQL Server — প্রায় সব বড় ডেটাবেসের সাথে।

python -m venv .venv
.venv\Scripts\activate             # Linux/macOS: source .venv/bin/activate
pip install sqlalchemy
python -c "import sqlalchemy; print(sqlalchemy.__version__)"

SQLAlchemy দুই তলার একটি বাড়ি। নিচের তলা Core: ডেটাবেসের সাথে সংযোগ, টেবিলের বর্ণনা, আর Python দিয়ে SQL বানানোর একটি ভাষা (select(), insert(), text())। ওপরের তলা ORM: class আর object, Session, relationship. ORM নিজের সব কাজ Core দিয়েই করে, আর দরকার হলে আপনি যেকোনো সময় নিচের তলায় নামতে পারেন। পুরো পথটি এমন:

Connection URL — ডেটাবেসের ঠিকানা

Engine বানাতে একটি connection URL লাগে, যা একটি ওয়েব ঠিকানার মতো দেখতে। কয়েকটি উদাহরণ:

import os
from sqlalchemy import create_engine

engine = create_engine("sqlite:///shop.db")        # এই ফোল্ডারের shop.db ফাইল
engine = create_engine("sqlite://")                # শুধু RAM-এ (in-memory)
engine = create_engine("sqlite:///shop.db", echo=True)   # প্রতিটি SQL লগে দেখাবে

# PostgreSQL: dialect+driver://user:password@host:port/database
# password কোডে নয় — environment variable থেকে
engine = create_engine(os.environ["DATABASE_URL"])
# DATABASE_URL=postgresql+psycopg://shop_user:গোপন@localhost:5432/shop

URL-এর গঠন: dialect+driver://ব্যবহারকারী:password@host:port/ডেটাবেস. postgresql+psycopg মানে PostgreSQL-এর dialect আর psycopg (সংস্করণ ৩) driver; driver আলাদা করে pip install "psycopg[binary]" দিয়ে ইনস্টল করতে হয়। SQLite-এর driver sqlite3 Python-এর সাথেই আসে। sqlite:///-এর তিনটি স্ল্যাশ লক্ষ করুন — পরের অংশটি ফাইলের পথ। আর echo=True এই পর্বের সবচেয়ে কাজের বন্ধু: এটি চালু থাকলে SQLAlchemy যা SQL পাঠায়, সব লগে দেখায়। তবে production-এ এটি বন্ধ রাখুন, কারণ লগ ভরে যায় আর তাতে ব্যবহারকারীর ডেটাও চলে আসে।

Engine একবারই বানাবেন, পুরো প্রোগ্রামের জন্য — সাধারণত একটি module-এর শুরুতে। এটি নিজে সংযোগ নয়, বরং সংযোগের একটি কারখানা আর ভাণ্ডার (pool): একটি সংযোগ খুলতে নেটওয়ার্ক আর লগইনের সময় লাগে, তাই engine কিছু সংযোগ খোলা রেখে বারবার ব্যবহার করে। প্রতিটি request-এ নতুন engine বানানো একটি পরিচিত ভুল।

৩ · Model লেখা — DeclarativeBase, Mapped আর mapped_column

এবার দোকানের দুটি টেবিল — ক্রেতা আর অর্ডার — Python class হিসেবে লিখি। SQLAlchemy 2.0-এর ধরনে টেবিলের নকশা লেখা হয় type hint দিয়ে (পর্ব ৪০), যাকে বলে declarative mapping:

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    district: Mapped[str] = mapped_column(String(30))
    orders: Mapped[list["Order"]] = relationship(back_populates="customer")

    def __repr__(self) -> str:
        return f"Customer(id={self.id}, name={self.name!r})"

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    amount: Mapped[int]
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    customer: Mapped["Customer"] = relationship(back_populates="orders")

লাইন ধরে পড়ুন:

• class Base(DeclarativeBase) — সব model-এর একটি সাধারণ মা-class. এর ভেতরে থাকে Base.metadata, যেখানে প্রতিটি টেবিলের বর্ণনা জমা হয়। পর্ব ৩৮-এর __init_subclass__-এর মতো কৌশলে, প্রতিটি subclass লেখার মুহূর্তেই SQLAlchemy সেটি পড়ে টেবিলের নকশা বানায়।
• __tablename__ — ডেটাবেসে টেবিলের নাম।
• Mapped[int] — এই attribute একটি কলাম, আর Python-এ এর type int. Mapped[str] মানে কলামটি NOT NULL; খালি রাখা যায় এমন কলাম চাইলে লিখুন Mapped[str | None]. অর্থাৎ type hint থেকেই SQLAlchemy কলামের type আর NULL হতে পারবে কি না — দুটোই বোঝে। mypy-ও এই hint পড়ে ভুল ধরতে পারে।
• mapped_column(...) — কলামের অতিরিক্ত তথ্য: primary_key=True, String(50) (সর্বোচ্চ দৈর্ঘ্য), ForeignKey("customers.id"), unique=True, index=True, default=.... কিছু না লাগলে (যেমন amount: Mapped[int]) পুরোটাই বাদ দেওয়া যায়।
• relationship(...) — এটি কোনো কলাম নয়। এটি বলে, “customer.orders পড়লে foreign key ধরে অর্ডারগুলো এনে দাও”। অংশ ৫-এ বিস্তারিত।

Model লেখা হলো, কিন্তু ডেটাবেসে এখনো কোনো টেবিল নেই। একটি লাইনে সব টেবিল বানানো যায়:

from sqlalchemy import create_engine
from models import Base

engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)       # যে টেবিল নেই, শুধু সেগুলো বানায়

প্রত্যাশিত আউটপুট (echo=True-এর লগ, সংক্ষেপে):

BEGIN (implicit)
PRAGMA main.table_info("customers")
PRAGMA main.table_info("orders")
CREATE TABLE customers (
	id INTEGER NOT NULL,
	name VARCHAR(50) NOT NULL,
	district VARCHAR(30) NOT NULL,
	PRIMARY KEY (id)
)
CREATE TABLE orders (
	id INTEGER NOT NULL,
	amount INTEGER NOT NULL,
	customer_id INTEGER NOT NULL,
	PRIMARY KEY (id),
	FOREIGN KEY(customer_id) REFERENCES customers (id)
)
COMMIT

প্রথমে PRAGMA table_info দিয়ে দেখে নিল টেবিল আগে থেকে আছে কি না (আসল লগে temp schema-র জন্যও একই প্রশ্ন আছে), তারপর CREATE TABLE. Type hint কীভাবে SQL হলো, মিলিয়ে দেখুন: Mapped[str] + String(50) → VARCHAR(50) NOT NULL; ForeignKey → FOREIGN KEY ... REFERENCES. একটি গুরুত্বপূর্ণ সীমা: create_all শুধু নতুন টেবিল বানায়, আগের টেবিল বদলায় না। পরে Customer-এ একটি phone কলাম যোগ করলে create_all কিছুই করবে না। চলমান ডেটাবেসের নকশা বদলানোর উপায় অংশ ৮-এর Alembic.

৪ · Session আর CRUD — unit of work, commit আর rollback

ORM-এর কেন্দ্রে আছে Session. একে ভাবুন একটি দোকানের হিসাবের খাতার মতো: দিনের মধ্যে আপনি খাতায় লিখে রাখেন — “রহিম নতুন ক্রেতা”, “জয়ার জেলা বদল”, “এই অর্ডার বাতিল”। দোকান বন্ধের সময় একবারে সব হিসাব মূল রেজিস্টারে তোলেন। মাঝপথে বড় ভুল ধরা পড়লে খাতার পাতা ছিঁড়ে ফেলেন — মূল রেজিস্টার যেমন ছিল তেমনই থাকে। Session ঠিক তাই করে: object-এর বদলগুলো জমিয়ে রাখে, আর commit()-এ সব একসাথে একটি transaction-এ ডেটাবেসে পাঠায়। এই নকশার নাম unit of work.

Create — object বানিয়ে add

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from models import Base, Customer, Order

engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    rahim = Customer(name="রহিম", district="ঢাকা")
    rahim.orders = [Order(amount=1200), Order(amount=450)]
    session.add(rahim)
    print("commit-এর আগে id:", rahim.id)
    session.commit()
    print("commit-এর পরে id:", rahim.id)

প্রত্যাশিত আউটপুট (create_all-এর লগ বাদ দিয়ে):

commit-এর আগে id: None
BEGIN (implicit)
INSERT INTO customers (name, district) VALUES (?, ?)
('রহিম', 'ঢাকা')
INSERT INTO orders (amount, customer_id) VALUES (?, ?) RETURNING id
(1200, 1)
INSERT INTO orders (amount, customer_id) VALUES (?, ?) RETURNING id
(450, 1)
COMMIT
BEGIN (implicit)
SELECT customers.id, customers.name, customers.district
FROM customers
WHERE customers.id = ?
(1,)
commit-এর পরে id: 1
ROLLBACK

এই আউটপুটে চারটি শিক্ষা আছে। (১) add()-এর পরেও id None — তখনো কোনো SQL যায়নি। (২) আমরা শুধু rahim-কে add করেছি, কিন্তু অর্ডার দুটিও INSERT হয়েছে — relationship ধরে Session নিজেই খুঁজে নিয়েছে (একে বলে cascade). আর অর্ডারের customer_id-এ 1 বসেছে, যা আগের INSERT-এ ডেটাবেস দিয়েছিল; ক্রমটিও Session ঠিক করেছে — আগে ক্রেতা, পরে অর্ডার। (৩) commit()-এর পরে rahim.id পড়তেই একটি নতুন SELECT গেল। কারণ commit-এর পর Session সব object-কে “মেয়াদোত্তীর্ণ” (expired) চিহ্নিত করে — অন্য কেউ এর মধ্যে ডেটা বদলে থাকতে পারে, তাই পরের বার পড়ার সময় তাজা মান আনে। (৪) শেষের ROLLBACK ভয়ের কিছু নয়: with শেষ হওয়ার সময় খোলা থাকা পড়ার transaction-টি বন্ধ হলো; লেখার কিছু বাকি ছিল না। পুরো প্রক্রিয়াটি ধাপে ধাপে দেখুন:

1with Session(engine) as session:2    karim = Customer(name="করিম", district="খুলনা")3    karim.orders.append(Order(amount=800))4    session.add(karim)5    session.commit()

Read, Update, Delete

ধরে নিই ডেটাবেসে এখন চারজন ক্রেতা: রহিম (ঢাকা, ২টি অর্ডার), করিম (খুলনা, ১টি), সালমা (ঢাকা, ৩টি) আর জয়া (রাজশাহী, কোনো অর্ডার নেই)। পড়ার জন্য SQLAlchemy 2.0-এ একটিই পথ — Core-এর select():

from sqlalchemy import select

with Session(engine) as session:
    stmt = select(Customer).where(Customer.district == "ঢাকা").order_by(Customer.name)
    for c in session.scalars(stmt):              # Read: অনেকগুলো
        print(c)

    joya = session.get(Customer, 4)              # Read: primary key দিয়ে একটি
    joya.district = "বগুড়া"                     # Update: শুধু attribute বদলান
    session.commit()

    session.delete(joya)                         # Delete
    session.commit()

SELECT customers.id, customers.name, customers.district
FROM customers
WHERE customers.district = ? ORDER BY customers.name
('ঢাকা',)
Customer(id=1, name='রহিম')
Customer(id=3, name='সালমা')
SELECT customers.id, customers.name, customers.district FROM customers WHERE customers.id = ?
(4,)
UPDATE customers SET district=? WHERE customers.id = ?
('বগুড়া', 4)
COMMIT
...
DELETE FROM customers WHERE customers.id = ?
(4,)
COMMIT

কয়েকটি বিষয় লক্ষ করুন। Customer.district == "ঢাকা" দেখতে একটি তুলনা, কিন্তু এটি True/False দেয় না — পর্ব ৩২-এর __eq__ overload করে SQLAlchemy একটি SQL expression বানায়, যা WHERE customers.district = ? হয়। মান যায় আলাদা করে (('ঢাকা',)), তাই SQL injection-এর ভয় নেই। session.scalars(stmt) প্রতিটি সারির প্রথম (এখানে একমাত্র) জিনিসটি — Customer object — দেয়; সাথে আছে .all() (list), .first() (প্রথমটি বা None), .one() (ঠিক একটি না হলে exception — একাধিক পেলে MultipleResultsFound) আর .one_or_none(). একাধিক কলাম বা object একসাথে চাইলে session.execute(stmt), যা সারি (row) দেয়।

Update-এ কোনো SQL লিখিনি, শুধু attribute বদলেছি। Session প্রতিটি object-এর আগের মান মনে রাখে, আর flush-এর সময় শুধু বদলানো কলামের জন্য UPDATE ... SET district=? পাঠায়। একই মান আবার বসালে (যেমন "ঢাকা"-কে "ঢাকা") কোনো UPDATE-ই যায় না। session.get()-এর একটি বিশেষ গুণ আছে, যা নিচের stepper-এ দেখবেন।

Rollback আর transaction-এর নিরাপদ ছাঁচ

with Session(engine) as session:
    karim = session.get(Customer, 2)
    karim.district = "ভুল জেলা"
    session.rollback()              # জমানো বদল বাতিল, object expired
    print(karim.district)           # নতুন SELECT → খুলনা

# ছাঁচ: সফল হলে নিজেই commit, exception হলে নিজেই rollback
with Session(engine) as session, session.begin():
    session.add(Customer(name="সালমা", district="ঢাকা"))

প্রত্যাশিত আউটপুট: খুলনা. rollback() শুধু ডেটাবেসের transaction বাতিল করে না, Python object-এর বদলানো মানও ফেলে দেয় — পরের বার পড়লে ডেটাবেসের আসল মান আসে। দ্বিতীয় ছাঁচটি (session.begin()) পর্ব ৩৯-এর context manager-এর সুন্দর ব্যবহার: block সফলভাবে শেষ হলে commit, মাঝে exception উঠলে rollback, আর শেষে Session বন্ধ। বড় প্রজেক্টে সাধারণত SessionLocal = sessionmaker(engine) একবার বানিয়ে প্রতিবার with SessionLocal.begin() as session: লেখা হয়।

Identity map — একই সারি, একই object

Session-এর ভেতরে একটি dict থাকে, যার চাবি “কোন class, কোন primary key”, আর মান সেই object. একে বলে identity map. এর ফল দেখতে অবাক লাগতে পারে:

1a = session.get(Customer, 1)2b = session.scalars(select(Customer)3        .where(Customer.name == "রহিম")).one()4print(a is b)5c = session.get(Customer, 1)

৫ · Relationship — একজন ক্রেতা, অনেক অর্ডার

অংশ ৩-এর model-এ সম্পর্কটি দুই স্তরে লেখা। ডেটাবেসের স্তরে: orders.customer_id একটি foreign key, যা customers.id-কে নির্দেশ করে। Python-এর স্তরে: দুই দিকে দুটি relationship() — Customer.orders (একটি list) আর Order.customer (একটি object). একজন ক্রেতার অনেক অর্ডার, কিন্তু প্রতিটি অর্ডারের একজনই ক্রেতা — এটি one-to-many সম্পর্ক, ডেটাবেসে সবচেয়ে বেশি দেখা ধরন।

back_populates দুই দিককে জুড়ে রাখে: Customer.orders-এ লেখা back_populates="customer" বলে, “উল্টো দিকের নাম Order.customer”. ফলে এক দিক বদলালে অন্য দিক Python-এর ভেতরেই সঙ্গে সঙ্গে ঠিক হয় — rahim.orders.append(o) করলে o.customer is rahim হয়, আবার o.customer = rahim লিখলে rahim.orders-এ o ঢোকে। দুই পাশে নাম ভুল মেলালে SQLAlchemy model প্রথমবার ব্যবহারের সময়ই error দেয়।

এবার মূল প্রশ্ন: customer.orders পড়লে অর্ডারগুলো আসে কখন? স্বাভাবিক নিয়ম হলো lazy loading — যখন প্রথমবার attribute-টি ছোঁয়া হয়, ঠিক তখন:

1c = session.get(Customer, 3)2print(c.name)3amounts = [o.amount for o in c.orders]4again = [o.amount for o in c.orders]5print(c.orders[0].customer is c)

মুছে ফেলার সময় সন্তানদের কী হবে — cascade

রহিমের দুটি অর্ডার আছে। session.delete(rahim) করে commit করলে কী হয়? SQLAlchemy-র স্বাভাবিক আচরণ হলো অর্ডারগুলোর customer_id-কে NULL করে দেওয়া। কিন্তু আমাদের model-এ customer_id: Mapped[int] মানে NOT NULL, তাই আসে IntegrityError: NOT NULL constraint failed: orders.customer_id, আর কিছুই মোছে না। এটি আসলে ভালো খবর — ডেটাবেসের নিয়ম ভুল আটকে দিল। কী চাইছেন, তা স্পষ্ট বলে দিতে হয়। একটি ব্লগে পোস্ট মুছলে তার মন্তব্যগুলো রাখার কোনো মানে নেই:

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    comments: Mapped[list["Comment"]] = relationship(
        back_populates="post", cascade="all, delete-orphan")

class Comment(Base):
    __tablename__ = "comments"
    id: Mapped[int] = mapped_column(primary_key=True)
    body: Mapped[str]
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id"))
    post: Mapped["Post"] = relationship(back_populates="comments")

cascade="all, delete-orphan" মানে: পোস্ট মুছলে তার সব মন্তব্যও মুছবে (cascade delete), আর কোনো মন্তব্যকে পোস্টের list থেকে সরিয়ে দিলে সেটিও মুছবে (সে তখন “এতিম”)। তিনটি মন্তব্যসহ দুটি পোস্টের একটিকে (দুটি মন্তব্যসহ) মুছে গুনলে প্রত্যাশিত আউটপুট 1 — একটি মন্তব্য বাকি। কিন্তু দোকানের অর্ডারে এই নিয়ম বিপজ্জনক: ক্রেতার account মুছলে বিক্রির হিসাবও মুছে যাবে, যা হিসাবরক্ষণ আর আইন — দুই দিক থেকেই সমস্যা। সেখানে বরং ক্রেতাকে “নিষ্ক্রিয়” চিহ্নিত করা হয়। কোন সম্পর্কে cascade চলবে, তা প্রোগ্রামিংয়ের নয়, ব্যবসার সিদ্ধান্ত।

৬ · N+1 সমস্যা — লুপের ভেতরে লুকানো query

Lazy loading একটি অর্ডারের জন্য চমৎকার। কিন্তু একটি তালিকার পেজ ভাবুন: “সব ক্রেতা আর প্রত্যেকের অর্ডার সংখ্যা”। স্বাভাবিকভাবে লিখলে:

with Session(engine) as session:
    for c in session.scalars(select(Customer)):
        print(c.name, len(c.orders))       # প্রতিটি c.orders → একটি নতুন SELECT

কোডটি নিরীহ দেখায়, আর আউটপুটও ঠিক — রহিম 2, করিম 1, সালমা 3, জয়া 0. কিন্তু echo=True লগ খুললে দেখবেন ১টি SELECT ক্রেতাদের জন্য, তারপর প্রত্যেক ক্রেতার জন্য আলাদা একটি SELECT ... FROM orders WHERE ? = orders.customer_id — মোট ৫টি query. ক্রেতা চারজন হলে ৫টি, এক হাজার হলে ১,০০১টি। একে বলে N+1 query problem: তালিকার জন্য ১টি, আর তালিকার N-টি উপাদানের প্রত্যেকটির জন্য ১টি করে। নিজের কম্পিউটারের SQLite-এ প্রতিটি query হয়তো এক মিলিসেকেন্ডেরও কম, তাই টেরই পাবেন না। কিন্তু production-এ ডেটাবেস থাকে আলাদা সার্ভারে, প্রতিটি query-তে নেটওয়ার্কের যাওয়া-আসা কয়েক মিলিসেকেন্ড; ১,০০১ × ২ মিলিসেকেন্ড = দুই সেকেন্ডের বেশি — শুধু একটি পেজের জন্য।

ORM ছাড়াও এই সমস্যা হয়; ORM শুধু এটিকে লুকিয়ে ফেলে। sqlite3 দিয়ে দুটি পদ্ধতির query গুনে দেখি — প্রতিটি SQL একটি list-এ জমা হবে:

import sqlite3

conn = sqlite3.connect(":memory:")
conn.executescript("""
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount INTEGER);
INSERT INTO customers (name) VALUES ('রহিম'), ('করিম'), ('সালমা'), ('জয়া');
INSERT INTO orders (customer_id, amount) VALUES (1, 1200), (1, 450), (2, 800), (3, 300);
""")
queries = []
conn.set_trace_callback(queries.append)          # প্রতিটি SQL গুনে রাখি

customers = conn.execute("SELECT id, name FROM customers").fetchall()
for cid, name in customers:                       # lazy: প্রতি ক্রেতায় একটি query
    conn.execute("SELECT amount FROM orders WHERE customer_id = ?", (cid,)).fetchall()
print("lazy loop  :", len(queries), "queries")
queries.clear()                                   # selectin: সব id একসাথে
ids = [cid for cid, _ in conn.execute("SELECT id, name FROM customers").fetchall()]
marks = ", ".join("?" * len(ids))
rows = conn.execute(f"SELECT customer_id, amount FROM orders WHERE customer_id IN ({marks})", ids).fetchall()
print("IN (...)   :", len(queries), "queries")
print(queries[-1])

সমাধান: আগে থেকে বলে দেওয়া — eager loading

যখন জানেন যে relationship লাগবেই, query-তেই সেটি বলে দিন। একে বলে eager loading, আর SQLAlchemy-তে দুটি প্রধান কৌশল:

1# (ক) lazy loading2for c in session.scalars(select(Customer)):3    print(c.name, len(c.orders))4# (খ) selectinload5stmt = select(Customer).options(6    selectinload(Customer.orders))7for c in session.scalars(stmt):8    print(c.name, len(c.orders))

দ্বিতীয় কৌশল joinedload: একটি মাত্র query-তে LEFT OUTER JOIN দিয়ে ক্রেতা আর অর্ডার একসাথে আনে। প্রত্যাশিত SQL: SELECT customers.id, customers.name, customers.district, orders_1.id AS id_1, ... FROM customers LEFT OUTER JOIN orders AS orders_1 ON customers.id = orders_1.customer_id. JOIN-এ একজন ক্রেতা যতগুলো অর্ডার, ততগুলো সারিতে আসে, তাই collection-এর জন্য joinedload ব্যবহার করলে SQLAlchemy 2.0 ফলের ওপর .unique() ডাকতে বাধ্য করে — না ডাকলে আসে InvalidRequestError: The unique() method must be invoked on this Result...:

from sqlalchemy.orm import joinedload, selectinload

stmt = select(Customer).options(joinedload(Customer.orders))
customers = session.scalars(stmt).unique().all()      # collection-এ unique() বাধ্যতামূলক

stmt = select(Order).options(joinedload(Order.customer))  # অর্ডার → একজন ক্রেতা: joinedload মানানসই

৭ · প্রশ্ন সাজানো — filter, join, group_by আর text()

select() দিয়ে প্রায় পুরো SQL-ই Python-এ লেখা যায়। প্রতিটি method একটি নতুন statement ফেরত দেয়, তাই একটির পর একটি জোড়া যায় — পর্ব ২৮-এর SQL-এর প্রতিটি অংশের একটি করে জোড়া আছে:

জেলাভিত্তিক বিক্রির রিপোর্ট

ব্যবস্থাপক জানতে চান, কোন জেলা থেকে কত অর্ডার আর কত টাকার বিক্রি। এখানে object নয়, শুধু হিসাবের ফল দরকার — তাই select()-এ পুরো class-এর বদলে কলাম আর aggregate function:

from sqlalchemy import func, select

stmt = (
    select(Customer.district, func.count(Order.id), func.sum(Order.amount))
    .join(Customer.orders)
    .group_by(Customer.district)
    .order_by(func.sum(Order.amount).desc())
)
with Session(engine) as session:
    for district, n, total in session.execute(stmt):
        print(f"{district}: {n}টি অর্ডার, মোট {total} টাকা")

SELECT customers.district, count(orders.id) AS count_1, sum(orders.amount) AS sum_1
FROM customers JOIN orders ON customers.id = orders.customer_id
GROUP BY customers.district ORDER BY sum(orders.amount) DESC
ঢাকা: 5টি অর্ডার, মোট 2800 টাকা
খুলনা: 1টি অর্ডার, মোট 800 টাকা

এখানে একটি মাত্র query; হিসাবটি করেছে ডেটাবেস, Python নয়। এটিই ঠিক পথ — সব অর্ডার Python-এ এনে লুপে যোগ করা হতো ধীর আর স্মৃতি-খরচে। session.execute() object নয়, সারি দেয়, যা tuple-এর মতো unpack করা যায়। আর রাজশাহী নেই কেন? জয়ার কোনো অর্ডার নেই, আর সাধারণ JOIN শুধু মিল পাওয়া সারি রাখে। শূন্য অর্ডারের জেলাও দেখাতে চাইলে .outerjoin(Customer.orders) লিখুন — তখন রাজশাহী আসবে 0 অর্ডার নিয়ে (তবে sum হবে None, তাই func.coalesce(func.sum(Order.amount), 0) লাগবে)। GROUP BY-এর পর দলের ওপর শর্ত দিতে .having(func.count(Order.id) >= 2).

একই ছাঁচে ব্লগের রিপোর্ট — প্রতিটি পোস্টে কয়টি মন্তব্য, মন্তব্যহীন পোস্টসহ:

stmt = (select(Post.title, func.count(Comment.id))
        .outerjoin(Post.comments)
        .group_by(Post.id)
        .order_by(func.count(Comment.id).desc(), Post.title))
for title, n in session.execute(stmt):
    print(f"{title}: {n}টি মন্তব্য")

# প্রত্যাশিত আউটপুট:
# Python শেখার রোডম্যাপ: 2টি মন্তব্য
# SQL-এর JOIN: 1টি মন্তব্য
# নতুন পোস্ট: 0টি মন্তব্য

count(Comment.id) (count(*) নয়) গুনেছে বলে মন্তব্যহীন পোস্টে 0 এসেছে — outer join-এ সেই সারির Comment.id NULL, আর COUNT NULL গোনে না।

Raw SQL দরকার হলে — text() আর bound parameter

কখনো এমন SQL লাগে, যা select() দিয়ে লেখা কঠিন বা ডেটাবেস-নির্দিষ্ট। তখন SQLAlchemy Core-এর text() ব্যবহার করুন। মান বসানোর নিয়ম পর্ব ২৮-এর মতোই কঠোর: মান যাবে :নাম placeholder দিয়ে, f-string দিয়ে কখনো নয়।

from sqlalchemy import text

district = "ঢাকা"
rows = session.execute(
    text("SELECT name FROM customers WHERE district = :d ORDER BY name"),
    {"d": district},
)
print([r.name for r in rows])                     # ['রহিম', 'সালমা']

evil = "x' OR '1'='1"                             # দুষ্ট ব্যবহারকারীর input
bad = text(f"SELECT name FROM customers WHERE district = '{evil}'")    # ভুল!
print(len(session.execute(bad).all()))            # 3 — সব ক্রেতা ফাঁস
good = text("SELECT name FROM customers WHERE district = :d")
print(len(session.execute(good, {"d": evil}).all()))   # 0 — নিরাপদ

f-string-এর সংস্করণে দুষ্ট input SQL-এর অংশ হয়ে গেছে: WHERE district = 'x' OR '1'='1' — শর্তটি সবসময় সত্য, তাই সব ক্রেতা বেরিয়ে এল। :d-এর সংস্করণে পুরো লেখাটি শুধু একটি মান, যার সাথে কোনো জেলা মেলে না। মনে রাখবেন, ORM ব্যবহার করলেই নিরাপদ নন: text()-এর ভেতরে f-string ঢোকালে ORM-এর সব সুরক্ষা অকেজো।

অনেকগুলো সারি একসাথে বদলাতে প্রতিটিকে object হিসেবে আনার দরকার নেই; Core-এর update() একটি SQL-এই কাজ সারে: session.execute(update(Order).where(Order.amount < 1000).values(amount=Order.amount + 50)). এটি প্রতিটি সারিকে object হিসেবে আনে না, তাই হাজারো সারিতে অনেক দ্রুত। আর Session-এর ভেতরে আগে থেকে থাকা মিলে যাওয়া object-গুলোর মানও SQLAlchemy 2.0 সাধ্যমতো মিলিয়ে নেয় (synchronize_session-এর স্বাভাবিক আচরণ)।

৮ · আসল প্রজেক্টে — Alembic, FastAPI, টেস্ট, async আর বিকল্প

Alembic — চলমান ডেটাবেসের নকশা বদলানো

অ্যাপ চালু হয়ে গেছে, ডেটাবেসে হাজারো ক্রেতা। এখন ক্রেতার ফোন নম্বর রাখতে হবে। Model-এ phone: Mapped[str | None] = mapped_column(String(15)) যোগ করলেন, কিন্তু অংশ ৩-এ দেখেছেন create_all পুরোনো টেবিল ছোঁয় না। টেবিল মুছে নতুন করে বানানো মানে সব ডেটা হারানো। সমাধান schema migration: নকশার প্রতিটি বদল একটি ছোট, সংস্করণ-নম্বরযুক্ত Python ফাইল, যা git-এ থাকে আর প্রতিটি সার্ভারে একই ক্রমে চলে। SQLAlchemy-র জন্য এই কাজের টুল Alembic, যা SQLAlchemy-র নির্মাতারই লেখা:

pip install alembic
alembic init migrations          # migrations/ ফোল্ডার আর alembic.ini তৈরি
# alembic.ini-তে sqlalchemy.url, আর migrations/env.py-তে
# target_metadata = Base.metadata বসান

alembic revision --autogenerate -m "add phone to customers"
alembic upgrade head             # সব বাকি migration চালাও
alembic downgrade -1             # শেষেরটি উল্টাও
alembic current                  # ডেটাবেস এখন কোন সংস্করণে

--autogenerate আপনার model আর আসল ডেটাবেস মিলিয়ে দেখে পার্থক্যের একটি খসড়া লেখে। আমাদের ক্ষেত্রে এটি জানায় Detected added column 'customers.phone', আর তৈরি ফাইলের ভেতরে থাকে:

def upgrade() -> None:
    op.add_column('customers', sa.Column('phone', sa.String(length=15), nullable=True))

def downgrade() -> None:
    op.drop_column('customers', 'phone')

ফাইলের নামের শুরুর অংশটি (এখানে 585395dd432c) প্রতিবার ভিন্ন একটি এলোমেলো সংস্করণ-চিহ্ন। ডেটাবেসের ভেতরে alembic_version নামের একটি ছোট টেবিল মনে রাখে কোন সংস্করণ পর্যন্ত চলেছে, তাই upgrade head শুধু বাকিগুলো চালায়। খসড়াটি সবসময় পড়ে দেখুন — autogenerate নাম বদলকে প্রায়ই “পুরোনো কলাম মুছে নতুন কলাম যোগ” হিসেবে দেখে, যা ডেটা মুছে ফেলবে। তখন হাতে op.alter_column(..., new_column_name=...) লিখতে হয়। নতুন NOT NULL কলামে পুরোনো সারির জন্য একটি default-ও দিতে হয়।

FastAPI-র সাথে — প্রতিটি request-এ একটি Session

পর্ব ৫১-এ Depends দিয়ে yield করা dependency দেখেছেন। Session-এর জন্য সেটিই আদর্শ: প্রতিটি request নিজের Session পায়, আর কাজ শেষে — সফল হোক বা exception — Session বন্ধ হয়।

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, selectinload
from models import Base, Customer

engine = create_engine("sqlite:///shop.db")       # একবারই, module-এর শুরুতে
Base.metadata.create_all(engine)                  # শেখার জন্য; আসল প্রজেক্টে Alembic
app = FastAPI()

def get_session():
    with Session(engine) as session:
        yield session                             # request শেষে with বন্ধ করবে

SessionDep = Annotated[Session, Depends(get_session)]

class CustomerOut(BaseModel):
    id: int
    name: str
    district: str
    order_count: int

@app.get("/customers/{customer_id}")
def get_customer(customer_id: int, session: SessionDep) -> CustomerOut:
    stmt = (select(Customer).where(Customer.id == customer_id)
            .options(selectinload(Customer.orders)))
    customer = session.scalars(stmt).one_or_none()
    if customer is None:
        raise HTTPException(status_code=404, detail="ক্রেতা পাওয়া যায়নি")
    return CustomerOut(id=customer.id, name=customer.name,
                       district=customer.district, order_count=len(customer.orders))

TestClient দিয়ে যাচাই করা প্রত্যাশিত উত্তর (FastAPI 0.144): একজন ক্রেতা যোগ করার পর GET /customers/1 → 200 {"id": 1, "name": "করিম", "district": "খুলনা", "order_count": 0}; GET /customers/999 → 404 {"detail": "ক্রেতা পাওয়া যায়নি"}. তিনটি নকশা-সিদ্ধান্ত লক্ষ করুন। (১) Handler সাধারণ def, কারণ সাধারণ SQLAlchemy blocking — পর্ব ৫১-এর নিয়মে FastAPI এটিকে thread pool-এ চালায়। (২) Response-এ ORM object সরাসরি না দিয়ে একটি pydantic model; এতে ভেতরের কলাম (যেমন ভবিষ্যতের কোনো গোপন field) ভুল করে বাইরে যায় না, আর response তৈরির সময় লুকানো lazy load-ও ঘটে না। (৩) order_count-এর জন্য selectinload আগে থেকে দেওয়া। তালিকার endpoint-এ এটি না দিলে প্রতিটি request-এ N+1. (শুধু সংখ্যা লাগলে অংশ ৭-এর মতো func.count-এর একটি query আরও ভালো।) SQLModel এই কাজে pydantic model আর table-কে একই class বানায় — নিচের তুলনা দেখুন।

টেস্ট — in-memory SQLite দিয়ে

পর্ব ৪২-এর pytest-এ প্রতিটি টেস্টের জন্য একটি নতুন, খালি ডেটাবেস চাই — আগের টেস্টের ডেটা যেন পরেরটিকে প্রভাবিত না করে। In-memory SQLite এর জন্য নিখুঁত: চোখের পলকে তৈরি, টেস্ট শেষে নিজেই উধাও।

import pytest
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session
from models import Base, Customer, Order

@pytest.fixture
def session():
    engine = create_engine("sqlite://")           # প্রতিটি টেস্টে নতুন ডেটাবেস
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

def test_orders_belong_to_customer(session):
    rahim = Customer(name="রহিম", district="ঢাকা",
                     orders=[Order(amount=1200), Order(amount=450)])
    session.add(rahim)
    session.commit()
    total = session.scalar(select(func.sum(Order.amount))
                           .where(Order.customer_id == rahim.id))
    assert total == 1650

প্রত্যাশিত: 1 passed. FastAPI-র TestClient দিয়ে পুরো API টেস্ট করতে পর্ব ৫১-এর app.dependency_overrides[get_session] দিয়ে আসল Session-এর বদলে in-memory ডেটাবেসের Session বসান। একটি খুঁটিনাটি: TestClient অন্য thread-এ চলে, আর প্রতিটি নতুন সংযোগে in-memory SQLite একটি আলাদা, খালি ডেটাবেস খোলে। তাই সেখানে engine বানাবেন create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False}) দিয়ে — সব জায়গায় একটিই সংযোগ (StaticPool আসে sqlalchemy.pool থেকে)। আরেকটি সতর্কতা: SQLite আর PostgreSQL পুরোপুরি এক নয় (type-এর কঠোরতা, কিছু function). তাই দ্রুত টেস্ট SQLite-এ চালালেও, release-এর আগে আসল ডেটাবেসের বিরুদ্ধে কিছু integration test চালানো ভালো অভ্যাস।

Async SQLAlchemy — এক ঝলক

পুরো অ্যাপ async def-এ লিখলে (পর্ব ৪৫) SQLAlchemy-র async সংস্করণ আছে: from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, URL-এ async driver (sqlite+aiosqlite://, postgresql+asyncpg://), আর প্রতিটি query-তে await: result = await session.scalars(stmt). এর জন্য pip install "sqlalchemy[asyncio]" aiosqlite লাগে। একটি বড় পার্থক্য: async-এ lazy loading চলে না (attribute পড়ার ভেতরে await করা যায় না), তাই relationship সাধারণত selectinload দিয়ে আগেই আনতে হয় (বিকল্প: AsyncAttrs-এর awaitable_attrs) — যা আসলে N+1 থেকে বাঁচার ভালো অভ্যাসও। Async তখনই নিন, যখন সত্যিই অনেক একসাথে অপেক্ষমাণ সংযোগ সামলাতে হয়; নাহলে সাধারণ SQLAlchemy সহজ আর পরিণত।

বিকল্প — আর কখন ORM নয়

কখন ORM ব্যবহার করবেন না: (১) ভারী রিপোর্টিং — window function, CTE, দশটি টেবিলের JOIN — যেখানে SQL-টিই আসল কাজ; ORM-এর ভাষায় অনুবাদ করলে শুধু পড়া কঠিন হয়। (২) লাখো সারি একসাথে বদলানো বা আমদানি — object বানানোর খরচ অপ্রয়োজনীয়; Core-এর bulk insert()/update() বা ডেটাবেসের নিজের টুল (যেমন PostgreSQL-এর COPY) ভালো। (৩) ডেটা বিশ্লেষণ — পর্ব ৪৯-এর pandas-এ read_sql দিয়ে সরাসরি টেবিল আনুন। (৪) ছোট স্ক্রিপ্ট — দুটি টেবিল, পাঁচটি query; sqlite3-ই যথেষ্ট। SQLAlchemy-র সুবিধা হলো এক প্রজেক্টে দুটোই চলে: সাধারণ CRUD-এ ORM, ভারী জায়গায় Core বা text().

৯ · AI-এর যুগে — AI-এর লেখা ডেটাবেস কোড যাচাই করা

“SQLAlchemy দিয়ে customer আর order-এর model আর API লিখে দাও” — AI মুহূর্তে লিখে দেবে। কিন্তু SQLAlchemy-র ইতিহাস দীর্ঘ: ২০০৬ থেকে ২০২৩ পর্যন্ত লেখা লক্ষ লক্ষ টিউটোরিয়াল আর উত্তর মূলত 1.x-এর ধরনে, আর AI সেগুলো থেকেই শিখেছে। ফলে AI প্রায়ই দুই যুগের কোড মিশিয়ে ফেলে, যা চলে, কিন্তু পুরোনো, ধীর বা অনিরাপদ। সবচেয়ে বেশি দেখা ভুলগুলো:

AI-এর লেখা SQLAlchemy কোড পেলে — যাচাইয়ের চেকলিস্ট

• query(, declarative_base, Column( খুঁজেছি (Ctrl+F) — পেলে 2.0-এর ধরনে বদলাতে বলেছি
• echo=True দিয়ে চালিয়ে প্রতিটি পেজ/endpoint-এ মোট query গুনেছি; ডেটা দ্বিগুণ করলে query দ্বিগুণ হচ্ছে না
• প্রতিটি Session কোথায় খোলে আর কোথায় বন্ধ হয়, দেখেছি; লেখার পর commit আছে
• প্রতিটি text() আর raw SQL-এ মান যাচ্ছে শুধু parameter দিয়ে
• cascade আর মুছে ফেলার নিয়ম ব্যবসার সিদ্ধান্তের সাথে মেলে (অর্ডার যেন ভুল করে না মোছে)
• নকশা বদলের জন্য migration ফাইল আছে, আর autogenerate-এর খসড়া নিজে পড়েছি
• টেস্ট নিজে চালিয়েছি — খালি ফল, ভুল id আর rollback-এর পথসহ