Python中SQLAlchemy增刪改查與表關(guān)聯(lián)應(yīng)用詳解(最新推薦)

更新時(shí)間：2025年12月25日 16:47:28 作者：就這個(gè)丶調(diào)調(diào)

本文詳細(xì)介紹了如何使用Python的ORM工具SQLAlchemy進(jìn)行數(shù)據(jù)庫操作,包括增刪改查、表關(guān)聯(lián)應(yīng)用等,通過實(shí)例,展示了如何定義模型類、創(chuàng)建數(shù)據(jù)庫表、執(zhí)行CRUD操作以及處理復(fù)雜的表關(guān)聯(lián)關(guān)系,感興趣的朋友跟隨小編一起看看吧

Python中SQLAlchemy的全面學(xué)習(xí)：增刪改查與表關(guān)聯(lián)應(yīng)用詳解

引言

SQLAlchemy 是 Python 中最流行的 ORM（對(duì)象關(guān)系映射）工具之一，它使得開發(fā)者能夠以面向?qū)ο蟮姆绞讲僮鲾?shù)據(jù)庫，極大地提高了開發(fā)效率和代碼可讀性。本文將詳細(xì)介紹如何使用 SQLAlchemy 實(shí)現(xiàn)數(shù)據(jù)庫的增刪改查（CRUD）操作，以及如何處理表之間的關(guān)聯(lián)關(guān)系。

1. 環(huán)境準(zhǔn)備

安裝 SQLAlchemy

pip install sqlalchemy

導(dǎo)入必要的模塊

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DateTime, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import datetime

2. 創(chuàng)建數(shù)據(jù)庫引擎和會(huì)話工廠

# 創(chuàng)建數(shù)據(jù)庫引擎，這里以 SQLite 為例（也可以使用 MySQL、PostgreSQL 等）
engine = create_engine('sqlite:///example.db', echo=True)  # echo=True 用于顯示 SQL 語句
# 創(chuàng)建基類（所有模型類的基類）
Base = declarative_base()
# 創(chuàng)建會(huì)話工廠（用于創(chuàng)建會(huì)話對(duì)象）
SessionLocal = sessionmaker(bind=engine)

3. 定義數(shù)據(jù)模型（表結(jié)構(gòu)）

3.1 定義用戶表（User）

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False, unique=True)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    # 定義與文章表的關(guān)聯(lián)關(guān)系（反向關(guān)系）
    posts = relationship("Post", back_populates="author")
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"

3.2 定義文章表（Post）

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    content = Column(String(1000))
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    user_id = Column(Integer, ForeignKey('users.id'))  # 外鍵關(guān)聯(lián)到 User 表
    # 定義與用戶表的關(guān)聯(lián)關(guān)系（正向關(guān)系）
    author = relationship("User", back_populates="posts")
    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title}', user_id={self.user_id})>"

4. 創(chuàng)建數(shù)據(jù)庫表

# 通過 Base.metadata.create_all() 創(chuàng)建所有定義的表
Base.metadata.create_all(engine)

5. 執(zhí)行 CRUD 操作（增刪改查）

5.1 增加（Create）

# 創(chuàng)建會(huì)話實(shí)例
session = SessionLocal()
try:
    # 創(chuàng)建新用戶
    new_user = User(username='alice', email='alice@example.com')
    session.add(new_user)
    session.commit()  # 提交事務(wù)，保存到數(shù)據(jù)庫
    print(f"用戶 {new_user.username} 添加成功！")
    # 創(chuàng)建新文章，關(guān)聯(lián)到用戶
    new_post = Post(title='我的第一篇博客', content='這是我的第一篇文章內(nèi)容。', user_id=new_user.id)
    session.add(new_post)
    session.commit()
    print(f"文章 {new_post.title} 添加成功！")
except Exception as e:
    session.rollback()  # 出錯(cuò)時(shí)回滾事務(wù)
    print(f"添加失?。簕e}")
finally:
    session.close()

5.2 查詢（Read）

5.2.1 按主鍵查詢單個(gè)記錄

session = SessionLocal()
try:
    user = session.query(User).filter(User.id == 1).first()
    if user:
        print(user)
    else:
        print("未找到該用戶")
finally:
    session.close()

5.2.2 條件查詢多個(gè)記錄

session = SessionLocal()
try:
    users = session.query(User).filter(User.username.like('%a%')).all()
    for user in users:
        print(user)
finally:
    session.close()

5.2.3 使用 join 進(jìn)行多表聯(lián)合查詢（關(guān)聯(lián)查詢）

session = SessionLocal()
try:
    # 查詢所有用戶的姓名及其發(fā)表的文章標(biāo)題
    results = session.query(User.username, Post.title).join(Post).all()
    for username, title in results:
        print(f"{username} - {title}")
finally:
    session.close()

5.2.4 使用with_parent查詢關(guān)聯(lián)數(shù)據(jù)（更簡(jiǎn)潔的方式）

session = SessionLocal()
try:
    # 獲取用戶 alice 及其所有文章
    alice = session.query(User).filter(User.username == 'alice').first()
    if alice:
        for post in alice.posts:
            print(post.title)
finally:
    session.close()

5.3 修改（Update）

session = SessionLocal()
try:
    # 更新用戶郵箱
    user = session.query(User).filter(User.username == 'alice').first()
    if user:
        user.email = 'alice_new@example.com'
        session.commit()
        print(f"用戶 {user.username} 的郵箱已更新為 {user.email}")
    else:
        print("用戶不存在")
