Databases in Python — sqlite3 and SQLAlchemy with Diagrams

The Story That Explains Databases 📚Databases समझाने वाली कहानी 📚一个讲清数据库的故事 📚

A database is organised storage you can query. Python ships with sqlite3 — a full SQL engine stored in a single file. For bigger work — PostgreSQL, MySQL, an ORM, migrations — most Python developers reach for SQLAlchemy. Together they cover most of what everyday Python programs need.

Database व्यवस्थित storage है जिससे सवाल पूछे जा सकते हैं। Python के साथ sqlite3 आता है — एक file में रखा पूरा SQL engine। बड़े काम के लिए — PostgreSQL, MySQL, ORM, migrations — ज़्यादातर Python developers SQLAlchemy लेते हैं। दोनों मिलकर रोज़ के Python programs की ज़्यादातर ज़रूरतें पूरी करते हैं।

数据库就是 可以查询的有序存储。Python 自带 sqlite3 — 一个存放在单个文件里的完整 SQL 引擎。更大的工作 — PostgreSQL、MySQL、ORM、数据库迁移 — 大多数 Python 开发者会选择 SQLAlchemy。二者合起来能满足日常 Python 程序的大部分需求。

Visual Diagram — How Your Program Reaches the Database 🔌चित्र — आपका Program Database तक कैसे पहुँचता है 🔌图解 — 程序如何到达数据库 🔌

The Three Building Blocks of Any Database 🧱हर Database की तीन ईंटें 🧱任何数据库的三块基石 🧱

Your First Database — Nothing to Installआपका पहला Database — कुछ install नहीं करना你的第一个数据库 — 无需安装

import sqlite3

# Creates the file if it doesn't exist; opens it if it does
conn = sqlite3.connect("bookstore.db")
cur  = conn.cursor()

# Schema - declared ONCE. IF NOT EXISTS makes it safe to re-run.
cur.execute("""
    CREATE TABLE IF NOT EXISTS books (
        id     INTEGER PRIMARY KEY,
        title  TEXT    NOT NULL,
        author TEXT    NOT NULL,
        price  REAL    NOT NULL,
        stock  INTEGER DEFAULT 0
    )
""")

conn.commit()      # save the schema
conn.close()       # always close when done
print("database ready")

INSERT — Adding Rows ➕INSERT — Rows जोड़ना ➕INSERT — 添加行 ➕

Never put values into SQL with f-strings. That is the classic way SQL injection gets in. Use parameterized queries with ? placeholders: the values travel separately from the SQL text, so they can never be read as SQL.

f-strings से values SQL में कभी मत डालिए। SQL injection घुसने का यही पुराना रास्ता है। ? placeholders वाली parameterized queries लीजिए: values SQL text से अलग जाती हैं, इसलिए उन्हें कभी SQL की तरह पढ़ा नहीं जा सकता।

永远不要用 f-string 把值塞进 SQL。这正是 SQL 注入的经典入口。请使用带 ? 占位符的 参数化查询：值与 SQL 文本 分开 传送，因此永远不会被当作 SQL 来解读。

import sqlite3

with sqlite3.connect("bookstore.db") as conn:
    cur = conn.cursor()

    # Single insert - parameterized with ? placeholders
    cur.execute(
        "INSERT INTO books (title, author, price, stock) VALUES (?, ?, ?, ?)",
        ("Dune", "Frank Herbert", 15.99, 12),
    )
    print("new id:", cur.lastrowid)

    # Bulk insert - one call for many rows
    rows = [
        ("1984",            "George Orwell",  9.99,  30),
        ("Brave New World", "Aldous Huxley",  11.50, 18),
        ("Foundation",      "Isaac Asimov",   13.75, 22),
        ("Neuromancer",     "William Gibson", 14.20, 8),
    ]
    cur.executemany(
        "INSERT INTO books (title, author, price, stock) VALUES (?, ?, ?, ?)",
        rows,
    )
    print("executemany rowcount:", cur.rowcount)
    conn.commit()

    total = conn.execute("SELECT COUNT(*) FROM books").fetchone()[0]
    print(f"books in the table: {total}")

import sqlite3

conn = sqlite3.connect(":memory:")            # a throw-away database in RAM
conn.execute("CREATE TABLE users (name TEXT, secret TEXT)")
conn.executemany("INSERT INTO users VALUES (?, ?)",
                 [("Ada", "ada-pin-1111"), ("Grace", "grace-pin-2222")])

name = "x' OR '1'='1"                         # typed into a login form

# WRONG - the input becomes part of the SQL text
sql = f"SELECT name, secret FROM users WHERE name = '{name}'"
print("SQL sent :", sql)
print("f-string :", conn.execute(sql).fetchall())

