pandas ও SQLAlchemy: SQL থেকে DataFrame, DataFrame থেকে SQL

ডেটাবেসে শুরু হওয়া বেশিরভাগ ডেটা সায়েন্সের কাজ শেষ হয় একটা DataFrame-এ: দরকারি সারিগুলো কোয়েরি করে আনেন, তারপর pandas-এ সেগুলো ঘেঁটে দেখেন, পরিষ্কার করেন, চার্ট বানান বা মডেল ট্রেন করেন। উল্টো পথটাও সমান সাধারণ: মডেলের প্রেডিকশন, পরিষ্কার করা ডেটাসেট আর মূল্যায়নের স্কোর আবার টেবিলে লিখে রাখা হয়, যাতে ড্যাশবোর্ড আর অন্য সার্ভিস সেগুলো ব্যবহার করতে পারে। pandas দুটো কাজই এক লাইনে করে, আর SQLAlchemy হলো সেই লাইব্রেরি, যার কারণে ওই লাইনগুলো SQLite, MySQL আর PostgreSQL-এ একইভাবে চলে।

এই পাতা Python ও ডেটাবেস: কানেকশন, প্যারামিটার আর ট্রানজ্যাকশন পাতার পরের ধাপ। এখানকার সবকিছু ভেতরে ভেতরে সেই DB-API কানেকশনেই চলে; আপনি শুধু তার ওপরে দুটো স্তর যোগ করছেন।

যা শিখবেন

• প্যারামিটার, তারিখ আর ইনডেক্সসহ pd.read_sql, আর বড় ফলাফল টুকরো টুকরো করে পড়া।
• DataFrame.to_sql: if_exists, index, dtype, chunksize আর method="multi", আর এটা আপনার স্কিমার সাথে কী করে।
• SQLAlchemy-র ইঞ্জিন (engine), এনভায়রনমেন্ট ভেরিয়েবল থেকে বানানো কানেকশন URL, আর ইঞ্জিনের কানেকশন পুল।
• SQLAlchemy Core (text(), select()) আর ORM-এর সাথে প্রথম পরিচয় (মডেল, Session)।
• কখন সরাসরি SQL, কখন Core, ORM বা pandas।

কোয়েরি থেকে DataFrame: read_sql

SQLite-এর বেলায় pandas সাধারণ একটা sqlite3 কানেকশনই নেয়। প্যারামিটার কাজ করে ঠিক আগের পাতার মতো:

import sqlite3
import pandas as pd

con = sqlite3.connect("shop.db")

orders = pd.read_sql(
    "SELECT order_id, customer_id, order_date, status FROM orders WHERE status = ? ORDER BY order_id",
    con,
    params=("delivered",),
    index_col="order_id",            # এই কলামটাই হবে সারির লেবেল
    parse_dates=["order_date"],      # লেখা '2026-01-12' -> আসল datetime
)
print(orders.head(3))
print(orders.dtypes)
print(len(orders), "delivered orders")

          customer_id order_date     status
order_id                                   
1                   1 2026-01-12  delivered
2                   2 2026-01-20  delivered
3                   1 2026-02-03  delivered
customer_id             int64
order_date     datetime64[us]
status                    str
dtype: object
11 delivered orders

SQLite তারিখ রাখে লেখা হিসেবে, তাই parse_dates না দিলে কলামটা হতো str (pandas 3-এর স্ট্রিং dtype), আর তাতে তারিখের হিসাব করা যেত না। লোড করার পর সবসময় dtypes দেখে নিন: বাগ ধরার এর চেয়ে সস্তা উপায় নেই।

কাজ ভাগ করার ভালো নিয়ম হলো: ফিল্টার, জয়েন আর অ্যাগ্রিগেট করবে SQL (ডেটা যেখানে আছে, কাজটা সেখানেই হয়), আর তার পরের কাজ করবে pandas (আকার বদলানো, চার্ট, ফিচার, মডেল):

sql = """
    SELECT c.name AS category, SUM(oi.quantity * oi.unit_price) AS revenue
    FROM order_items oi
    JOIN orders o     ON o.order_id = oi.order_id
    JOIN products p   ON p.product_id = oi.product_id
    JOIN categories c ON c.category_id = p.category_id
    WHERE o.status <> 'cancelled'
    GROUP BY c.name
    ORDER BY revenue DESC
"""
by_category = pd.read_sql(sql, con)
by_category["share_pct"] = (100 * by_category["revenue"] / by_category["revenue"].sum()).round(1)
print(by_category)

      category  revenue  share_pct
