Python ও ডেটাবেস: কানেকশন, প্যারামিটার আর ট্রানজ্যাকশন

একটা AI অ্যাপ্লিকেশন প্রায় কখনোই SQL এডিটরের পাশে বসে থাকে না। ট্রেনিং স্ক্রিপ্ট ডেটাবেস থেকে লেবেল-করা উদাহরণ টেনে আনে, চ্যাটবট প্রতিটা প্রম্পট, উত্তর আর টোকেনের হিসাব ডেটাবেসে লিখে রাখে, ফিচার সার্ভিস মডেল স্কোর করার আগে কয়েক মিলিসেকেন্ডে একজন গ্রাহকের ইতিহাস পড়ে নেয়। এর সবটাই Python কোড, যা একটা ড্রাইভার (driver)-এর মাধ্যমে ডেটাবেসের সাথে কথা বলে। এই পাতা সেই কাজটা ঠিকভাবে করা নিয়ে: সঠিক প্যারামিটার, আসল ট্রানজ্যাকশন, গোছানো এরর হ্যান্ডলিং, আর এমন কোড যা টেস্ট করা যায়।

আপনি তিনটা ড্রাইভার ব্যবহার করবেন: sqlite3 (Python-এর সাথেই আসে), MySQL-এর জন্য PyMySQL, আর PostgreSQL-এর জন্য psycopg2 — শেষ দুটো আসল সার্ভারে। pandas আর SQLAlchemy আসবে পরের পাতায়; ওরা যা করে, তার ভিত্তি এই পাতায় শেখা জিনিসগুলোই।

যা শিখবেন

• সব Python ড্রাইভারের একই DB-API গড়ন: connect, cursor, execute, fetch, commit, close।
• প্যারামিটারের সঠিক ব্যবহার (?, :name, %s), মানের লিস্টসহ।
• Python থেকে ট্রানজ্যাকশন: commit, rollback, with con: আর তার ভেতরের লুকানো ফাঁদ।
• executemany দিয়ে একসাথে অনেক সারি ঢোকানো, আর বড় ফলাফল টুকরো টুকরো করে পড়া।
• এনভায়রনমেন্ট ভেরিয়েবল থেকে সেটিং নিয়ে MySQL আর PostgreSQL-এ কানেক্ট করা, কানেকশন পুল, আর ডেটাবেস কোড টেস্ট করা।

এক ইন্টারফেস, অনেক ড্রাইভার: DB-API

Python-এর সব ডেটাবেস ড্রাইভার একটাই স্পেসিফিকেশন মেনে চলে, DB-API 2.0 (PEP 249)। এর গড়নটা একবার শিখে নিলে যেকোনো ডেটাবেস ব্যবহার করতে পারবেন:

connect(...)  ──►  connection  ──►  cursor()  ──►  execute(sql, params)
                      │                              │
                      │                              ├─ fetchone()   one row or None
                      │                              ├─ fetchmany(n) a list of up to n rows
                      │                              └─ fetchall()   every remaining row
                      ├─ commit()     make the changes permanent
                      ├─ rollback()   throw the changes away
                      └─ close()      give the connection back

ড্রাইভারভেদে মূল পার্থক্য হলো প্লেসহোল্ডারের ধরন, আর প্রতিটা ড্রাইভার নিজেই বলে দেয় সে কোনটা ব্যবহার করে:

import sqlite3
import pymysql
import psycopg2

for driver in (sqlite3, pymysql, psycopg2):
    print(f"{driver.__name__:8} apilevel={driver.apilevel} paramstyle={driver.paramstyle}")

sqlite3  apilevel=2.0 paramstyle=qmark
pymysql  apilevel=2.0 paramstyle=pyformat
psycopg2 apilevel=2.0 paramstyle=pyformat

নামগুলো একটু গোলমেলে: psycopg 3 ইমপোর্ট করা হয় psycopg নামে, কোনো সংখ্যা ছাড়া, আর পুরোনো ভার্সনটা হলো psycopg2। নতুন প্রজেক্টে psycopg 3 বেছে নেওয়াই ভালো; এই পাতায় psycopg2 ব্যবহার করা হয়েছে, কারণ এখনকার বেশিরভাগ কোডে সেটাই আছে, আর এখানে শেখা কলগুলো দুটোতেই প্রায় একই রকম।

sqlite3: connect, execute, fetch

import sqlite3

con = sqlite3.connect("shop.db")            # ফাইলটা খোলে (আর না থাকলে নতুন বানিয়ে ফেলে!)
con.execute("PRAGMA foreign_keys = ON")     # SQLite ফরেন কি চেক করে শুধু বলে দিলে

cur = con.cursor()
cur.execute("SELECT product_id, name, price FROM products ORDER BY price DESC")

print([d[0] for d in cur.description])      # কলামের নাম
print(cur.fetchone())                       # পরের সারি, একটা টাপল হিসেবে
print(cur.fetchmany(2))                     # পরের ২টা সারি, একটা লিস্ট হিসেবে
rest = cur.fetchall()                       # বাকি যা আছে সব
print(len(rest), "more rows, the last is", rest[-1])

