Python 数据库实战：SQLite、PostgreSQL、SQLAlchemy 与 ORM 完全指南

import sqlite3
from pathlib import Path

DB_PATH = Path("app.db")

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                email TEXT UNIQUE,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

def add_user(name: str, email: str):
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("INSERT INTO users (name, email) VALUES (?, ?)", (name, email))
        conn.commit()

def get_users():
    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row   # 返回字典式结果
        return conn.execute("SELECT * FROM users").fetchall()

# 使用
init_db()
add_user("Alice", "alice@example.com")
for user in get_users():
    print(dict(user))

# 事务控制
with sqlite3.connect(DB_PATH) as conn:
    try:
        conn.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
        conn.execute("INSERT INTO invalid_table ...")   # 会失败
        conn.commit()
    except sqlite3.Error:
        conn.rollback()

# 返回字典结果
conn.row_factory = lambda c, r: {col[0]: r[idx] for idx, col in enumerate(c.description)}

# FTS5 全文搜索（需要编译时启用）
conn.execute("CREATE VIRTUAL TABLE docs USING fts5(title, content)")

# pip install psycopg2-binary
import psycopg2
from psycopg2.extras import RealDictCursor

conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="myuser",
    password="mypass",
)

with conn.cursor(cursor_factory=RealDictCursor) as cur:
    cur.execute("SELECT * FROM users WHERE age > %s", (18,))
    for row in cur.fetchall():
        print(dict(row))

conn.close()

# pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    
    author = relationship("User", back_populates="posts")

# 创建表
engine = create_engine("sqlite:///app.db", echo=True)
Base.metadata.create_all(engine)

# Session
Session = sessionmaker(bind=engine)

# 创建
with Session() as session:
    user = User(name="Alice", email="alice@example.com")
    session.add(user)
    session.commit()
    print(user.id)   # 自动获取自增 ID

# 查询
with Session() as session:
    # 主键查询
    user = session.get(User, 1)
    
    # 条件查询
    users = session.query(User).filter(User.name.like("A%")).all()
    
    # 分页
    page = session.query(User).offset(10).limit(10).all()

# 更新
with Session() as session:
    user = session.get(User, 1)
    user.name = "New Name"
    session.commit()

# 删除
with Session() as session:
    user = session.get(User, 1)
    session.delete(user)
    session.commit()

# 一对多：用户 - 文章
with Session() as session:
    user = session.get(User, 1)
    for post in user.posts:
        print(post.title)

# 预加载（避免 N+1 查询）
from sqlalchemy.orm import joinedload

users = session.query(User).options(joinedload(User.posts)).all()

# 聚合查询
from sqlalchemy import func

result = session.query(
    User.name,
    func.count(Post.id).label("post_count")
).join(Post).group_by(User.id).all()

# pip install asyncpg
import asyncpg
import asyncio

async def main():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    
    # 查询
    rows = await conn.fetch("SELECT * FROM users WHERE age > $1", 18)
    for row in rows:
        print(row["name"])
    
    # 事务
    async with conn.transaction():
        await conn.execute("INSERT INTO users (name) VALUES ($1)", "Bob")
    
    await conn.close()

asyncio.run(main())

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(async_engine, class_=AsyncSession)

async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User).where(User.age > 18))
        return result.scalars().all()

# 安装
pip install alembic

# 初始化
alembic init migrations

# 修改 alembic.ini → sqlalchemy.url = postgresql://user:pass@localhost/db
# 修改 migrations/env.py → 导入你的 Base

# 生成迁移脚本
alembic revision --autogenerate -m "create users table"

# 升级
alembic upgrade head

# 降级
alembic downgrade -1