# RIGHT - the input is sent separately, as a value
print("? param  :", conn.execute("SELECT name, secret FROM users WHERE name = ?",
                                 (name,)).fetchall())

🛡️ Simulator: SQL Injection Lab🛡️ Simulator: SQL Injection Lab🛡️ 模拟器：SQL 注入实验室

Pick what a user types into a login box. You see the SQL an f-string builds (the user's text is in red) and what really came back — next to the safe ? version. Every result is a real run.

चुनिए user login box में क्या लिखता है। आप f-string से बना SQL देखते हैं (user का text लाल में) और असल में क्या लौटा — साथ में सुरक्षित ? version। हर result असली run है।

选择用户在登录框里输入的内容。你会看到 f-string 拼出的 SQL（用户的文字标红）以及真实返回的结果 — 旁边是安全的 ? 版本。每个结果都来自真实运行。

SELECT — Asking Questions ❓SELECT — सवाल पूछना ❓SELECT — 提出问题 ❓

import sqlite3

with sqlite3.connect("bookstore.db") as conn:
    # Row factory: rows behave like dicts as well as tuples
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    # All books
    cur.execute("SELECT id, title, author, price FROM books")
    for row in cur.fetchall():
        print(f"#{row['id']}  {row['title']:20s} £{row['price']:.2f}")

    # With a WHERE filter - parameterized (safe against injection)
    cur.execute("SELECT * FROM books WHERE price < ? AND stock > ?", (14, 0))
    affordable = cur.fetchall()
    print("affordable:", [r["title"] for r in affordable])

    # Aggregate query
    cur.execute("SELECT COUNT(*), AVG(price), MAX(stock) FROM books")
    count, avg_price, max_stock = cur.fetchone()
    print(f"total={count}  avg=£{avg_price:.2f}  most-stocked={max_stock}")

🔎 Simulator: Query Playground🔎 Simulator: Query Playground🔎 模拟器：查询游乐场

Click a query. Rows that match the WHERE stay bright in the table on the left; the result comes in on the right. All results came from running the query on the real bookstore.db built above.

एक query पर click कीजिए। WHERE से मेल खाने वाली rows बाईं table में चमकती रहती हैं; result दाईं ओर आता है। हर result ऊपर बने असली bookstore.db पर query चलाकर आया है।

点击一条查询。符合 WHERE 条件的行在左侧表中保持高亮；结果出现在右侧。所有结果都来自在上面建好的真实 bookstore.db 上运行查询。

UPDATE and DELETE — Changing Data ✏️UPDATE और DELETE — Data बदलना ✏️UPDATE 与 DELETE — 修改数据 ✏️

import sqlite3

with sqlite3.connect("bookstore.db") as conn:
    cur = conn.cursor()

    # Reduce stock when a book is sold - the WHERE clause is essential!
    cur.execute("UPDATE books SET stock = stock - 1 WHERE id = ?", (1,))
    print(f"rows updated: {cur.rowcount}")

    # Price rise on cheap books
    cur.execute("UPDATE books SET price = ROUND(price * 1.1, 2) WHERE price < ?", (12,))
    print(f"prices raised: {cur.rowcount}")

    # Delete out-of-stock, cheap books
    cur.execute("DELETE FROM books WHERE stock = 0 AND price < ?", (10,))
    print(f"rows deleted: {cur.rowcount}")

    conn.commit()
    for row in conn.execute("SELECT id, title, price, stock FROM books"):
        print(row)

cur.rowcount tells you how many rows each statement touched. Here the DELETE matched nothing: no book had stock = 0. Checking rowcount is a cheap way to catch a WHERE that matched less (or more) than you expected.

cur.rowcount बताता है कि हर statement ने कितनी rows छुईं। यहाँ DELETE किसी से मेल नहीं खाया: किसी किताब का stock = 0 नहीं था। rowcount जाँचना उस WHERE को पकड़ने का सस्ता तरीका है जो उम्मीद से कम (या ज़्यादा) rows से मेल खाए।

cur.rowcount 告诉你每条语句影响了多少行。这里的 DELETE 一行也没匹配到：没有哪本书 stock = 0。检查 rowcount 是发现 WHERE 匹配得比预期少（或多）的廉价办法。

Does "with sqlite3.connect()" close the connection?क्या "with sqlite3.connect()" connection बंद करता है?"with sqlite3.connect()" 会关闭连接吗？

import sqlite3
from contextlib import closing

with sqlite3.connect("bookstore.db") as conn:
    conn.execute("UPDATE books SET stock = stock + 1 WHERE id = 1")
# the with-block COMMITTED... but did it close the connection?
print("still open:", conn.execute("SELECT COUNT(*) FROM books").fetchone())

# closing() really closes it; the inner "with conn" handles the transaction
with closing(sqlite3.connect("bookstore.db")) as conn:
    with conn:
        conn.execute("UPDATE books SET stock = stock + 1 WHERE id = 1")
try:
    conn.execute("SELECT 1")
except sqlite3.ProgrammingError as e:
    print("after closing():", e)

Transactions — All-or-Nothing Changes 🔒Transactions — सब या कुछ नहीं 🔒事务 — 要么全成，要么全不成 🔒

A transaction is a group of changes treated as one unit. Either every change lands, or none does. Classic example: moving money from one account to another must change both sides — or neither.

Transaction बदलावों का ऐसा समूह है जिसे एक इकाई माना जाता है। या तो हर बदलाव होता है, या कोई नहीं। पुराना उदाहरण: एक खाते से दूसरे में पैसे भेजने पर दोनों तरफ़ बदलाव होना चाहिए — या किसी भी तरफ़ नहीं।

事务 是被当作一个整体的一组修改。要么全部生效，要么一个都不生效。经典例子：从一个账户转钱到另一个账户，两边都必须改动 — 或者都不改。

import sqlite3

def setup(path="bank.db"):
    with sqlite3.connect(path) as conn:
        conn.executescript("""
            DROP TABLE IF EXISTS accounts;
            CREATE TABLE accounts (
                id      INTEGER PRIMARY KEY,
                owner   TEXT NOT NULL,
                balance REAL NOT NULL CHECK (balance >= 0)   -- no overdrafts
            );
            INSERT INTO accounts VALUES (1, 'Ada', 500), (2, 'Grace', 100);
        """)

def balances():
    with sqlite3.connect("bank.db") as conn:
        return dict(conn.execute("SELECT owner, balance FROM accounts"))

def transfer(from_id: int, to_id: int, amount: float) -> None:
    conn = sqlite3.connect("bank.db")
    try:
        cur = conn.cursor()
        cur.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?",
                    (amount, to_id))
        if cur.rowcount != 1:
            raise ValueError(f"no account {to_id}")
        cur.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?",
                    (amount, from_id))
        conn.commit()               # both updates land together
    except Exception:
        conn.rollback()             # any error -> undo BOTH updates
        raise
    finally:
        conn.close()

setup()
print("start          :", balances())
transfer(1, 2, 100)
print("after 100      :", balances())
for args in [(2, 1, 999), (1, 7, 50)]:
    try:
        transfer(*args)
    except Exception as e:
        print(f"transfer{args} failed: {type(e).__name__}: {e}")
    print("               :", balances())

🏦 Simulator: Transaction Lab🏦 Simulator: Transaction Lab🏦 模拟器：事务实验室

Six transfers, each really run against a fresh bank.db. Watch the statements, then compare the balances. The total money in the bank must never change.

छह transfers, हर एक नए bank.db पर सच में चलाया गया। Statements देखिए, फिर balances मिलाइए। बैंक में कुल पैसा कभी नहीं बदलना चाहिए।

六次转账，每次都在全新的 bank.db 上真实运行。观察执行的语句，再比较余额。银行里的总金额绝不应该改变。

Multiple Tables — Foreign Keys and JOINs 🔗कई Tables — Foreign Keys और JOINs 🔗多张表 — 外键与 JOIN 🔗

Real data has relationships. A customer places orders; an order names a book. We keep them in separate tables and connect them with a foreign key: a column that holds the id of a row in another table.

असली data में रिश्ते होते हैं। ग्राहक orders देता है; order में किताब का नाम होता है। हम उन्हें अलग tables में रखते हैं और foreign key से जोड़ते हैं: ऐसा column जो दूसरी table की किसी row का id रखता है।

真实数据之间有关系。顾客下订单，订单写着书名。我们把它们放在不同的表里，用 外键 连接：一个保存另一张表中某行 id 的列。

import sqlite3

with sqlite3.connect("shop.db") as conn:
    conn.execute("PRAGMA foreign_keys = ON")   # enforce FKs (SQLite has them OFF by default)

    conn.executescript("""
        CREATE TABLE IF NOT EXISTS customers (
            id    INTEGER PRIMARY KEY,
            name  TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL
        );
        CREATE TABLE IF NOT EXISTS orders (
            id          INTEGER PRIMARY KEY,
            customer_id INTEGER NOT NULL,
            book_title  TEXT    NOT NULL,
            qty         INTEGER NOT NULL,
            FOREIGN KEY (customer_id) REFERENCES customers(id)
        );
    """)

    # Insert customers
    conn.execute("INSERT OR IGNORE INTO customers (id, name, email) VALUES (?, ?, ?)",
                 (1, "Ada", "ada@example.com"))
    conn.execute("INSERT OR IGNORE INTO customers (id, name, email) VALUES (?, ?, ?)",
                 (2, "Grace", "grace@example.com"))

    # Insert orders that reference customers
    conn.executemany("INSERT INTO orders (customer_id, book_title, qty) VALUES (?, ?, ?)",
                     [(1, "Dune", 2), (1, "1984", 1), (2, "Foundation", 3)])

    # JOIN - combine rows from both tables
    rows = conn.execute("""
        SELECT c.name, o.book_title, o.qty
        FROM   orders o
        JOIN   customers c ON c.id = o.customer_id
        ORDER BY c.name, o.id
    """).fetchall()

    for name, title, qty in rows:
        print(f"{name:8s}  {title:12s}  x{qty}")