['product_id', 'name', 'price']
(5, '27-inch Monitor', 18500)
[(7, 'AI Engineering Bootcamp', 15000), (10, 'Noise-Cancelling Headphones', 7500)]
8 more rows, the last is (4, 'Wireless Mouse', 900)

কার্সর (cursor) হলো একটা ফলাফলের ভেতরে একটা অবস্থান। প্রতিবার fetch করলে সেটা সামনে এগোয়, তাই fetchall() ফেরত দিল শুধু বাকি থাকা ৮টা সারি। con.execute(...) একটা শর্টকাট, যা আপনার হয়ে কার্সর বানিয়ে দেয় (sqlite3 আর psycopg 3-এ আছে); PyMySQL আর psycopg2-এ আগে con.cursor() লাগে। কার্সরের ওপর লুপও চালানো যায়: for row in cur: সারিগুলো একটা একটা করে আনে, যা এই পাতাতেই পরে কাজে লাগবে।

টাপল দ্রুত, কিন্তু নামহীন। সারিগুলো যখন অন্য কোডের হাতে দেবেন (বা কোনো API কিংবা LLM প্রম্পটের জন্য JSON বানাবেন), তখন নাম ধরে পড়া অনেক পরিষ্কার। sqlite3-এ এর জন্য আছে sqlite3.Row:

con.row_factory = sqlite3.Row
row = con.execute("SELECT customer_id, name, city FROM customers WHERE customer_id = 5").fetchone()
print(row["name"], row["city"])
print(dict(row))
con.row_factory = None                      # আবার সাধারণ টাপলে ফেরা

Sadia Chowdhury None
{'customer_id': 5, 'name': 'Sadia Chowdhury', 'city': None}

খেয়াল করুন, SQL-এর NULL Python-এ আসে None হয়ে, আর প্যারামিটার হিসেবে None পাঠালে সেটা হয় NULL।

প্যারামিটার: SQL-এর সাথে মান কখনো জোড়া লাগাবেন না

