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

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}")