0      Courses    36000       42.3
1  Electronics    31550       37.1
2  Accessories     8900       10.5
3        Books     8600       10.1

DataFrame থেকে টেবিল: to_sql

ধরুন একটা churn মডেল প্রত্যেক গ্রাহককে স্কোর দিয়েছে। স্কোরগুলো আবার লিখে রাখতে একটা কলই যথেষ্ট; দরকার হলে এটা টেবিল বানায়, আর কয়টা সারি লেখা হলো তা ফেরত দেয়:

scores = pd.DataFrame({
    "customer_id": [1, 2, 3, 4, 5, 6, 7, 8],
    "churn_score": [0.12, 0.35, 0.08, 0.91, 0.47, 0.22, 0.64, 0.88],
    "model": "churn-v1",
})
print(scores.to_sql("churn_scores", con, index=False))
print(con.execute("SELECT sql FROM sqlite_master WHERE name = 'churn_scores'").fetchone()[0])

try:
    scores.to_sql("churn_scores", con, index=False)          # if_exists="fail" হলো ডিফল্ট
except ValueError as exc:
    print("ValueError:", exc)

8
CREATE TABLE "churn_scores" (
"customer_id" INTEGER,
  "churn_score" REAL,
  "model" TEXT
)
ValueError: Table 'churn_scores' already exists.

pandas যে টেবিল বানাল সেটা দেখুন: কোনো প্রাইমারি কি নেই, NOT NULL নেই, কোনো কনস্ট্রেইন্টই নেই, আর টাইপগুলো DataFrame থেকে আন্দাজ করা। খসড়া টেবিলের জন্য এটা চলে, কিন্তু অন্য কোড যে টেবিলের ওপর নির্ভর করে তার জন্য নয়। আন্দাজ করা টাইপ বদলাতে চাইলে dtype দিন — কলামের নাম থেকে টাইপের একটা dict: ইঞ্জিনের সাথে SQLAlchemy-র টাইপ, যেমন {"model": String(20)}, আর সাধারণ sqlite3 কানেকশনের সাথে টাইপের নাম, যেমন {"model": "TEXT"}। তবু এতে কোনো কি বা কনস্ট্রেইন্ট যোগ হয় না। টেবিল আগে থেকেই থাকলে কী হবে, তা ঠিক করে if_exists:

পেশাদার পদ্ধতি হলো: সঠিক কি আর কনস্ট্রেইন্টসহ টেবিলটা নিজে বানান (ডিজাইনের পাতাগুলোর DDL দিয়ে), তারপর append করুন। তখন ডেটা পাহারা দেয় ডেটাবেস নিজেই, যেমন একই গ্রাহককে দুবার স্কোর করা আটকায়। pandas ড্রাইভারের এক্সেপশনকে pd.errors.DatabaseError-এ মুড়ে দেয়; আসলটা থাকে __cause__-এ:

con.execute("DROP TABLE churn_scores")
con.execute("""
    CREATE TABLE churn_scores (
        customer_id INTEGER PRIMARY KEY,
        churn_score REAL NOT NULL CHECK (churn_score BETWEEN 0 AND 1),
        model       TEXT NOT NULL,
        FOREIGN KEY (customer_id) REFERENCES customers (customer_id)
    )""")
print(scores.to_sql("churn_scores", con, if_exists="append", index=False))
try:
    scores.head(2).to_sql("churn_scores", con, if_exists="append", index=False)
except pd.errors.DatabaseError as exc:                 # pandas ড্রাইভারের এররকে মুড়ে দেয়
    print(type(exc.__cause__).__name__, "-", exc.__cause__)
print(pd.read_sql("SELECT * FROM churn_scores WHERE churn_score > 0.6 ORDER BY customer_id", con))

8
IntegrityError - UNIQUE constraint failed: churn_scores.customer_id
   customer_id  churn_score     model
0            4         0.91  churn-v1
1            7         0.64  churn-v1
2            8         0.88  churn-v1