SQL ইনজেকশন আপনি দেখেছেন নিরাপত্তা: রোল, SQL ইনজেকশন, সংবেদনশীল ডেটা আর ব্যাকআপ পাতায়। Python থেকে নিয়মটা সোজা: SQL লেখাটা স্থির থাকবে, মানগুলো আলাদাভাবে প্যারামিটার হয়ে যাবে। ড্রাইভার হয় সেগুলো SQL থেকে আলাদা করে পাঠায় (sqlite3, psycopg 3), নয়তো নিজেই নিরাপদে কোট করে নেয় (PyMySQL, psycopg2), তাই সেগুলো কখনো SQL হিসেবে পড়া হয় না, আর উদ্ধৃতিচিহ্ন, অ্যাপোস্ট্রফি (O'Brien) বা তারিখ নিয়ে কোনো ঝামেলাই থাকে না।

city = "Dhaka"
rows = con.execute(
    "SELECT name FROM customers WHERE city = ? ORDER BY customer_id",
    (city,),                                # একটা টাপল: কমাটা খেয়াল করুন
).fetchall()
print(rows)

rows = con.execute(
    "SELECT name, price FROM products WHERE price BETWEEN :low AND :high ORDER BY price",
    {"low": 2000, "high": 4000},
).fetchall()
print(rows)

ids = [3, 7, 11]
marks = ", ".join("?" * len(ids))           # "?, ?, ?": প্রতিটা মানের জন্য একটা প্লেসহোল্ডার
rows = con.execute(
    f"SELECT product_id, name FROM products WHERE product_id IN ({marks}) ORDER BY product_id",
    ids,
).fetchall()
print(rows)

[('Nadia Rahman',), ('Farhana Akter',), ('Mitu Das',)]
[('USB-C Hub', 2200), ('Hands-On Machine Learning', 2500), ('SQL Masterclass', 3000)]
[(3, 'Mechanical Keyboard'), (7, 'AI Engineering Bootcamp'), (11, 'Gift Card')]

শেষ কোয়েরিতে f-string আছে, আর এখানে সেটা নিরাপদ, কারণ এটা শুধু প্রশ্নবোধক চিহ্ন বসায়, কোনো ডেটা নয়। প্লেসহোল্ডার শুধু মানের জায়গা নিতে পারে। টেবিল বা কলামের নাম প্যারামিটার হতে পারে না। নাম বদলাতে হলে আগে সেটা একটা নির্দিষ্ট তালিকার সাথে মিলিয়ে নিন (if column not in {"price", "stock"}: raise ValueError), তারপর SQL-এ বসান।

ডেটা লেখা, আর কখন তা সত্যি হয়

INSERT, UPDATE আর DELETE চলে SELECT-এর মতোই, কিন্তু commit না করা পর্যন্ত পরিবর্তন স্থায়ী হয় না। ডিফল্ট মোডে Python-এর sqlite3 প্রথম ডেটা-বদলানো স্টেটমেন্টের আগে চুপচাপ একটা ট্রানজ্যাকশন খুলে নেয়। এর মানে কী, তা দেখতে একটা দ্বিতীয় কানেকশন থেকে নজর রাখুন, যেন সেটা অন্য কোনো প্রোগ্রাম:

other = sqlite3.connect("shop.db")          # দ্বিতীয় কানেকশন: ধরুন "অন্য একটা প্রোগ্রাম"

cur = con.execute(
    "INSERT INTO customers (name, email, city, joined_on) VALUES (?, ?, ?, ?)",
    ("Rumana Begum", "rumana@example.com", "Barishal", "2026-06-15"),
)
print("new id:", cur.lastrowid, "| rows:", cur.rowcount, "| in transaction:", con.in_transaction)
print("other sees", other.execute("SELECT COUNT(*) FROM customers").fetchone()[0], "customers")

con.commit()
print("after commit, other sees", other.execute("SELECT COUNT(*) FROM customers").fetchone()[0], "customers")
other.close()

new id: 9 | rows: 1 | in transaction: True
other sees 8 customers
after commit, other sees 9 customers

• lastrowid হলো ডেটাবেসের দেওয়া id; rowcount হলো স্টেটমেন্টটা কয়টা সারি বদলেছে।
• commit-এর আগ পর্যন্ত নতুন গ্রাহকটা ছিল শুধু con-এর ট্রানজ্যাকশনের ভেতরে। প্রোগ্রাম তখন ক্র্যাশ করলে সেটা হারিয়ে যেত।
• rollback() শেষ commit-এর পর থেকে সবকিছু বাতিল করে দেয়:

cur = con.execute("UPDATE products SET price = price + 100 WHERE category_id = ?", (4,))
print(cur.rowcount, "rows changed")
con.rollback()                              # মত বদলালাম
print(con.execute("SELECT name, price FROM products WHERE category_id = 4 ORDER BY product_id").fetchall())

2 rows changed
[('Laptop Stand', 1500), ('USB-C Hub', 2200)]

Python 3.12-এ এসেছে autocommit অ্যাট্রিবিউট (আর্গুমেন্ট হিসেবেও দেওয়া যায়: sqlite3.connect(..., autocommit=False))। autocommit=False দিলে sqlite3 হুবহু DB-API মেনে চলে: একটা ট্রানজ্যাকশন সবসময় খোলা থাকে, SELECT-এর বেলাতেও, আর আপনাকে সবসময় commit বা rollback করতে হয়। autocommit=True দিলে প্রতিটা স্টেটমেন্ট নিজে নিজেই commit হয়, যদি না আপনি নিজে BEGIN লেখেন। ডিফল্ট এখনো এই পাতায় ব্যবহৃত পুরোনো "legacy" মোড (sqlite3.LEGACY_TRANSACTION_CONTROL, যা নিয়ন্ত্রণ করে isolation_level), যেটা ট্রানজ্যাকশন খোলে শুধু INSERT, UPDATE, DELETE বা REPLACE-এর আগে; Python-এর ডকুমেন্টেশন বলছে, ভবিষ্যতের কোনো রিলিজে ডিফল্ট বদলে False হবে। এই পাতার INSERT, UPDATE আর DELETE-এর বেলায় দুটোর আচরণ একই।

ট্রানজ্যাকশন আর এরর হ্যান্ডলিং

একটা অর্ডার মানে একটা অর্ডারের সারি আর তার আইটেমের সারিগুলো। হয় সবগুলো সেভ হবে, নয়তো একটাও না: আধখানা অর্ডার মানে নষ্ট ডেটা। কাজটা try-এর ভেতরে রাখুন, শেষে commit করুন, আর যেকোনো ডেটাবেস এররে rollback করুন:

def place_order(con, customer_id, order_date, items):
    """items: (product_id, quantity)-এর লিস্ট। হয় সব, নয় কিছুই না।"""
    try:
        cur = con.execute(
            "INSERT INTO orders (customer_id, order_date, status) VALUES (?, ?, 'pending')",
            (customer_id, order_date),
        )
        order_id = cur.lastrowid
        for product_id, quantity in items:
            con.execute(
                "INSERT INTO order_items (order_id, product_id, quantity, unit_price) "
                "SELECT ?, product_id, ?, price FROM products WHERE product_id = ?",
                (order_id, quantity, product_id),
            )
        con.commit()
        return order_id
    except sqlite3.Error as exc:
        con.rollback()
        print("rolled back:", type(exc).__name__, "-", exc)
        return None

print(place_order(con, 8, "2026-06-20", [(1, 1), (4, 2)]))
print(place_order(con, 8, "2026-06-21", [(1, 1), (4, 0)]))      # quantity 0 একটা CHECK ভাঙে
print(con.execute("SELECT order_id, customer_id, status FROM orders WHERE order_id > 14").fetchall())

15
rolled back: IntegrityError - CHECK constraint failed: quantity > 0
None
[(15, 8, 'pending')]

দ্বিতীয় কলটা CHECK ব্যর্থ হওয়ার আগে একটা অর্ডার আর একটা আইটেম ঢুকিয়ে ফেলেছিল, আর rollback দুটোই সরিয়ে দিয়েছে: আছে শুধু অর্ডার ১৫। সব Exception না ধরে শুধু sqlite3.Error ধরার মানে হলো, আপনার Python-এর আসল কোনো বাগ, যেমন একটা TypeError, তখনো স্পষ্টভাবে ক্র্যাশ করবে, চাপা পড়বে না।

সব ড্রাইভার একই পরিবারের এক্সেপশন তোলে (প্রত্যেকের নিজস্ব কপি: sqlite3.IntegrityError, pymysql.IntegrityError, psycopg2.IntegrityError), তাই এই প্যাটার্ন সবখানে চলে:

with con: commit করে, কিন্তু বন্ধ করে না

কানেকশন নিজেও একটা কনটেক্সট ম্যানেজার: with con: ব্লক সফল হলে commit করে, এক্সেপশন উঠলে rollback করে। এটা ছোট, আর ভুল করা কঠিন:

try:
    with con:
        con.execute("UPDATE products SET stock = stock - 1 WHERE product_id = 3")
        con.execute("UPDATE products SET stock = stock - 1 WHERE product_id = 9")   # স্টক 0: CHECK ব্যর্থ
except sqlite3.IntegrityError as exc:
    print("IntegrityError:", exc)
print(con.execute("SELECT product_id, stock FROM products WHERE product_id IN (3, 9) ORDER BY product_id").fetchall())

IntegrityError: CHECK constraint failed: stock >= 0
[(3, 12), (9, 0)]

প্রোডাক্ট ৩-এর স্টক এখনো ১২: দ্বিতীয়টার সাথে প্রথম আপডেটটাও rollback হয়েছে। এবার ফাঁদটা। অনেকে ধরে নেন with sqlite3.connect(...) as c: কানেকশন বন্ধ করে দেয়, যেমন with open(...) ফাইল বন্ধ করে। আসলে করে না:

from contextlib import closing

with sqlite3.connect("shop.db") as c:
    print(c.execute("SELECT COUNT(*) FROM products").fetchone())
print(c.execute("SELECT 'still open'").fetchone())       # with ব্লক এটা বন্ধ করেনি
c.close()

with closing(sqlite3.connect("shop.db")) as c:           # closing() শেষে বন্ধ করে
    with c:                                              # আর এটা commit বা rollback করে
        print(c.execute("SELECT COUNT(*) FROM orders").fetchone())

(11,)
('still open',)
(15,)

psycopg2-ও একই আচরণ করে (with conn: ট্রানজ্যাকশন শেষ করে, কানেকশন নয়)। psycopg 3 commit (বা rollback) করে এবং কানেকশন বন্ধ করে; PyMySQL commit না করেই বন্ধ করে দেয়। যে ড্রাইভার ব্যবহার করছেন, তার with কী করে, তা দেখে নিন। একটা কানেকশন খোলা স্ক্রিপ্টে এতে খুব একটা কিছু যায় আসে না। কিন্তু ওয়েব সার্ভার বা লম্বা ট্রেনিং জব যদি প্রতিটা রিকোয়েস্ট বা ব্যাচে একটা করে কানেকশন খোলে, তাহলে বন্ধ না হওয়া কানেকশন জমতে জমতে একসময় সার্ভার নতুন কানেকশন নেওয়াই বন্ধ করে দেয়।

অনেক সারি: executemany আর টুকরো করে পড়া

একটা চ্যাটবট প্রতিটা LLM কলের জন্য একটা লগের সারি লেখে, আর পরে আপনি হাজার হাজার সারি বিশ্লেষণ করেন। ৫,০০০টা নকল কল বানান (seed দেওয়া, তাই সংখ্যাগুলো সবসময় একই) আর একটা ব্যাচে, একটা ট্রানজ্যাকশনে ঢোকান:

import random

con.execute("""
    CREATE TABLE llm_calls (
        call_id       INTEGER PRIMARY KEY,
        model         TEXT    NOT NULL,
        prompt_tokens INTEGER NOT NULL,
        output_tokens INTEGER NOT NULL
    )""")

random.seed(42)
models = ["chat-small", "chat-large", "embed-v1"]
rows = [(random.choice(models), random.randint(50, 2000), random.randint(0, 800)) for _ in range(5000)]

with con:
    con.executemany(
        "INSERT INTO llm_calls (model, prompt_tokens, output_tokens) VALUES (?, ?, ?)", rows
    )
print(con.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0], "rows")