except Exception as e:
    session.rollback()
    print(f"更新失敗：{e}")
finally:
    session.close()

5.4 刪除（Delete）

session = SessionLocal()
try:
    # 刪除文章，先刪除文章再刪除用戶（注意外鍵約束）
    post = session.query(Post).filter(Post.title == '我的第一篇博客').first()
    if post:
        session.delete(post)
        session.commit()
        print(f"文章 {post.title} 已刪除")
    else:
        print("文章不存在")
except Exception as e:
    session.rollback()
    print(f"刪除失?。簕e}")
finally:
    session.close()

6. 表之間關(guān)聯(lián)的應(yīng)用詳解

6.1 一對(duì)一（One-to-One）

# 例如：用戶有一個(gè)唯一的個(gè)人資料卡片（Profile）
class Profile(Base):
    __tablename__ = 'profiles'
    id = Column(Integer, primary_key=True)
    bio = Column(String(500))
    user_id = Column(Integer, ForeignKey('users.id'), unique=True)  # 一對(duì)一，外鍵唯一
    user = relationship("User", backref="profile")

6.2 一對(duì)多（One-to-Many）

# 已在 User 和 Post 之間實(shí)現(xiàn)：一個(gè)用戶可以有多個(gè)文章（一對(duì)多）
# 通過 `relationship` 和 `back_populates` 實(shí)現(xiàn)雙向關(guān)系。

6.3 多對(duì)多（Many-to-Many）

# 例如：用戶和標(biāo)簽之間是多對(duì)多關(guān)系（一個(gè)用戶可以有多個(gè)標(biāo)簽，一個(gè)標(biāo)簽可以被多個(gè)用戶使用）
# 創(chuàng)建中間表（關(guān)聯(lián)表）
user_tag = Table('user_tag', Base.metadata,
                Column('user_id', Integer, ForeignKey('users.id')),
                Column('tag_id', Integer, ForeignKey('tags.id'))
               )
class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    # 與用戶的關(guān)系（多對(duì)多）
    users = relationship("User", secondary=user_tag, back_populates="tags")
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), nullable=False)
    # 與標(biāo)簽的關(guān)系（多對(duì)多）
    tags = relationship("Tag", secondary=user_tag, back_populates="users")

6.3.1 使用多對(duì)多關(guān)系示例：添加標(biāo)簽

session = SessionLocal()
try:
    # 創(chuàng)建用戶和標(biāo)簽
    user = User(username='bob')
    tag1 = Tag(name='Python')
    tag2 = Tag(name='Web')
    # 將標(biāo)簽添加到用戶（自動(dòng)創(chuàng)建中間表記錄）
    user.tags.append(tag1)
    user.tags.append(tag2)
    session.add(user)
    session.commit()
    print(f"用戶 {user.username} 已添加標(biāo)簽: {[t.name for t in user.tags]}")
except Exception as e:
    session.rollback()
    print(f"添加標(biāo)簽失?。簕e}")
finally:
    session.close()

7. 高級(jí)技巧與最佳實(shí)踐

7.1 事務(wù)管理（Transaction Management）

session = SessionLocal()
try:
    # 多個(gè)操作在一個(gè)事務(wù)中執(zhí)行，保證一致性
    user = User(username='charlie', email='charlie@example.com')
    post = Post(title='測(cè)試文章', content='測(cè)試內(nèi)容', user_id=user.id)
    session.add(user)
    session.add(post)
    session.commit()  # 一次性提交，如果出錯(cuò)則回滾
except Exception as e:
    session.rollback()
    print(f"事務(wù)失?。簕e}")
finally:
    session.close()

7.2 優(yōu)雅地關(guān)閉會(huì)話（使用上下文管理器）

from contextlib import contextmanager
@contextmanager
def get_db_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()
# 使用示例：
with get_db_session() as session:
    user = User(username='david', email='david@example.com')
    session.add(user)
    # 無需手動(dòng)調(diào)用 commit()，上下文管理器會(huì)自動(dòng)處理

7.3 使用query的鏈?zhǔn)秸{(diào)用進(jìn)行復(fù)雜查詢

# 查詢用戶名包含 'a' 且文章數(shù)量大于 1 的用戶
results = (session.query(User)
           .join(Post)
           .group_by(User.id)
           .having(text("count(Post.id) > 1"))
           .filter(User.username.like('%a%'))
           .all())
for user in results:
    print(f"{user.username} 有 {len(user.posts)} 篇文章")

總結(jié)

SQLAlchemy 提供了強(qiáng)大而靈活的 ORM 功能，使我們能夠以簡(jiǎn)潔、直觀的方式進(jìn)行數(shù)據(jù)庫操作。本文系統(tǒng)地介紹了：

如何定義模型類并創(chuàng)建數(shù)據(jù)庫表；
如何執(zhí)行基本的 CRUD 操作；
如何處理一對(duì)一、一對(duì)多、多對(duì)多等復(fù)雜的表關(guān)聯(lián)關(guān)系；
一些高級(jí)技巧，如事務(wù)管理、上下文管理器、復(fù)雜查詢等。

掌握這些知識(shí)后，你將能夠在實(shí)際項(xiàng)目中高效地使用 SQLAlchemy 操控?cái)?shù)據(jù)庫，提升開發(fā)效率和代碼質(zhì)量。

到此這篇關(guān)于Python中SQLAlchemy增刪改查與表關(guān)聯(lián)應(yīng)用詳解(最新推薦)的文章就介紹到這了,更多相關(guān)Python SQLAlchemy增刪改查內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: