PYTHON利用SQLAlchemy库高效操作关联的数据表

小咪咪 2024-02-20 09:08 51阅读 0赞

SQLAlchemy是一个Python库，用于处理关系型数据库的ORM（对象关系映射）和SQL表达式的生成。它提供了许多功能，包括：

1.  **ORM（对象关系映射）**：允许将数据库表映射到Python对象，使开发人员可以使用Python代码而不是SQL语句来操作数据库。
2.  **数据库连接管理**：SQLAlchemy管理数据库连接池，处理连接的创建、关闭和连接池的配置，从而提高了性能和资源利用率。
3.  **SQL表达式语言**：SQLAlchemy提供了强大的SQL表达式语言，可以用于构建SQL查询、插入、更新和删除等操作，同时支持数据库的不同类型和方言。
4.  **事务管理**：支持数据库事务，确保一系列数据库操作的原子性和数据一致性。
5.  **连接多种数据库**：SQLAlchemy支持多种关系型数据库，包括SQLite、MySQL、PostgreSQL、Oracle等，使得代码可以跨不同数据库进行移植。
6.  **数据模型定义**：使用SQLAlchemy，可以定义数据模型类，这些类可以映射到数据库表，简化了数据库结构的管理。
7.  **查询构建**：SQLAlchemy允许使用Python代码来构建复杂的查询，支持过滤、排序、连接、子查询等功能。
8.  **双向关系**：通过定义双向关系，可以轻松地在数据模型之间进行导航和访问相关的数据。
9.  **批量操作**：支持批量插入、批量更新和批量删除等操作，提高了数据库操作的效率。
10. **数据类型转换**：SQLAlchemy支持将Python数据类型与数据库数据类型进行映射，处理数据的序列化和反序列化。
11. **表间关联**：支持多表之间的关联，包括一对一、一对多和多对多等关系的建立和管理。

总之，SQLAlchemy是一个功能强大的工具，用于处理数据库交互和数据持久化，它提供了高度抽象的方式来操作数据库，同时支持灵活的SQL表达式，使开发人员更容易管理和操作数据库。

from sqlalchemy import create_engine, Column, Integer, String
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.orm import declarative_base
    from sqlalchemy import ForeignKey
    from sqlalchemy.orm import relationship, joinedload
    from sqlalchemy import UniqueConstraint
    from sqlalchemy import func
    from sqlalchemy import distinct
    
    # 创建引擎
    engine = create_engine('sqlite:///test456.db')
    
    # 创建会话
    Session = sessionmaker(bind=engine)
    session = Session()
    
    # 创建 SQLAlchemy 数据模型基类
    Base = declarative_base()
    
    # 定义数据模型类，继承自 Base
    class Address(Base):
        __tablename__ = 'addresses'
        id = Column(Integer, primary_key=True)
        user_id = Column(Integer, ForeignKey('users.id'))
        street = Column(String)
        user = relationship('User', back_populates='addresses')
    
        __table_args__ = (
            UniqueConstraint('user_id', 'street', name='unique_user_address'),
        )
    
    class User(Base):
        __tablename__ = 'users'
        id = Column(Integer, primary_key=True)
        name = Column(String)
        age = Column(Integer)
        addresses = relationship('Address', back_populates='user')
    
    # 创建数据库表
    Base.metadata.create_all(engine)
    
    # 插入用户数据
    user1 = User(name='Alice', age=30)
    user2 = User(name='Bob', age=25)
    
    # 插入地址数据
    address1 = Address(street='123 Main St', user=user1)
    address3 = Address(street='789 Oak St', user=user2)
    
    session.add(user1)
    session.add(user2)
    session.commit()
    
    # 查询所有用户以验证插入操作
    users = session.query(User).all()
    print('\nquery all users')
    for user in users:
        print(f"ID: {user.id:2} | Name: {user.name:7} | Age: {user.age:4}")
    
    # 执行关联查询并输出完整的用户信息
    users_with_addresses = session.query(User).options(joinedload(User.addresses)).all()
    print(' \n执行关联查询并输出完整的用户信息')
    for user in users_with_addresses:
        user_info = f"User ID: {user.id:3}, Name: {user.name:10}, Age: {user.age:4}"
        for address in user.addresses:
            address_info = f"，  Street: {address.street}"
            print(user_info.ljust(40) + address_info)
    
    # for user in users_with_addresses:
    #     print(f"User ID: {user.id}, Name: {user.name}, Age: {user.age}, Street: {', '.join([address.street for address in user.addresses])}")
    
    # for user in users_with_addresses:
    #     print(f"User ID: {user.id}, Name: {user.name}, Age: {user.age}")
    #     for address in user.addresses:
    #         print(f"  Address ID: {address.id}, Street: {address.street}")
    
    # 查询重复的用户记录
    duplicate_users = (
        session.query(User)
        .group_by(User.name, User.age)
        .having(func.count(User.id) > 1)
        .all()
    )
    
    # 删除重复记录，保留第一个记录
    for user in duplicate_users:
        first_user = (
            session.query(User)
            .filter_by(name=user.name, age=user.age)
            .order_by(User.id)
            .first()
        )
        session.query(User).filter_by(name=user.name, age=user.age).delete(synchronize_session=False)
        session.commit()
    
    # 查询所有用户以验证删除操作
    users = session.query(User).all()
    print(' \nquery all users')
    for user in users:
        print(f"ID: {user.id:3} | Name: {user.name:10} | Age: {user.age:4}")
    
    # 查询所有的地址
    all_addresses = session.query(Address).all()
    
    # 获取用户的 ID 列表
    user_ids = [user.id for user in session.query(User).all()]
    
    # 删除不对应的地址信息
    for address in all_addresses:
        if address.user_id not in user_ids:
            session.delete(address)
    session.commit()
    
    # 查询 Address 表以验证删除操作
    addresses = session.query(Address).all()
    print(' \nquery all addresses')
    for address in addresses:
        print(f"ID: {address.id:2} | Street: {address.street:20} | User ID: {address.user_id}")