5000 rows

sqlite3-এ executemany স্টেটমেন্টটা একবার তৈরি করে প্রতিটা টাপলের জন্য চালায়। অন্য ড্রাইভার এটাকে নিজের মতো দ্রুত করে: PyMySQL সাধারণ INSERT … VALUES-কে একসাথে অনেক সারির INSERT-এ বদলে নেয়, psycopg 3 স্টেটমেন্টগুলো পাইপলাইনে পাঠায়, কিন্তু psycopg2 শুধু লুপ চালায় (সেখানে গতির জন্য psycopg2.extras.execute_values ব্যবহার করুন)। তবে বড় লাভটা হলো একটাই ট্রানজ্যাকশন: প্রতি সারির পর commit করলে ডেটাবেসকে ৫,০০০ বার ডিস্কে লিখে নিশ্চিত হতে হয়, যা একশো গুণ পর্যন্ত ধীর হতে পারে।

পড়ার বেলায় উল্টো দিক। এক কোটি সারির ওপর fetchall() সবগুলো একসাথে মেমরিতে রাখতে চায়। fetchmany(n) দিয়ে একবারে নির্দিষ্ট আকারের একটা ব্যাচ নিয়ে কাজ করা যায়, আর ছোট একটা জেনারেটর কাজটা আরামদায়ক করে:

def in_batches(cursor, size=1000):
    """ফলাফল শেষ না হওয়া পর্যন্ত সর্বোচ্চ `size`টা সারির লিস্ট দিতে থাকে।"""
    while True:
        batch = cursor.fetchmany(size)
        if not batch:
            return
        yield batch

cur = con.execute("SELECT model, prompt_tokens + output_tokens FROM llm_calls ORDER BY call_id")
totals, batches = {}, 0
for batch in in_batches(cur, 1000):
    batches += 1
    for model, tokens in batch:
        totals[model] = totals.get(model, 0) + tokens
print(batches, "batches")
for model in sorted(totals):
    print(f"{model:11} {totals[model]:>9,} tokens")

5 batches
chat-large  2,400,092 tokens
chat-small  2,282,629 tokens
embed-v1    2,416,957 tokens

এরকম সাধারণ যোগফল SQL-এরই কাজ (GROUP BY model)। টুকরো করে পড়া সেই কাজের জন্য, যা প্রতিটা সারিতে শুধু Python-ই করতে পারে: প্রতিটা প্রোডাক্টের বর্ণনার এমবেডিং হিসাব, ট্রেনিংয়ের আগে লেখা পরিষ্কার করা, সারিগুলো ফাইলে লেখা।

সার্ভারের ড্রাইভারে সাধারণ কার্সর execute()-এর সময়েই পুরো ফলাফল ডাউনলোড করে ফেলে, তাই fetchmany মেমরি বাঁচায় না। সত্যিকারের স্ট্রিমিং চাইলে সার্ভার-সাইড কার্সর লাগবে: psycopg2-এ নাম দেওয়া কার্সর (নিচে দেখানো হয়েছে), PyMySQL-এ pymysql.cursors.SSCursor।

Python থেকে MySQL আর PostgreSQL

সার্ভারে কানেক্ট করতে লাগে হোস্ট, পোর্ট, ইউজার, পাসওয়ার্ড আর ডেটাবেস। এগুলো কনফিগারেশন, কোড নয়: আপনার ল্যাপটপ, টেস্ট সার্ভার আর প্রোডাকশনে এগুলো আলাদা, আর পাসওয়ার্ড কখনো git-এ commit হওয়া চলবে না। প্রচলিত সমাধান হলো এনভায়রনমেন্ট ভেরিয়েবল, যা সেট করে আপনার শেল, python-dotenv দিয়ে লোড করা একটা .env ফাইল, Docker, বা প্রোডাকশনে একটা সিক্রেট ম্যানেজার:

export MYSQL_HOST=127.0.0.1 MYSQL_PORT=3306 MYSQL_USER=root MYSQL_PASSWORD='…' MYSQL_DATABASE=shop
export PGHOST=127.0.0.1 PGPORT=5432 PGUSER=postgres PGPASSWORD='…' PGDATABASE=shop

আগে প্রতিটা সার্ভারে shop.sql লোড করে নিন (বিগিনার টিউটোরিয়ালের সেটআপ: আপনার অনুশীলনের ডেটাবেস পাতা দেখুন)। তারপর PyMySQL দিয়ে MySQL:

import os
import pymysql

my = pymysql.connect(
    host=os.environ["MYSQL_HOST"],
    port=int(os.environ["MYSQL_PORT"]),
    user=os.environ["MYSQL_USER"],
    password=os.environ["MYSQL_PASSWORD"],
    database=os.environ["MYSQL_DATABASE"],
)
with my.cursor() as cur:                     # কার্সর বন্ধ করে, কানেকশন নয়
    cur.execute(
        "SELECT name, joined_on FROM customers WHERE city = %s ORDER BY customer_id",
        ("Dhaka",),
    )
    for name, joined_on in cur.fetchall():
        print(name, repr(joined_on))

    cur.execute("UPDATE products SET price = price * 2 WHERE product_id = %s", (1,))
    print("rows changed:", cur.rowcount)
my.rollback()                                # PyMySQL নিজে commit করে না: পরিবর্তনটা বাতিল

with my.cursor(pymysql.cursors.DictCursor) as cur:      # সারি আসবে ডিকশনারি হয়ে
    cur.execute("SELECT product_id, price FROM products WHERE product_id = %s", (1,))
    print(cur.fetchone())
my.close()

Nadia Rahman datetime.date(2025, 11, 3)
Farhana Akter datetime.date(2026, 1, 9)
Mitu Das datetime.date(2026, 3, 30)
rows changed: 1
{'product_id': 1, 'price': 1200}

SQLite-এর সাথে এখনই দুটো পার্থক্য: প্লেসহোল্ডার %s (সংখ্যার জন্যও সবসময় %s, আর এর চারপাশে কখনো উদ্ধৃতিচিহ্ন দেবেন না), আর তারিখ ফেরত আসে আসল datetime.date অবজেক্ট হয়ে, কারণ MySQL-এ সত্যিকারের DATE টাইপ আছে। SQLite-এ একই কলাম হলো স্ট্রিং '2025-11-03'।

psycopg2 দিয়ে PostgreSQL দেখতে প্রায় একই রকম। সেটিংগুলো একটা ডিকশনারিতে রাখুন, যাতে কানেকশন আর পুল দুটোই সেগুলো ব্যবহার করতে পারে:

import psycopg2
import psycopg2.errors

pg_settings = {
    "host": os.environ["PGHOST"],
    "port": os.environ["PGPORT"],
    "user": os.environ["PGUSER"],
    "password": os.environ["PGPASSWORD"],
    "dbname": os.environ["PGDATABASE"],
}
pg = psycopg2.connect(**pg_settings)

with pg.cursor() as cur:
    cur.execute("SELECT name, price FROM products WHERE price > %(min)s ORDER BY price DESC", {"min": 10000})
    print(cur.fetchall())
    cur.execute("SELECT ROUND(AVG(price), 2) FROM products")
    print(repr(cur.fetchone()[0]))

[('27-inch Monitor', 18500), ('AI Engineering Bootcamp', 15000)]
Decimal('5254.55')

নিখুঁত সংখ্যা (NUMERIC, আর পূর্ণসংখ্যার AVG) আসে decimal.Decimal হয়ে, float নয়, তাই রাউন্ডিংয়ে টাকা হারায় না। json.dumps কিন্তু Decimal সিরিয়ালাইজ করতে পারে না: কোনো সারি API-তে পাঠানো বা LLM প্রম্পটে বসানোর আগে float() বা str() দিয়ে বদলে নিন।

PostgreSQL-এর বাতিল হয়ে যাওয়া ট্রানজ্যাকশন

PostgreSQL-এ একটা স্টেটমেন্ট ব্যর্থ হলে পুরো ট্রানজ্যাকশনটাই নষ্ট হয়ে যায়। rollback না করা পর্যন্ত পরের প্রতিটা স্টেটমেন্ট ফিরিয়ে দেওয়া হয়। যাঁরা এরর ধরে ফেলে কাজ চালিয়ে যান, তাঁরা এতে অবাক হন:

with pg.cursor() as cur:
    try:
        cur.execute("SELECT nme FROM customers")              # টাইপো
    except psycopg2.errors.UndefinedColumn as exc:
        print("error", exc.pgcode, "-", str(exc).splitlines()[0])
    try:
        cur.execute("SELECT 1")
    except psycopg2.errors.InFailedSqlTransaction as exc:
        print("next statement:", str(exc).splitlines()[0])

pg.rollback()                                                 # ব্যর্থ ট্রানজ্যাকশন শেষ করা
with pg.cursor() as cur:
    cur.execute("SELECT COUNT(*) FROM orders")
    print("after rollback:", cur.fetchone())

error 42703 - column "nme" does not exist
next statement: current transaction is aborted, commands ignored until end of transaction block
after rollback: (14,)

psycopg2.errors-এ PostgreSQL-এর প্রতিটা এরর কোডের জন্য একটা করে ক্লাস আছে (UndefinedColumn হলো 42703, UniqueViolation হলো 23505), আর প্রতিটাই স্ট্যান্ডার্ড DB-API ক্লাসগুলোর সাবক্লাস, তাই except psycopg2.IntegrityError একটা UniqueViolation-ও ধরবে।

সার্ভার-সাইড কার্সর দিয়ে স্ট্রিমিং

psycopg2-এর কার্সরকে একটা নাম দিলে ফলাফল সার্ভারেই থাকে; লুপ চলার সাথে সাথে Python প্রতি নেটওয়ার্ক রাউন্ড ট্রিপে itersizeটা করে সারি টেনে আনে:

with pg.cursor(name="orders_stream") as cur:     # নাম দেওয়া = সার্ভার-সাইড
    cur.itersize = 5                             # প্রতি রাউন্ড ট্রিপে কয়টা সারি
    cur.execute("SELECT order_id, status FROM orders ORDER BY order_id")
    seen = 0
    for order_id, status in cur:
        seen += 1
print(seen, "orders streamed, the last was", order_id, status)
pg.rollback()                                    # নাম দেওয়া কার্সর থাকে একটা ট্রানজ্যাকশনের ভেতরে
pg.close()

14 orders streamed, the last was 14 delivered

কানেকশন পুল

সার্ভারে একটা কানেকশন খুলতে লাগে নেটওয়ার্ক হ্যান্ডশেক, অথেনটিকেশন, আর একটা নতুন সার্ভার প্রসেস (PostgreSQL) বা থ্রেড (MySQL): প্রায়ই কয়েক মিলিসেকেন্ড, কখনো তার চেয়ে অনেক বেশি। যে ওয়েব API প্রতিটা রিকোয়েস্টে নতুন কানেকশন খোলে, সে প্রতিটা কলে এই সময় নষ্ট করে, আর চাপ বাড়লে সার্ভারের কানেকশনের সীমা ফুরিয়ে ফেলতে পারে। একটা পুল (pool) কয়েকটা কানেকশন খুলে রাখে আর সেগুলো ধার দেয়:

from psycopg2.pool import SimpleConnectionPool

pool = SimpleConnectionPool(minconn=1, maxconn=4, **pg_settings)

conn = pool.getconn()                    # ধার নেওয়া
try:
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM customers")
        print("customers:", cur.fetchone()[0])
    conn.commit()
finally:
    pool.putconn(conn)                   # সবসময় ফেরত দেওয়া
pool.closeall()

customers: 8

একাধিক থ্রেড একই পুল ব্যবহার করলে ThreadedConnectionPool নিন। (psycopg 3-এর নিজস্ব পুল আছে আলাদা psycopg_pool প্যাকেজে।) বাস্তবে বেশিরভাগ অ্যাপ্লিকেশন পুলিং পায় SQLAlchemy থেকে (পরের পাতা), অথবা PostgreSQL-এর সামনে বসানো PgBouncer-এর মতো আলাদা একটা পুলার থেকে।

ডেটাবেস কোড টেস্ট করা

ডেটাবেসের কোডও কোড, আর অন্য যেকোনো কোডের মতোই এতে বাগ থাকে। সবচেয়ে সস্তা আর নির্ভরযোগ্য টেস্ট ডেটাবেস হলো মেমরিতে SQLite: ":memory:" একটা ব্যক্তিগত ডেটাবেস বানায়, যা কানেকশন বন্ধ হলেই মুছে যায়, তাই প্রতিটা টেস্ট পরিষ্কার অবস্থা থেকে দ্রুত শুরু হয়। যে ফাংশন এটা বানায়, তাকে বলে ফিক্সচার (fixture):

def make_test_db():
    """ফিক্সচার: একটা টেস্টের জন্য মেমরিতে নতুন shop ডেটাবেস।"""
    db = sqlite3.connect(":memory:")
    db.execute("PRAGMA foreign_keys = ON")
    with open("shop.sql", encoding="utf-8") as f:
        db.executescript(f.read())
    return db

def test_good_order_is_saved():
    db = make_test_db()
    order_id = place_order(db, 8, "2026-06-20", [(1, 2)])
    assert db.execute("SELECT quantity, unit_price FROM order_items WHERE order_id = ?",
                      (order_id,)).fetchall() == [(2, 1200)]

def test_bad_order_leaves_nothing_behind():
    db = make_test_db()
    assert place_order(db, 8, "2026-06-21", [(1, 1), (4, 0)]) is None
    assert db.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 14

test_good_order_is_saved()
test_bad_order_leaves_nothing_behind()
print("both tests passed")

rolled back: IntegrityError - CHECK constraint failed: quantity > 0
both tests passed

pytest ব্যবহার করলে টেস্টগুলো নিজে কল করতে হবে না: make_test_db-এর ওপর @pytest.fixture লিখুন, টেস্টের আর্গুমেন্ট হিসেবে db নিন, আর pytest চালান। লজিকের টেস্ট করুন মেমরির SQLite-এ, আর ডায়ালেক্টভেদে আলাদা জিনিসের জন্য (প্লেসহোল্ডার, টাইপ, RETURNING, লকিং) একটা ছোট সেট চালান আসল MySQL বা PostgreSQL-এ (Docker-এ, CI-তে)।

বাস্তবে কোথায় দেখবেন

• LLM অ্যাপ প্রতিটা কল (মডেল, টোকেন, ল্যাটেন্সি, খরচ) প্যারামিটারসহ INSERT দিয়ে লগ করে, ব্যাচে, যাতে লগিংয়ের জন্য ব্যবহারকারীর কাজ ধীর না হয়।
• ট্রেনিং জব লাখ লাখ লেবেল-করা সারি একসাথে লোড না করে সার্ভার-সাইড কার্সর দিয়ে স্ট্রিম করে।
• API আর ফিচার সার্ভিস কানেকশন পুল থেকে রিকোয়েস্ট সামলায়, আর ডিপ্লয়মেন্ট প্ল্যাটফর্ম ক্রেডেনশিয়াল দেয় এনভায়রনমেন্ট ভেরিয়েবল হিসেবে।

সাধারণ ভুল

1. f-string বা + দিয়ে SQL বানানো। এটা ইনজেকশনের দরজা খুলে দেয়, আর প্রথম অ্যাপোস্ট্রফিতেই ভেঙে পড়ে। মান দিন প্যারামিটারে: con.execute("... WHERE name = ?", (name,))।
2. অন্য ড্রাইভারের প্লেসহোল্ডার ব্যবহার। sqlite3-এর কাছে %s-এর কোনো মানে নেই, আর PyMySQL বা psycopg2-এর কাছে ?-এর:
   Pythonকপি

   try:
       con.execute("SELECT name FROM customers WHERE city = %s", ("Dhaka",))
   except sqlite3.OperationalError as exc:
       print("OperationalError:", exc)

   আউটপুটকপি

   OperationalError: near "%": syntax error