অনেক সারির বেলায় আরও দুটো অপশন জরুরি। chunksize ততগুলো সারির ব্যাচে লেখে (প্রতিটা ব্যাচ একটা executemany), আর method="multi" অনেকগুলো সারি একটাই INSERT … VALUES (…), (…), … স্টেটমেন্টে বসায়, যা নেটওয়ার্কের ওপারের MySQL বা PostgreSQL-এ সাধারণত অনেক দ্রুত (তবু মেপে দেখুন: কিছু ড্রাইভারে তা হয় না)। SQLite একটা স্টেটমেন্টে ৩২,৭৬৬টার বেশি প্যারামিটার নেয় না, তাই "multi"-এর সাথে chunksize × কলাম এর নিচে রাখুন।

import numpy as np

rng = np.random.default_rng(7)
events = pd.DataFrame({
    "event_id": np.arange(1, 20_001),
    "customer_id": rng.integers(1, 9, 20_000),
    "kind": rng.choice(["view", "search", "add_to_cart", "chat"], 20_000),
    "seconds": rng.integers(1, 600, 20_000),
})
written = events.to_sql("events", con, index=False, chunksize=5_000, method="multi")
print(written, "rows written")

20000 rows written

টুকরো টুকরো করে পড়া

chunksize দিলে read_sql একটা বড় DataFrame-এর বদলে DataFrame-এর একটা ইটারেটর ফেরত দেয়, ফলে মেমরিতে আঁটে না এমন টেবিলও প্রসেস করতে পারেন, আর হাতে রাখেন শুধু ছোট একটা চলতি ফলাফল। sqlite3-এ সারিগুলো সত্যিই টুকরো ধরে ধরে আসে। কিন্তু PostgreSQL আর MySQL-এর ড্রাইভারে ডিফল্ট কার্সর আগে পুরো ফলাফল নামিয়ে আনে, pandas শুধু পরে সেটা ভাগ করে; সত্যিকারের স্ট্রিমিং চাইলে engine.connect().execution_options(stream_results=True) দিয়ে খোলা কানেকশন দিন, যা সার্ভারের দিকের কার্সর ব্যবহার করে।

totals = None
for i, chunk in enumerate(pd.read_sql("SELECT kind, seconds FROM events ORDER BY event_id", con, chunksize=6_000)):
    part = chunk.groupby("kind")["seconds"].agg(["count", "sum"])
    totals = part if totals is None else totals.add(part, fill_value=0)
    print(f"chunk {i}: {len(chunk)} rows")
print(totals.astype(int).sort_index())

chunk 0: 6000 rows
chunk 1: 6000 rows
chunk 2: 6000 rows
chunk 3: 2000 rows
             count      sum
kind                       
add_to_cart   5081  1531748
chat          5020  1521329
search        4945  1496979
view          4954  1492420

এক টুকরো থেকে পরের টুকরোয় যায় শুধু আংশিক যোগফল। যেমন গড় বানাতে হবে শেষে যোগফল আর গণনা থেকে; টুকরোগুলোর আকার আলাদা হলে টুকরোর গড়গুলোর গড় নেওয়া ভুল।

SQLAlchemy: ইঞ্জিন আর URL

SQLite ছাড়া অন্য যেকোনো ডেটাবেসের জন্য pandas চায় একটা SQLAlchemy ইঞ্জিন (engine) (সরাসরি PyMySQL বা psycopg2 কানেকশন দিলেও চলে, তবে সতর্কবার্তা দেয় যে শুধু SQLAlchemy, ADBC আর sqlite3 কানেকশন সমর্থিত)। একটা ইঞ্জিন জানে একটা ডেটাবেসে কীভাবে পৌঁছাতে হয়, কানেকশনের একটা পুল রাখে, আর আপনার কোড আর ওই ডেটাবেসের ডায়ালেক্টের মধ্যে অনুবাদ করে। একে বর্ণনা করা হয় একটা URL দিয়ে:

dialect+driver://username:password@host:port/database

sqlite:///shop.db                               (a file next to the script)
sqlite://                                       (in memory)
mysql+pymysql://app:secret@127.0.0.1:3306/shop
postgresql+psycopg2://app:secret@127.0.0.1:5432/shop

f-string দিয়ে URL জোড়া লাগাবেন না: পাসওয়ার্ডে @, : বা / থাকলে সেটা ভেঙে যায়। URL.create অংশগুলো থেকে URL বানায়, আর অংশগুলো আসে এনভায়রনমেন্ট থেকে:

import os
from sqlalchemy import URL, create_engine, text

sqlite_engine = create_engine("sqlite:///shop.db")

my_engine = create_engine(URL.create(
    "mysql+pymysql",
    username=os.environ["MYSQL_USER"],
    password=os.environ["MYSQL_PASSWORD"],
    host=os.environ["MYSQL_HOST"],
    port=int(os.environ["MYSQL_PORT"]),
    database=os.environ["MYSQL_DATABASE"],
))

pg_engine = create_engine(URL.create(
    "postgresql+psycopg2",
    username=os.environ["PGUSER"],
    password=os.environ["PGPASSWORD"],
    host=os.environ["PGHOST"],
    port=int(os.environ["PGPORT"]),
    database=os.environ["PGDATABASE"],
), pool_size=5, max_overflow=10, pool_pre_ping=True)

for engine in (sqlite_engine, my_engine, pg_engine):
    print(f"{engine.dialect.name:10} driver={engine.driver}")

sqlite     driver=pysqlite
mysql      driver=pymysql
postgresql driver=psycopg2

create_engine তখনই কানেক্ট করে না; করে প্রথম কোয়েরির সময়। প্রোগ্রাম শুরুর সময় প্রতিটা ডেটাবেসের জন্য একটা করে ইঞ্জিন বানিয়ে পুরো প্রোগ্রামে সেটাই ব্যবহার করুন, প্রতি কোয়েরিতে নতুন একটা নয়।

Core: আপনার লেখা SQL, সবখানে একইভাবে চলে

SQLAlchemy Core হলো DB-API-র ঠিক ওপরের স্তর। এর text() আপনার নিজের SQL চালায় :name প্যারামিটার দিয়ে, যেকোনো ডেটাবেসে। SQLAlchemy সেগুলোকে ড্রাইভারের ধরনে (? বা %s) বদলে নেয়, তাই এই একটা ফাংশন তিনটা ইঞ্জিনেই কোনো পরিবর্তন ছাড়া চলে:

def city_customers(engine, city):
    with engine.connect() as conn:                      # পুল থেকে ধার, শেষে ফেরত
        result = conn.execute(
            text("SELECT name, joined_on FROM customers WHERE city = :city ORDER BY customer_id"),
            {"city": city},
        )
        return [(row.name, str(row.joined_on)) for row in result]

for engine in (sqlite_engine, my_engine, pg_engine):
    print(engine.dialect.name, city_customers(engine, "Chattogram"))

sqlite [('Tanvir Ahmed', '2025-12-14')]
mysql [('Tanvir Ahmed', '2025-12-14')]
postgresql [('Tanvir Ahmed', '2025-12-14')]

সারিগুলো নামওয়ালা টাপলের মতো কাজ করে (row.name)। engine.connect()-এর কানেকশন কখনো নিজে থেকে commit করে না: conn.commit() ডাকলে তবেই লেখা টিকে থাকে, আর commit না হওয়া সবকিছু ব্লক শেষে rollback হয়ে যায়। লেখার কাজে engine.begin() আরও সহজ: এটা একটা ট্রানজ্যাকশন খোলে, যা ব্লক শেষে commit করে আর এক্সেপশন উঠলে rollback করে; DB-API-র with con:-এর উল্টো, এখানে কানেকশনটাও পুলে ফেরত যায়:

with pg_engine.begin() as conn:
    conn.execute(text("UPDATE products SET stock = stock + :n WHERE product_id = :id"), {"n": 10, "id": 9})

try:
    with pg_engine.begin() as conn:
        conn.execute(text("UPDATE products SET stock = stock + :n WHERE product_id = :id"), {"n": 5, "id": 9})
        conn.execute(text("UPDATE products SET price = 0 WHERE product_id = 9"))       # CHECK (price > 0) ভাঙে
except Exception as exc:
    print(type(exc).__name__, "- rolled back")

print(pd.read_sql(text("SELECT product_id, stock FROM products WHERE product_id = :id"), pg_engine, params={"id": 9}))

IntegrityError - rolled back
   product_id  stock
0           9     10