Run it a second time…इसे दूसरी बार चलाइए…再运行一次…

import sqlite3

with sqlite3.connect("shop.db") as conn:
    conn.execute("PRAGMA foreign_keys = ON")   # enforce FKs (SQLite has them OFF by default)

    conn.executescript("""
        CREATE TABLE IF NOT EXISTS customers (
            id    INTEGER PRIMARY KEY,
            name  TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL
        );
        CREATE TABLE IF NOT EXISTS orders (
            id          INTEGER PRIMARY KEY,
            customer_id INTEGER NOT NULL,
            book_title  TEXT    NOT NULL,
            qty         INTEGER NOT NULL,
            FOREIGN KEY (customer_id) REFERENCES customers(id)
        );
    """)

    # Insert customers
    conn.execute("INSERT OR IGNORE INTO customers (id, name, email) VALUES (?, ?, ?)",
                 (1, "Ada", "ada@example.com"))
    conn.execute("INSERT OR IGNORE INTO customers (id, name, email) VALUES (?, ?, ?)",
                 (2, "Grace", "grace@example.com"))

    # Insert orders that reference customers
    conn.executemany("INSERT INTO orders (customer_id, book_title, qty) VALUES (?, ?, ?)",
                     [(1, "Dune", 2), (1, "1984", 1), (2, "Foundation", 3)])

    # JOIN - combine rows from both tables
    rows = conn.execute("""
        SELECT c.name, o.book_title, o.qty
        FROM   orders o
        JOIN   customers c ON c.id = o.customer_id
        ORDER BY c.name, o.id
    """).fetchall()

    for name, title, qty in rows:
        print(f"{name:8s}  {title:12s}  x{qty}")

Foreign keys are OFF by default in SQLiteSQLite में Foreign keys default रूप से OFF हैंSQLite 默认关闭外键约束

import sqlite3

for fk in ("OFF", "ON"):
    conn = sqlite3.connect(":memory:")
    conn.execute(f"PRAGMA foreign_keys = {fk}")
    conn.executescript("""
        CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE orders (id INTEGER PRIMARY KEY,
                             customer_id INTEGER REFERENCES customers(id));
        INSERT INTO customers VALUES (1, 'Ada');
    """)
    try:
        conn.execute("INSERT INTO orders (customer_id) VALUES (99)")   # no customer 99!
        print(f"foreign_keys={fk}: orphan order saved ->",
              conn.execute("SELECT * FROM orders").fetchall())
    except sqlite3.IntegrityError as e:
        print(f"foreign_keys={fk}: IntegrityError: {e}")
    conn.close()

With foreign_keys OFF, SQLite happily saved an order for customer 99, who does not exist. Turn it ON for every connection — it is a per-connection setting, not saved in the file.

foreign_keys OFF होने पर SQLite ने ग्राहक 99 का order ख़ुशी से save कर लिया, जो मौजूद ही नहीं। इसे हर connection के लिए ON कीजिए — यह हर connection की setting है, file में save नहीं होती।

foreign_keys 关闭时，SQLite 照样为并不存在的 99 号顾客保存了订单。要为 每一个 连接都打开它 — 这是连接级的设置，不会保存在文件里。

🔗 Simulator: JOIN Lab🔗 Simulator: JOIN Lab🔗 模拟器：JOIN 实验室

Three customers, three orders — and Linus has no orders. Pick a query and watch which rows get matched. All results are from real queries.

तीन ग्राहक, तीन orders — और Linus का कोई order नहीं। एक query चुनिए और देखिए कौन-सी rows मेल खाती हैं। सारे results असली queries से हैं।

三位顾客，三个订单 — Linus 没有订单。选一条查询，看看哪些行被匹配。所有结果都来自真实查询。

Enter SQLAlchemy — The Toolkit for Bigger Apps 🧰SQLAlchemy — बड़े Apps का Toolkit 🧰走进 SQLAlchemy — 大型应用的工具包 🧰

sqlite3 is perfect for a single script. Bigger applications often need PostgreSQL or MySQL support, connection pooling, schema migrations (with Alembic), an object mapper and a query builder. That is SQLAlchemy — first released in 2006 and today one of the most widely used database toolkits in Python. Flask and FastAPI projects commonly use it (Flask-SQLAlchemy, SQLModel); Django is the main exception, with its own ORM.

sqlite3 एक script के लिए बढ़िया है। बड़े applications को अक्सर PostgreSQL या MySQL support, connection pooling, schema migrations (Alembic से), object mapper और query builder चाहिए। यही SQLAlchemy है — 2006 में पहली बार आया और आज Python के सबसे ज़्यादा इस्तेमाल होने वाले database toolkits में से एक। Flask और FastAPI projects आम तौर पर इसे लेते हैं (Flask-SQLAlchemy, SQLModel); Django मुख्य अपवाद है, जिसका अपना ORM है।

sqlite3 非常适合单个脚本。更大的应用通常需要 PostgreSQL 或 MySQL 支持、连接池、数据库迁移（配合 Alembic）、对象映射器和查询构建器。这就是 SQLAlchemy — 2006 年首次发布，如今是 Python 中使用最广泛的数据库工具包之一。Flask 和 FastAPI 项目常用它（Flask-SQLAlchemy、SQLModel）；Django 是主要例外，它有自己的 ORM。

SetupSetup安装

pip install sqlalchemy

SQLAlchemy Core — SQL, but Composable 🧩SQLAlchemy Core — SQL, पर जोड़ने लायक 🧩SQLAlchemy Core — 可组合的 SQL 🧩

The Core layer lets you build SQL as Python expressions. It is still SQL — you can even print it — but values are always sent as parameters, and a wrong column name fails at once.

Core layer आपको SQL को Python expressions की तरह बनाने देती है। यह अब भी SQL है — आप इसे print भी कर सकते हैं — पर values हमेशा parameters की तरह जाती हैं, और ग़लत column नाम तुरंत fail होता है।

Core 层让你用 Python 表达式构建 SQL。它仍然是 SQL — 甚至可以打印出来 — 但值总是作为参数发送，列名写错会立刻报错。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float, select

# The URL tells SQLAlchemy which database to use:
#   sqlite:///bookstore_core.db         <- SQLite file
#   postgresql+psycopg://user:pw@host/db
#   mysql+pymysql://user:pw@host/db
engine = create_engine("sqlite:///bookstore_core.db", echo=False)

metadata = MetaData()

books = Table(
    "books", metadata,
    Column("id",     Integer, primary_key=True),
    Column("title",  String,  nullable=False),
    Column("author", String,  nullable=False),
    Column("price",  Float,   nullable=False),
    Column("stock",  Integer, default=0),
)

metadata.create_all(engine)   # CREATE TABLE only for tables that don't exist yet

# Insert
with engine.begin() as conn:                # begin() commits on exit
    conn.execute(books.insert(), [
        {"title": "Dune",       "author": "Frank Herbert", "price": 15.99, "stock": 12},
        {"title": "1984",       "author": "George Orwell", "price": 9.99,  "stock": 30},
        {"title": "Foundation", "author": "Isaac Asimov",  "price": 13.75, "stock": 22},
    ])

# SELECT - as a Python expression
with engine.connect() as conn:
    stmt = (select(books.c.title, books.c.price)
            .where(books.c.price < 14)
            .order_by(books.c.price))
    print(stmt)                              # the SQL SQLAlchemy builds
    print()
    for row in conn.execute(stmt):
        print(row.title, row.price)

# A column-name typo fails in Python, before any SQL is sent
try:
    select(books.c.titel)
except AttributeError as e:
    print("\ntypo caught:", type(e).__name__, e)

Look at the printed SQL: books.price < :price_1. The number 14 was not pasted into the text — it travels as a parameter, just like ? in sqlite3. We also gave this example its own file, bookstore_core.db, so it does not add duplicate books to the bookstore.db from Sections 03–06.

Print हुआ SQL देखिए: books.price < :price_1। संख्या 14 text में चिपकाई नहीं गई — यह parameter की तरह जाती है, बिल्कुल sqlite3 के ? जैसे। हमने इस example को अपनी file bookstore_core.db भी दी, ताकि यह Sections 03–06 के bookstore.db में दोहरी किताबें न जोड़े।

看打印出来的 SQL：books.price < :price_1。数字 14 没有被粘贴进文本 — 它作为参数传送，就像 sqlite3 中的 ?。我们还给这个示例单独用了文件 bookstore_core.db，这样它就不会往第 03–06 节的 bookstore.db 里添加重复的书。

SQLAlchemy ORM — Rows Become Python Objects 🧬SQLAlchemy ORM — Rows बनती हैं Python Objects 🧬SQLAlchemy ORM — 行变成 Python 对象 🧬

The ORM (Object-Relational Mapper) lets you treat rows as normal Python objects. Each class is a table. Each instance is a row. The Session tracks your changes and writes them to the database when you commit.

