如何使用Python批量處理Excel和CSV文件

更新時間：2026年03月30日 09:08:59 作者：qq_14811537

本文介紹了SQLAlchemyORM的基本用法,包括安裝、核心概念、連接數據庫、定義數據模型、基本CRUD操作、查詢數據、事務管理和最佳實踐等,通過本文,讀者可以掌握使用SQLAlchemy進行高效數據庫操作的方法,感興趣的朋友跟隨小編一起看看吧

安裝

pip install sqlalchemy

如果需要連接特定數據庫，還需安裝相應的驅動程序：

# PostgreSQL
pip install psycopg2-binary
# MySQL
pip install mysql-connector-python
# SQLite (Python標準庫已包含，無需額外安裝)

核心概念

Engine：數據庫連接的引擎，負責與數據庫通信
Session：數據庫會話，管理所有持久化操作
Model：數據模型類，對應數據庫中的表
Query：查詢對象，用于構建和執(zhí)行數據庫查詢

連接數據庫

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 創(chuàng)建數據庫連接引擎
# SQLite示例
engine = create_engine('sqlite:///example.db', echo=True)
# PostgreSQL示例
# engine = create_engine('postgresql://username:password@localhost:5432/mydatabase')
# MySQL示例
# engine = create_engine('mysql+mysqlconnector://username:password@localhost:3306/mydatabase')
# 創(chuàng)建會話工廠
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 創(chuàng)建會話實例
session = SessionLocal()

定義數據模型

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
# 創(chuàng)建基類
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    # 定義一對多關系
    posts = relationship("Post", back_populates="author")
class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    content = Column(String(500))
    author_id = Column(Integer, ForeignKey('users.id'))
    # 定義多對一關系
    author = relationship("User", back_populates="posts")
    # 定義多對多關系（通過關聯(lián)表）
    tags = relationship("Tag", secondary="post_tags", back_populates="posts")
class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(30), unique=True, nullable=False)
    posts = relationship("Post", secondary="post_tags", back_populates="tags")
# 關聯(lián)表（用于多對多關系）
class PostTag(Base):
    __tablename__ = 'post_tags'
    post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True)
    tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

創(chuàng)建數據庫表

# 創(chuàng)建所有表
Base.metadata.create_all(bind=engine)
# 刪除所有表
# Base.metadata.drop_all(bind=engine)

基本CRUD操作

創(chuàng)建數據

# 創(chuàng)建新用戶
new_user = User(name="張三", email="zhangsan@example.com")
session.add(new_user)
session.commit()
# 批量創(chuàng)建
session.add_all([
    User(name="李四", email="lisi@example.com"),
    User(name="王五", email="wangwu@example.com")
])
session.commit()

讀取數據

# 獲取所有用戶
users = session.query(User).all()
# 獲取第一個用戶
first_user = session.query(User).first()
# 根據ID獲取用戶
user = session.query(User).get(1)

更新數據

# 查詢并更新
user = session.query(User).get(1)
user.name = "張三四"
session.commit()
# 批量更新
session.query(User).filter(User.name.like("張%")).update({"name": "張氏"}, synchronize_session=False)
session.commit()

刪除數據

# 查詢并刪除
user = session.query(User).get(1)
session.delete(user)
session.commit()
# 批量刪除
session.query(User).filter(User.name == "李四").delete(synchronize_session=False)
session.commit()

查詢數據

基本查詢

# 獲取所有記錄
users = session.query(User).all()
# 獲取特定字段
names = session.query(User.name).all()
# 排序
users = session.query(User).order_by(User.name.desc()).all()
# 限制結果數量
users = session.query(User).limit(10).all()
# 偏移量
users = session.query(User).offset(5).limit(10).all()

過濾查詢

from sqlalchemy import or_
# 等值過濾
user = session.query(User).filter(User.name == "張三").first()
# 模糊查詢
users = session.query(User).filter(User.name.like("張%")).all()
# IN查詢
users = session.query(User).filter(User.name.in_(["張三", "李四"])).all()
# 多條件查詢
users = session.query(User).filter(
    User.name == "張三", 
    User.email.like("%@example.com")
).all()
# 或條件
users = session.query(User).filter(
    or_(User.name == "張三", User.name == "李四")
).all()
# 不等于
users = session.query(User).filter(User.name != "張三").all()

聚合查詢

from sqlalchemy import func
# 計數
count = session.query(User).count()
# 分組計數
user_post_count = session.query(
    User.name, 
    func.count(Post.id)
).join(Post).group_by(User.name).all()
# 求和、平均值等
avg_id = session.query(func.avg(User.id)).scalar()

連接查詢

# 內連接
results = session.query(User, Post).join(Post).filter(Post.title.like("%Python%")).all()
# 左外連接
results = session.query(User, Post).outerjoin(Post).all()
# 指定連接條件
results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

關系操作

# 創(chuàng)建帶關系的對象
user = User(name="趙六", email="zhaoliu@example.com")
post = Post(title="我的第一篇博客", content="Hello World!", author=user)
session.add(post)
session.commit()
# 通過關系訪問
print(f"文章 '{post.title}' 的作者是 {post.author.name}")
print(f"用戶 {user.name} 的所有文章:")
for p in user.posts:
    print(f"  - {p.title}")
# 多對多關系操作
python_tag = Tag(name="Python")
sqlalchemy_tag = Tag(name="SQLAlchemy")
post.tags.append(python_tag)
post.tags.append(sqlalchemy_tag)
session.commit()
print(f"文章 '{post.title}' 的標簽:")
for tag in post.tags:
    print(f"  - {tag.name}")

事務管理

# 自動提交事務
try:
    user = User(name="測試用戶", email="test@example.com")
    session.add(user)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"發(fā)生錯誤: {e}")
# 使用事務上下文管理器
from sqlalchemy.orm import Session
def create_user(session: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        session.add(user)
        session.commit()
        return user
    except:
        session.rollback()
        raise
# 嵌套事務
with session.begin_nested():
    user = User(name="事務用戶", email="transaction@example.com")
    session.add(user)
# 保存點
savepoint = session.begin_nested()
try:
    user = User(name="保存點用戶", email="savepoint@example.com")
    session.add(user)
    savepoint.commit()
except:
    savepoint.rollback()

最佳實踐

會話管理：為每個請求創(chuàng)建新會話，請求結束后關閉
異常處理：始終處理異常并適當回滾事務
延遲加載：注意N+1查詢問題，使用 eager loading 優(yōu)化
連接池：合理配置連接池大小和超時設置
數據驗證：在模型層或應用層驗證數據完整性

# 使用上下文管理器管理會話
from contextlib import contextmanager
@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()
# 使用示例
with get_db() as db:
    user = User(name="上下文用戶", email="context@example.com")
    db.add(user)

總結

SQLAlchemy ORM提供了強大而靈活的數據庫操作方式，通過本文的介紹，您應該能夠：

安裝和配置SQLAlchemy
定義數據模型和關系
執(zhí)行基本的CRUD操作
構建復雜查詢
管理數據庫事務
遵循最佳實踐

SQLAlchemy還有更多高級特性，如混合屬性、事件監(jiān)聽、自定義查詢等，值得進一步探索學習。

到此這篇關于如何使用Python批量處理Excel和CSV文件的文章就介紹到這了,更多相關Python批量處理Excel和CSV文件內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

完美解決TensorFlow和Keras大數據量內存溢出的問題
這篇文章主要介紹了完美解決TensorFlow和Keras大數據量內存溢出的問題，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2020-07-07
Python pandas 的索引方式 data.loc[],data[][]示例詳解
這篇文章主要介紹了Python pandas 的索引方式 data.loc[], data[][]的相關資料,其中data.loc[index,column]使用.loc[ ]第一個參數是行索引，第二個參數是列索引，本文結合實例代碼講解的非常詳細，需要的朋友可以參考下
2023-02-02
Numpy中對向量、矩陣的使用詳解
這篇文章主要介紹了Numpy中對向量、矩陣的使用詳解，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2019-10-10
Python實現(xiàn)的從右到左字符串替換方法示例
這篇文章主要介紹了Python實現(xiàn)的從右到左字符串替換方法,涉及Python字符串遍歷、運算、判斷、替換等相關操作技巧,需要的朋友可以參考下
2018-07-07
使用Python實現(xiàn)圖像LBP特征提取的操作方法
LBP特征叫做局部二值模式,常用于紋理特征提取,并在紋理分類中具有較強的區(qū)分能力,本文給大家介紹了如何使用Python實現(xiàn)圖像LBP特征提取的操作方法,需要的朋友可以參考下
2025-04-04
純Python開發(fā)的nosql數據庫CodernityDB介紹和使用實例
這篇文章主要介紹了純Python開發(fā)的nosql數據庫CodernityDB介紹和使用實例,本文實例包含數據插入、數據更新、數據刪除、數據查詢等,需要的朋友可以參考下
2014-10-10
Python從list類型、range()序列簡單認識類(class)【可迭代】
這篇文章主要介紹了Python從list類型、range()序列簡單認識類(class),結合實例形式分析了list、range及自定義類等可迭代數據類型相關使用技巧,需要的朋友可以參考下
2019-05-05
Python自動化測試框架之unittest使用詳解
unittest是Python自動化測試框架之一，提供了一系列測試工具和接口，支持單元測試、功能測試、集成測試等多種測試類型。unittest使用面向對象的思想實現(xiàn)測試用例的編寫和管理，可以方便地擴展和定制測試框架，支持多種測試結果輸出格式
2023-04-04
Python中的Pandas?時間函數?time?、datetime?模塊和時間處理基礎講解
Python?中提供了對時間日期的多種多樣的處理方式，主要是在?time?和?datetime?這兩個模塊里，這篇文章主要介紹了Python中的Pandas?時間函數?time?、datetime?模塊和時間處理基礎,需要的朋友可以參考下
2023-03-03
python實現(xiàn)selenium截圖的兩種方法
本文主要介紹了python實現(xiàn)selenium截圖的兩種方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2023-04-04