স্টক ১০, ১৫ নয়: দ্বিতীয় ব্লকটা পুরোটাই বাতিল হয়েছে। SQLAlchemy ড্রাইভারের এররকে নিজের ক্লাসে মুড়ে দেয় (এখানে sqlalchemy.exc.IntegrityError), যার .orig অ্যাট্রিবিউটে থাকে ড্রাইভারের আসল এক্সেপশন।

একই কোয়েরি, তিনটা ডেটাবেস থেকে

read_sql যেকোনো ইঞ্জিন নেয় বলে একটা বিশ্লেষণ ল্যাপটপের SQLite থেকে কোম্পানির PostgreSQL-এ সরানো মানে এক শব্দের পরিবর্তন। তবে টাইপগুলো অবশ্যই মিলিয়ে নিন:

revenue_sql = text("""
    SELECT o.status, SUM(oi.quantity * oi.unit_price) AS revenue
    FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
    GROUP BY o.status ORDER BY o.status
""")
for engine in (sqlite_engine, my_engine, pg_engine):
    df = pd.read_sql(revenue_sql, engine)
    print(f"{engine.dialect.name:10} total={df['revenue'].sum()}  dtype={df['revenue'].dtype}")

sqlite     total=103550  dtype=int64
mysql      total=103550.0  dtype=float64
postgresql total=103550  dtype=int64

যোগফল একই, টাইপ আলাদা। পূর্ণসংখ্যার ওপর MySQL-এর SUM ফেরত দেয় DECIMAL; ড্রাইভার দেয় Python-এর Decimal অবজেক্ট, আর read_sql সেগুলোকে বানিয়ে ফেলে float64 (এর ডিফল্ট coerce_float=True)। চার্টের জন্য এতে ক্ষতি নেই। কিন্তু টাকা, id বা হুবহু তুলনা করার মতো কিছুর বেলায় ক্ষতি আছে: float সব দশমিক মান (০.১ নিজেই আনুমানিক) বা 2^53-এর বড় সব পূর্ণসংখ্যা নিখুঁতভাবে রাখতে পারে না, ফলে হুবহু তুলনা আর টাকার যোগফল একটু একটু সরে যেতে পারে। SQL-এই কাস্ট করুন (MySQL-এ CAST(SUM(…) AS SIGNED), PostgreSQL-এ ::bigint), অথবা pandas-এ (df["revenue"].astype("int64"))।

select(): Python-এ বানানো কোয়েরি

Core Python অবজেক্ট থেকেও SQL বানাতে পারে। টেবিলটা একবার বর্ণনা করুন (এখানে SQLAlchemy কলামগুলো ডেটাবেস থেকেই পড়ে নেয়: একে বলে রিফ্লেকশন), তারপর কোয়েরি সাজান; প্রতিটা ডায়ালেক্টের জন্য ঠিক SQL লিখে দেয় SQLAlchemy:

from sqlalchemy import MetaData, Table, func, select
from sqlalchemy.dialects import mssql, postgresql

products = Table("products", MetaData(), autoload_with=sqlite_engine)

stmt = (
    select(products.c.name, products.c.price)
    .where(products.c.price >= 3000)
    .order_by(products.c.price.desc())
    .limit(3)
)
with sqlite_engine.connect() as conn:
    print(conn.execute(stmt).all())

print(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
print(stmt.compile(dialect=mssql.dialect(), compile_kwargs={"literal_binds": True}))

[('27-inch Monitor', 18500), ('AI Engineering Bootcamp', 15000), ('Noise-Cancelling Headphones', 7500)]
SELECT products.name, products.price 
FROM products 
WHERE products.price >= 3000 ORDER BY products.price DESC 
 LIMIT 3
SELECT TOP 3 products.name, products.price 
FROM products 
WHERE products.price >= 3000 ORDER BY products.price DESC

একই Python অবজেক্ট PostgreSQL-এর জন্য হলো LIMIT 3, আর SQL Server-এর জন্য TOP 3। চলার সময় টুকরো জুড়ে যখন কোয়েরি বানাতে হয় (সার্চ ফর্ম থেকে ঐচ্ছিক ফিল্টার, অনুমোদিত তালিকা থেকে ব্যবহারকারীর বাছাই করা কলাম), তখন select() সবচেয়ে কাজের, কারণ কখনো স্ট্রিং জোড়া লাগাতে হয় না।

ORM: সারি যখন Python অবজেক্ট

ORM (object-relational mapper) একটা ক্লাসকে একটা টেবিলের সাথে, আর প্রতিটা অবজেক্টকে একটা সারির সাথে মেলায়। মডেলটা একবার ঘোষণা করুন:

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

class Base(DeclarativeBase):
    pass

class Customer(Base):
    __tablename__ = "customers"
    customer_id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(100), unique=True)
    city: Mapped[str | None] = mapped_column(String(50))       # "| None" = NULL হতে পারে
    joined_on: Mapped[date]
    orders: Mapped[list["Order"]] = relationship(back_populates="customer")

class Order(Base):
    __tablename__ = "orders"
    order_id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.customer_id"))
    order_date: Mapped[date]
    status: Mapped[str] = mapped_column(String(20))
    customer: Mapped[Customer] = relationship(back_populates="orders")

with Session(sqlite_engine) as session:
    nadia = session.get(Customer, 1)                                  # প্রাইমারি কি দিয়ে
    print(nadia.name, nadia.joined_on, [(o.order_id, o.status) for o in nadia.orders])

    no_city = session.scalars(select(Customer).where(Customer.city.is_(None))).all()
    print([c.name for c in no_city])

    session.add(Customer(name="Jamal Uddin", email="jamal@example.com",
                         city="Cumilla", joined_on=date(2026, 6, 1)))
    session.commit()                                                  # INSERT এখানেই হয়
    print(session.scalar(select(func.count()).select_from(Customer)), "customers")

Nadia Rahman 2025-11-03 [(1, 'delivered'), (3, 'delivered'), (10, 'delivered')]
['Sadia Chowdhury']
9 customers

• টেবিলগুলো আগে থেকেই আছে, তাই মডেল শুধু সেগুলোর বর্ণনা দেয়। নতুন প্রজেক্টে Base.metadata.create_all(engine) টেবিল বানায়, আর পরে সেগুলো বদলায় একটা মাইগ্রেশন টুল (Alembic)।
• session.scalars(stmt) হলো session.execute(stmt).scalars()-এর সংক্ষিপ্ত রূপ: execute সারি ফেরত দেয় (প্রতিটায় একটা করে Customer), আর scalars() সেগুলো খুলে সরাসরি অবজেক্টগুলো দেয়।
• Mapped[date] SQLite-এর লেখা-তারিখকে আসল date অবজেক্টে বদলায়, তাই একই মডেল সব ডেটাবেসে চলে।
• nadia.orders ছোঁয়ামাত্র আরেকটা কোয়েরি চলেছে (লেজি লোডিং, lazy loading)। ১,০০০ গ্রাহকের ওপর লুপ চালিয়ে প্রত্যেকের .orders ছুঁলে চলে ১,০০১টা কোয়েরি: EXPLAIN আর কোয়েরি অপ্টিমাইজেশন পাতার N+1 সমস্যা। সম্পর্কিত সারিগুলো একবারেই আনুন select(Customer).options(selectinload(Customer.orders)) দিয়ে।

কানেকশন পুল

প্রতিটা ইঞ্জিনের নিজের একটা পুল আছে। engine.connect() একটা কানেকশন ধার নেয়, আর with ব্লক শেষ হলে সেটা খোলা অবস্থাতেই পরের জনের জন্য ফেরত যায়। pg_engine-কে যে সেটিং দিয়েছিলেন, তার মানে:

print(type(pg_engine.pool).__name__, "| size:", pg_engine.pool.size())
print(pg_engine.pool.status())
pg_engine.dispose()           # পুলের সব কানেকশন বন্ধ (প্রোগ্রাম শেষে, বা fork-এর পরে)
my_engine.dispose()

QueuePool | size: 5
Pool size: 5  Connections in pool: 1 Current Overflow: -4 Current Checked out connections: 0

ওপরের কোয়েরিগুলো একটা কানেকশন খুলেছিল, সেটা এখন পুলে অপেক্ষা করছে; overflow −৪ মানে পুলের আকার ৫-এ পৌঁছানোর আগে আরও চারটা খোলা যাবে।