ORM (Object-Relational Mapper) आपको rows को सामान्य Python objects की तरह इस्तेमाल करने देता है। हर class एक table है। हर instance एक row है। Session आपके बदलाव याद रखता है और commit करने पर database में लिखता है।

ORM（对象关系映射器）让你把行当作普通的 Python 对象来使用。每个类是一张表，每个实例是一行。Session 会跟踪你的修改，并在提交时写入数据库。

from sqlalchemy import create_engine, String, Integer, Float, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# 1) Base class for all models
class Base(DeclarativeBase):
    pass

# 2) A model - one class = one table
class Book(Base):
    __tablename__ = "books"

    id:     Mapped[int]   = mapped_column(Integer, primary_key=True)
    title:  Mapped[str]   = mapped_column(String(200), nullable=False)
    author: Mapped[str]   = mapped_column(String(100), nullable=False)
    price:  Mapped[float] = mapped_column(Float, nullable=False)
    stock:  Mapped[int]   = mapped_column(Integer, default=0)

    def __repr__(self) -> str:
        return f"Book({self.title!r}, £{self.price})"

# 3) Create the engine and tables
engine = create_engine("sqlite:///bookstore_orm.db")
Base.metadata.create_all(engine)

# 4) Insert - just create Python objects
with Session(engine) as session:
    session.add_all([
        Book(title="Dune",       author="Frank Herbert", price=15.99, stock=12),
        Book(title="1984",       author="George Orwell", price=9.99,  stock=30),
        Book(title="Foundation", author="Isaac Asimov",  price=13.75, stock=22),
    ])
    session.commit()

# 5) Query - Python style
with Session(engine) as session:
    stmt = select(Book).where(Book.price < 14).order_by(Book.price)
    for book in session.scalars(stmt):
        print(book)                      # uses __repr__

    # Update - change the object, then commit
    dune = session.scalar(select(Book).where(Book.title == "Dune"))
    dune.stock -= 1                      # plain Python attribute assignment
    session.commit()                     # SQL UPDATE runs automatically
    print("Dune stock now:", dune.stock)

Relationships — One-to-Many Made Easy 👨‍👩‍👧Relationships — One-to-Many आसान 👨‍👩‍👧关系 — 轻松实现一对多 👨‍👩‍👧

Two related tables — users and their posts — connect naturally in the ORM. You can move in both directions: user.posts gives a list; post.author gives the user.

दो जुड़ी tables — users और उनके posts — ORM में स्वाभाविक रूप से जुड़ती हैं। आप दोनों दिशाओं में जा सकते हैं: user.posts एक list देता है; post.author user देता है।

两张相关的表 — 用户和他们的帖子 — 在 ORM 中自然地连接起来。你可以双向访问：user.posts 给出列表；post.author 给出用户。

from sqlalchemy import ForeignKey, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

    # A User has many Posts - this line creates the collection
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id:        Mapped[int] = mapped_column(primary_key=True)
    title:     Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # Back-reference - post.author gives you the User
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite:///blog.db")
Base.metadata.create_all(engine)

# Create a user with posts, all in one go
with Session(engine) as s:
    ada = User(name="Ada", email="ada@x.io", posts=[
        Post(title="Hello, world"),
        Post(title="Learning SQLAlchemy"),
        Post(title="Why relationships rule"),
    ])
    s.add(ada)
    s.commit()

# Navigate BOTH ways - no SQL written by you
with Session(engine) as s:
    ada = s.scalar(select(User).where(User.name == "Ada"))
    print(f"{ada.name} has {len(ada.posts)} posts:")
    for p in ada.posts:
        print(f"  - {p.title:24s} (by {p.author.name})")

Run it a second time…इसे दूसरी बार चलाइए…再运行一次…

from sqlalchemy import ForeignKey, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

    # A User has many Posts - this line creates the collection
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id:        Mapped[int] = mapped_column(primary_key=True)
    title:     Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # Back-reference - post.author gives you the User
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite:///blog.db")
Base.metadata.create_all(engine)

# Create a user with posts, all in one go
with Session(engine) as s:
    ada = User(name="Ada", email="ada@x.io", posts=[
        Post(title="Hello, world"),
        Post(title="Learning SQLAlchemy"),
        Post(title="Why relationships rule"),
    ])
    s.add(ada)
    s.commit()

# Navigate BOTH ways - no SQL written by you
with Session(engine) as s:
    ada = s.scalar(select(User).where(User.name == "Ada"))
    print(f"{ada.name} has {len(ada.posts)} posts:")
    for p in ada.posts:
        print(f"  - {p.title:24s} (by {p.author.name})")

Lazy loading and the N+1 problemLazy loading और N+1 समस्या延迟加载与 N+1 问题