3. commit করতে ভুলে যাওয়া। প্রোগ্রাম কোনো এরর ছাড়াই চলে, ভেতরে ডেটাও দেখা যায়, আর প্রোগ্রাম শেষ হলেই সব উধাও। প্রতিটা কাজের এককের চারপাশে with con: দিন, তাহলে commit ভোলার সুযোগই থাকবে না।
4. ভাবা যে with sqlite3.connect(...) কানেকশন বন্ধ করে। এটা শুধু ট্রানজ্যাকশন শেষ করে। contextlib.closing(...)-এ মুড়ে দিন, অথবা একটা finally-তে close() কল করুন।
5. ভুল বানানের ফাইলে কানেক্ট করা। sqlite3.connect("shpo.db") চুপচাপ নতুন একটা খালি ডেটাবেস বানিয়ে ফেলে, আর আপনি পান no such table: customers। আগে থেকে থাকা ফাইল খুলুন sqlite3.connect("file:shop.db?mode=rw", uri=True) দিয়ে, যা ফাইল না থাকলে ব্যর্থ হয়।

নিজে চেষ্টা করুন

1. সহজ: customers_in(con, city) লিখুন, যা প্যারামিটার ব্যবহার করে একটা শহরের গ্রাহকদের নামের লিস্ট customer_id অনুযায়ী সাজিয়ে ফেরত দেয়। "Dhaka" দিয়ে চালিয়ে দেখুন।
2. মাঝারি: restock(con, product_id, amount) লিখুন, যা with con:-এর ভেতরে একটা প্রোডাক্টের স্টকে amount যোগ করে, আর ওই id-র কোনো প্রোডাক্ট না থাকলে ValueError তোলে (ইঙ্গিত: rowcount)। প্রোডাক্ট ৯-এ ১০টা যোগ করুন, তারপর প্রোডাক্ট ৯৯ দিয়ে চেষ্টা করুন।
3. কঠিন: PostgreSQL থেকে সব গ্রাহককে একটা নতুন SQLite টেবিল pg_customers-এ (একই কলাম) কপি করুন: পড়বেন psycopg2 দিয়ে, লিখবেন executemany দিয়ে। তারিখ আসবে datetime.date হয়ে; সেগুলো ISO স্ট্রিং হিসেবে রাখুন। কয়টা সারি কপি হলো আর প্রথম সারিটা প্রিন্ট করুন।

উত্তর

# ১. সহজ
def customers_in(con, city):
    rows = con.execute(
        "SELECT name FROM customers WHERE city = ? ORDER BY customer_id", (city,)
    ).fetchall()
    return [name for (name,) in rows]

print(customers_in(con, "Dhaka"))

# ২. মাঝারি
def restock(con, product_id, amount):
    with con:
        cur = con.execute(
            "UPDATE products SET stock = stock + ? WHERE product_id = ?", (amount, product_id)
        )
        if cur.rowcount == 0:
            raise ValueError(f"no product with id {product_id}")

restock(con, 9, 10)
print(con.execute("SELECT stock FROM products WHERE product_id = 9").fetchone())
try:
    restock(con, 99, 10)
except ValueError as exc:
    print(exc)

# ৩. কঠিন
src = psycopg2.connect(**pg_settings)
try:
    with src.cursor() as cur:
        cur.execute("SELECT customer_id, name, email, city, joined_on FROM customers ORDER BY customer_id")
        rows = [(cid, name, email, city, joined.isoformat()) for cid, name, email, city, joined in cur]
finally:
    src.close()

with con:
    con.execute("CREATE TABLE pg_customers (customer_id INTEGER PRIMARY KEY, name TEXT, "
                "email TEXT, city TEXT, joined_on TEXT)")
    con.executemany("INSERT INTO pg_customers VALUES (?, ?, ?, ?, ?)", rows)
print(len(rows), "rows copied")
print(con.execute("SELECT * FROM pg_customers ORDER BY customer_id").fetchone())

সারসংক্ষেপ

• Python-এর প্রতিটা ড্রাইভার DB-API মেনে চলে: connect, cursor, execute, fetchone/fetchmany/fetchall, commit/rollback, close। মূল পার্থক্য প্লেসহোল্ডারের ধরনে (sqlite3-এ ?, PyMySQL আর psycopg2-তে %s)।
• মান সবসময় প্যারামিটারে; যে নাম বদলাতে হয়, তা যাবে একটা অনুমোদিত তালিকার (allow-list) মধ্য দিয়ে।
• পরিবর্তন সত্যি হয় শুধু commit()-এর পর। with con: আপনার হয়ে commit বা rollback করে, কিন্তু কানেকশন বন্ধ করে না।
• একটা ট্রানজ্যাকশনের ভেতরে executemany দিয়ে ব্যাচে ঢোকান; বড় ফলাফল পড়ুন fetchmany দিয়ে, আর সার্ভারে সার্ভার-সাইড কার্সর দিয়ে।
• সার্ভারের সেটিং আসে এনভায়রনমেন্ট ভেরিয়েবল থেকে; পুল কানেকশন বারবার ব্যবহার করে; মেমরির SQLite ফিক্সচার ডেটাবেস কোডকে টেস্টযোগ্য করে।

এরপর: pandas ও SQLAlchemy: SQL থেকে DataFrame, DataFrame থেকে SQL পাতা এই কানেকশনগুলোর ওপর একটা DataFrame বসায়, ফলে পুরো একটা কোয়েরির ফলাফল এক লাইনেই বিশ্লেষণযোগ্য টেবিল হয়ে যায়; সেই সাথে পরিচয় করিয়ে দেয় SQLAlchemy-র engine, Core আর ORM-এর সাথে।