SQLAlchemy Hello World：声明式 ORM 入门

目标

使用 SQLAlchemy 2.0 声明式模型创建表、插入数据、查询数据，展示 Core 和 ORM 两种模式。

完整代码

from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import Session, DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, ForeignKey, DateTime
from datetime import datetime


# === 1. 连接数据库 ===
engine = create_engine("sqlite:///demo.db", echo=True)


# === 2. 声明式模型 ===
class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)

    posts: Mapped[list["Post"]] = relationship(back_populates="author")

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    content: Mapped[str] = mapped_column(String(500))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped["User"] = relationship(back_populates="posts")


# === 3. 建表 ===
Base.metadata.create_all(engine)


# === 4. CRUD 操作 ===
with Session(engine) as session:
    # CREATE
    alice = User(name="Alice", email="alice@example.com")
    bob = User(name="Bob", email="bob@example.com")
    session.add_all([alice, bob])
    session.flush()  # 获取 id

    session.add_all([
        Post(title="Python Tips", content="Use list comprehensions!", user_id=alice.id),
        Post(title="SQL Guide", content="Always use indexes.", user_id=alice.id),
        Post(title="Flask vs FastAPI", content="Depends...", user_id=bob.id),
    ])
    session.commit()

    # READ — 查询所有用户
    users = session.execute(select(User)).scalars().all()
    print("\n所有用户:", users)

    # READ — 带条件的查询
    alice = session.execute(
        select(User).where(User.name == "Alice")
    ).scalar_one()
    print(f"\n{alice.name} 的文章:")
    for post in alice.posts:
        print(f"  - {post.title}")

    # AGGREGATE
    from sqlalchemy import func
    count_stmt = select(User.name, func.count(Post.id)).join(Post).group_by(User.name)
    for name, cnt in session.execute(count_stmt):
        print(f"  {name}: {cnt} 篇文章")

    # UPDATE
    session.execute(select(Post)).scalars().first()
    bob_post = session.execute(
        select(Post).where(Post.author.has(name="Bob"))
    ).scalar_one()
    bob_post.title = "Flask vs FastAPI (Updated)"
    session.commit()

    # DELETE
    session.delete(bob)
    session.commit()
    remaining = session.execute(select(func.count(User.id))).scalar()
    print(f"\n剩余用户数: {remaining}")

运行步骤

pip install sqlalchemy
python hello_sqlalchemy.py

预期输出

所有用户: [<User(id=1, name='Alice')>, <User(id=2, name='Bob')>]

Alice 的文章:
  - Python Tips
  - SQL Guide
  Alice: 2 篇文章
  Bob: 1 篇文章

剩余用户数: 1