By default user.posts is loaded lazily: the first time you touch it, the ORM runs one more query. Loop over 20 users and touch each user's posts, and you run 1 + 20 queries. That is the N+1 problem. selectinload() fetches all the posts in one extra query.

Default रूप से user.posts lazy load होता है: पहली बार छूने पर ORM एक और query चलाता है। 20 users पर loop चलाइए और हर user के posts छुइए, तो 1 + 20 queries चलती हैं। यही N+1 समस्या है। selectinload() सारे posts एक अतिरिक्त query में ले आता है।

默认情况下 user.posts 是 延迟 加载的：第一次访问时，ORM 会再执行一条查询。遍历 20 个用户并访问每个人的帖子，就会执行 1 + 20 条查询。这就是 N+1 问题。selectinload() 只需一条额外查询就能取回所有帖子。

from sqlalchemy import ForeignKey, create_engine, select, event
from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column, relationship,
                            Session, selectinload)

class Base(DeclarativeBase): pass
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite://")                 # in-memory database
Base.metadata.create_all(engine)
with Session(engine) as s:
    s.add_all([User(name=f"user{i}", posts=[Post(title="p1"), Post(title="p2")])
               for i in range(20)])
    s.commit()

queries = []
event.listen(engine, "before_cursor_execute",
             lambda conn, cur, sql, *a: queries.append(sql))

for label, stmt in [("lazy (default)", select(User)),
                    ("selectinload  ", select(User).options(selectinload(User.posts)))]:
    queries.clear()
    with Session(engine) as s:
        total = sum(len(u.posts) for u in s.scalars(stmt))
    print(f"{label}: {total} posts loaded with {len(queries)} SQL queries")

🔢 Simulator: N+1 Query Counter🔢 Simulator: N+1 Query Counter🔢 模拟器：N+1 查询计数器

Each dot is one SQL query that SQLAlchemy really sent (counted with an event listener). Change the number of users and the loading strategy.

हर बिंदु एक SQL query है जो SQLAlchemy ने सच में भेजी (event listener से गिनी गई)। Users की संख्या और loading strategy बदलिए।

每个圆点代表 SQLAlchemy 真实发出的一条 SQL 查询（用事件监听器统计）。改变用户数量和加载策略试试。

Common Pitfalls (and Fixes) 🕳️आम ग़लतियाँ (और सुधार) 🕳️常见陷阱（及修正）🕳️

Real-World Example — A Full Task Tracker 📋असली उदाहरण — पूरा Task Tracker 📋实战示例 — 完整的任务跟踪器 📋

A complete mini-app: users have many tasks, tasks have a status, and we can query, update and report — all with the ORM and typed models.

एक पूरा mini-app: users के कई tasks हैं, tasks का status है, और हम query, update और report कर सकते हैं — सब ORM और typed models से।

一个完整的小应用：用户有多个任务，任务有状态，我们可以查询、更新和生成报告 — 全部使用 ORM 和带类型的模型。

from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import create_engine, String, ForeignKey, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str] = mapped_column(String(80))
    tasks: Mapped[list[Task]] = relationship(
        back_populates="owner", cascade="all, delete-orphan"
    )

class Task(Base):
    __tablename__ = "tasks"
    id:       Mapped[int] = mapped_column(primary_key=True)
    title:    Mapped[str] = mapped_column(String(200))
    status:   Mapped[str] = mapped_column(String(20), default="open")
    created:  Mapped[datetime] = mapped_column(default=lambda: datetime.now(timezone.utc))
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    owner:    Mapped[User] = relationship(back_populates="tasks")

engine = create_engine("sqlite:///tracker.db", echo=False)
Base.metadata.create_all(engine)

# -- Seed data -----------------------------------
with Session(engine) as s:
    if not s.scalar(select(User)):          # only on the first run
        s.add_all([
            User(name="Ada", tasks=[
                Task(title="Design schema",   status="done"),
                Task(title="Write ORM models", status="done"),
                Task(title="Add tests",       status="open"),
            ]),
            User(name="Grace", tasks=[
                Task(title="Draft docs",      status="open"),
                Task(title="Ship v1",         status="blocked"),
            ]),
        ])
        s.commit()

# -- Report: open tasks per user -----------------
with Session(engine) as s:
    stmt = (
        select(User.name, func.count(Task.id).label("n_open"))
        .join(Task, Task.owner_id == User.id)
        .where(Task.status == "open")
        .group_by(User.name)
    )
    print("Open tasks by user:")
    for name, n in s.execute(stmt):
        print(f"  {name:8s} {n} open")

    # Mark one task as done - the ORM notices the change
    t = s.scalar(select(Task).where(Task.title == "Add tests", Task.status == "open"))
    if t:
        t.status = "done"
        s.commit()
        print(f"\nmarked done: {t.title}")

Run it a second time…इसे दूसरी बार चलाइए…再运行一次…

from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import create_engine, String, ForeignKey, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str] = mapped_column(String(80))
    tasks: Mapped[list[Task]] = relationship(
        back_populates="owner", cascade="all, delete-orphan"
    )

class Task(Base):
    __tablename__ = "tasks"
    id:       Mapped[int] = mapped_column(primary_key=True)
    title:    Mapped[str] = mapped_column(String(200))
    status:   Mapped[str] = mapped_column(String(20), default="open")
    created:  Mapped[datetime] = mapped_column(default=lambda: datetime.now(timezone.utc))
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    owner:    Mapped[User] = relationship(back_populates="tasks")

engine = create_engine("sqlite:///tracker.db", echo=False)
Base.metadata.create_all(engine)

# -- Seed data -----------------------------------
with Session(engine) as s:
    if not s.scalar(select(User)):          # only on the first run
        s.add_all([
            User(name="Ada", tasks=[
                Task(title="Design schema",   status="done"),
                Task(title="Write ORM models", status="done"),
                Task(title="Add tests",       status="open"),
            ]),
            User(name="Grace", tasks=[
                Task(title="Draft docs",      status="open"),
                Task(title="Ship v1",         status="blocked"),
            ]),
        ])
        s.commit()

# -- Report: open tasks per user -----------------
with Session(engine) as s:
    stmt = (
        select(User.name, func.count(Task.id).label("n_open"))
        .join(Task, Task.owner_id == User.id)
        .where(Task.status == "open")
        .group_by(User.name)
    )
    print("Open tasks by user:")
    for name, n in s.execute(stmt):
        print(f"  {name:8s} {n} open")

    # Mark one task as done - the ORM notices the change
    t = s.scalar(select(Task).where(Task.title == "Add tests", Task.status == "open"))
    if t:
        t.status = "done"
        s.commit()
        print(f"\nmarked done: {t.title}")

Bug Doctor, Golden Rules and Quiz 🏆Bug Doctor, सुनहरे नियम और Quiz 🏆Bug 医生、黄金法则与小测验 🏆

🩺 Simulator: Bug Doctor🩺 Simulator: Bug Doctor🩺 模拟器：Bug 医生

Six real database mistakes. Buggy code on the left, fixed code on the right, changed lines marked. Every output is from a real run.

छह असली database ग़लतियाँ। बाईं ओर buggy code, दाईं ओर ठीक किया हुआ, बदली lines चिह्नित। हर output असली run से है।

六个真实的数据库错误。左边是有 bug 的代码，右边是修正后的代码，改动的行已标出。每个输出都来自真实运行。

How much faster is executemany()?executemany() कितना तेज़ है?executemany() 到底快多少？

import sqlite3, time, os

ROWS = [(f"Book {i}", "Author", 9.99, 5) for i in range(2000)]
def fresh():
    if os.path.exists("speed.db"): os.remove("speed.db")
    conn = sqlite3.connect("speed.db")
    conn.execute("CREATE TABLE books (title TEXT, author TEXT, price REAL, stock INT)")
    return conn
SQL = "INSERT INTO books VALUES (?, ?, ?, ?)"

conn = fresh(); t0 = time.perf_counter()
for r in ROWS:
    conn.execute(SQL, r); conn.commit()           # commit after EVERY row
a = time.perf_counter() - t0; conn.close()

conn = fresh(); t0 = time.perf_counter()
for r in ROWS:
    conn.execute(SQL, r)                          # loop, ONE commit at the end
conn.commit(); b = time.perf_counter() - t0; conn.close()

conn = fresh(); t0 = time.perf_counter()
conn.executemany(SQL, ROWS); conn.commit()        # executemany, one commit
c = time.perf_counter() - t0; conn.close()

print(f"loop + commit per row : {a * 1000:7.1f} ms")
print(f"loop + one commit     : {b * 1000:7.1f} ms")
print(f"executemany + commit  : {c * 1000:7.1f} ms")

The big win is not executemany() itself — it is committing once. Committing after every row forces SQLite to save to disk each time. Inside one transaction, a loop of execute() is already fast, and executemany() is a little faster and shorter to write.

बड़ा फ़ायदा ख़ुद executemany() नहीं — एक बार commit करना है। हर row के बाद commit करने से SQLite को हर बार disk पर save करना पड़ता है। एक transaction के अंदर execute() का loop पहले से तेज़ है, और executemany() थोड़ा और तेज़ और लिखने में छोटा है।

真正的提速不在 executemany() 本身 — 而在于 只提交一次。每插一行就提交一次，会迫使 SQLite 每次都写盘。在同一个事务中，循环 execute() 本来就很快，executemany() 只是更快一点、写起来更短。

Quick Quiz — 8 Questionsछोटा Quiz — 8 सवाल小测验 — 8 道题