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