• pool_size=5: সর্বোচ্চ ৫টা কানেকশন খোলা রাখে; max_overflow=10: চাপের সময় আরও ১০টা বাড়তি খুলতে দেয়, পরে আবার বন্ধ করে।
• pool_pre_ping=True: ধার দেওয়ার আগে কানেকশনটা পরীক্ষা করে, তাই সার্ভার বা ফায়ারওয়াল চুপচাপ যে কানেকশন কেটে দিয়েছে, সেটা আপনার কোয়েরি ব্যর্থ না করে বদলে যায়।
• ৫ আর ১০ ডিফল্ট মানও। ফাইলভিত্তিক SQLite ইঞ্জিনও QueuePool পায়, কিন্তু মেমরির ভেতরের ডেটাবেস (sqlite://) পায় SingletonThreadPool, যা প্রতি থ্রেডে একটাই কানেকশন রাখে — কারণ :memory:-এ প্রতিটা নতুন কানেকশন মানে নতুন, খালি একটা ডেটাবেস।
• pool_recycle=1800 (MySQL-এর জন্য যোগ করা ভালো): ৩০ মিনিটের পুরোনো কানেকশন বদলে দেয় (MySQL অলস কানেকশন বন্ধ করে দেয় wait_timeout-এর পর)।

কখন কোন টুল

আসল প্রজেক্টে এগুলো মিলেমিশে থাকে: ORM-ভিত্তিক ওয়েব অ্যাপ, হাতে টিউন করা কয়েকটা রিপোর্টের জন্য Core text(), আর একই ডেটাবেস পড়া নোটবুক ও ট্রেনিং পাইপলাইনে pandas।

সাধারণ ভুল

1. index=False দিতে ভুলে যাওয়া। ডিফল্টভাবে to_sql DataFrame-এর ইনডেক্সকে index নামে একটা বাড়তি কলাম হিসেবে লিখে ফেলে:
   Pythonকপি

   scores.to_sql("scratch_scores", con)
   print([row[1] for row in con.execute("PRAGMA table_info(scratch_scores)")])

   আউটপুটকপি

   ['index', 'customer_id', 'churn_score', 'model']

2. আসল টেবিলে if_exists="replace"। এটা টেবিলটা ড্রপ করে, আর সাথে চলে যায় প্রাইমারি কি, কনস্ট্রেইন্ট, ইনডেক্স আর পারমিশন। DDL দিয়ে টেবিল বানিয়ে append করুন; আগের সারি হালনাগাদ করতে হলে SQL দিয়ে আপসার্ট করুন (ON CONFLICT / ON DUPLICATE KEY UPDATE) — সরাসরি DataFrame-এর সারি থেকে, বা আগে একটা স্টেজিং টেবিলে লোড করে (দেখুন ETL ও ELT: নির্ভরযোগ্য ডেটা পাইপলাইন বানানো)।
3. পুরো টেবিল লোড করে pandas-এ ফিল্টার করা। pd.read_sql("SELECT * FROM events", con)-এর পর df[df.customer_id == 3] মানে নেটওয়ার্ক দিয়ে প্রতিটা সারি টেনে আনা। ফিল্টারটা প্যারামিটারসহ SQL-এ রাখুন।
4. কোয়েরি স্ট্রিংয়ের ভেতরে মান বসানো। read_sql(f"... WHERE city = '{city}'") নিরাপত্তা: রোল, SQL ইনজেকশন, সংবেদনশীল ডেটা আর ব্যাকআপ পাতায় দেখানো সেই একই ইনজেকশনের ফুটো। params ব্যবহার করুন (ইঞ্জিনের বেলায় :name-সহ text())।
5. না দেখে টাইপে ভরসা করা। MySQL-এর যোগফল আসে float64 হয়ে, একটা NULL থাকলে পূর্ণসংখ্যার কলাম হয়ে যায় float64, SQLite-এর তারিখ আসে লেখা হয়ে। প্রতিবার পড়ার পর dtypes প্রিন্ট করুন, আর ফিচার বানানোর আগে কাস্ট করুন।

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

1. সহজ: প্যারামিটারসহ read_sql দিয়ে Electronics প্রোডাক্টগুলো (category_id = 2) দামের ক্রমে একটা DataFrame-এ পড়ুন, product_id হবে ইনডেক্স।
2. মাঝারি: pg_engine আর text() দিয়ে বাতিল নয় এমন অর্ডারের মাসভিত্তিক আয় (to_char(o.order_date, 'YYYY-MM')) পড়ুন, pandas-এ cumsum() দিয়ে একটা running_total কলাম যোগ করুন, আর to_sql দিয়ে ফলাফলটা PostgreSQL-এর নতুন টেবিল monthly_revenue-এ লিখুন। তারপর সারির সংখ্যা পড়ে দেখুন।
3. কঠিন: select() দিয়ে (text নয়) এমন একটা কোয়েরি বানান, যা প্রত্যেক গ্রাহকের নাম আর অর্ডারের সংখ্যা দেয়, যাঁদের কোনো অর্ডার নেই তাঁদেরসহ (outerjoin, group_by), আর যাচাই করুন যে sqlite_engine আর pg_engine-এ একই সারি আসে। customers আর orders-এর জন্য Table(..., autoload_with=...) ব্যবহার করুন।

উত্তর

# ১. সহজ
electronics = pd.read_sql(
    "SELECT product_id, name, price FROM products WHERE category_id = ? ORDER BY price",
    con, params=(2,), index_col="product_id",
)
print(electronics)

# ২. মাঝারি
monthly = pd.read_sql(text("""
    SELECT to_char(o.order_date, 'YYYY-MM') AS month,
           SUM(oi.quantity * oi.unit_price)::bigint AS revenue
    FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
    WHERE o.status <> :cancelled
    GROUP BY month ORDER BY month
"""), pg_engine, params={"cancelled": "cancelled"})
monthly["running_total"] = monthly["revenue"].cumsum()
monthly.to_sql("monthly_revenue", pg_engine, index=False, if_exists="replace")
print(monthly)
print(pd.read_sql(text("SELECT COUNT(*) AS n FROM monthly_revenue"), pg_engine))

# ৩. কঠিন
def orders_per_customer(engine):
    meta = MetaData()
    customers = Table("customers", meta, autoload_with=engine)
    orders_t = Table("orders", meta, autoload_with=engine)
    stmt = (
        select(customers.c.name, func.count(orders_t.c.order_id).label("n_orders"))
        .select_from(customers.outerjoin(orders_t, orders_t.c.customer_id == customers.c.customer_id))
        .where(customers.c.customer_id <= 8)
        .group_by(customers.c.customer_id, customers.c.name)
        .order_by(customers.c.customer_id)
    )
    with engine.connect() as conn:
        return [tuple(row) for row in conn.execute(stmt)]

lite, pg = orders_per_customer(sqlite_engine), orders_per_customer(pg_engine)
print(lite)
print("same on both:", lite == pg)
pg_engine.dispose()

সারসংক্ষেপ

• pd.read_sql(sql, con_or_engine, params=…, parse_dates=…, index_col=…, chunksize=…) একটা কোয়েরিকে DataFrame বানায় (বা DataFrame-এর একটা ইটারেটর)। ফিল্টার আর অ্যাগ্রিগেট করুক SQL; বাকিটা pandas।
• df.to_sql(name, con, if_exists=…, index=False, chunksize=…, method="multi") একটা DataFrame লেখে। গুরুত্বপূর্ণ টেবিল নিজে ডিজাইন করে append করুন; replace টেবিলটাই ড্রপ করে দেয়।
• SQLAlchemy ইঞ্জিন বানানো হয় একটা URL থেকে (এনভায়রনমেন্টের মান দিয়ে URL.create ব্যবহার করুন), এর নিজের কানেকশন পুল থাকে, আর প্রোগ্রামে একবারই বানানো হয়।
• Core text() :name প্যারামিটার দিয়ে আপনার SQL সবখানে চালায়; select() প্রতিটা ডায়ালেক্টের জন্য SQL বানায়; ORM ক্লাসকে টেবিলের সাথে আর অবজেক্টকে সারির সাথে মেলায়।
• সবসময় dtypes দেখুন: SQLite-এর তারিখ লেখা, MySQL-এর যোগফল float হয়ে যায়।

এরপর: pandas দিয়ে ডেটা পরিষ্কার পাতা ইচ্ছে করে এলোমেলো বানানো একটা অর্ডারের ফাইল নিয়ে ধাপে ধাপে সেটাকে এমন ডেটায় পরিণত করে, যার ওপর ভরসা করে মডেল ট্রেন করা যায় — আর তাতে কাজে লাগে সেই DataFrame-গুলো, যা এখন আপনি ডেটাবেসে ঢোকাতে আর বের করতে জানেন।