当前位置： 首页 > news >正文

SQLAlchemy 介绍与实践

news 2024/11/6 14:51:09

postgresql 实践
pydantic 实践

1. SQLAlchemy 介绍

SQLAlchemy 是一个 ORM 框架。SQLAlchemy 是一个用于 Python 的 SQL 工具和对象关系映射（ORM）库。它允许你通过 Python 代码来与关系型数据库交互，而不必直接编写SQL语句。
简单介绍一下对象关系映射吧，对象关系映射（英语：Object Relational Mapping，简称 ORM，或O/RM，或O/R mapping），是一种程序设计技术， 用于实现面向对象编程语言里不同类型系统的数据之间的转换。 从效果上说，它其实是创建了一个可在编程语言里使用的“虚拟对象数据库”。大白话：对象模型与数据库表的映射。

1.1. SQLAlchemy 使用场景

SQLAlchemy 是一个强大的 Python ORM 框架，主要应用于以下场景：
数据库访问和操作： SQLAlchemy 提供了高层抽象来操作数据库，可以避免写原生 SQL 语句。支持多种数据库后端（MySQL、MongoDB、SQLite、PostgreSQL）。
ORM映射： 建立 Python 类与数据库表的映射关系，简化数据模型的操作，支持声明式操作。
复杂查询： SQLAlchemy 提供丰富的查询方式，如过滤、分组、联结等，可以构建复杂查询。
异步查询： 基于Greenlet 等实现异步查询，提高查询效率。
事务控制： 通过 Session 管理数据库会话和事务。
工具集成： 如数据迁移工具 Alembic，可以实现 Schema 版本控制和迁移。
大数据集查询： 基于 Pagination 实现数据分页，避免大量数据查询内存溢出。
多数据库支持： 支持 Postgres、MySQL、Oracle 等主流数据库。
Web框架集成： 框架如 Flask 可以集成 SQLAlchemy，便于 Web 应用开发。

2. SQLAlchemy 基本用法

参考：
https://www.jb51.net/python/325524ud6.htm
https://blog.51cto.com/u_13019/12307379

2.1. 安装 SQLAlchemy

在使用 SQLAlchemy 之前，首先需要安装它。可以使用以下命令使用 pip 安装：

pip install sqlalchemy 
pip install pymysql		# 安装 MySQL

2.2. 连接数据库

使用 SQLAlchemy 连接到数据库，需要提供数据库的连接字符串，其中包含有关数据库类型、用户名、密码、主机和数据库名称的信息。

from sqlalchemy import create_engine# 例如，连接到 SQLite 数据库
engine = create_engine('sqlite:///example.db')# 例如，连接到 MySQL 数据库
username = 'your_mysql_username'
password = 'your_mysql_password'
host = 'your_mysql_host'  # 例如：'localhost' 或 '127.0.0.1'
port = 'your_mysql_port'  # 通常是 3306
database = 'your_database_name'
# 创建连接引擎
engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}')

2.3. 定义数据模型

使用 SQLAlchemy 的 ORM 功能，可以定义 Python 类来映射数据库中的表。每个类对应数据库中的一张表，类的属性对应表中的列。

# 导入必要的模块
from sqlalchemy import Column, Integer, String, Sequence
from sqlalchemy.ext.declarative import declarative_base# 创建一个基类，用于定义数据模型的基本结构
Base = declarative_base()# 定义一个数据模型类，对应数据库中的 'users' 表
class User(Base):# 定义表名__tablename__ = 'users'# 定义列：id，是整数类型，主键（primary_key=True），并使用 Sequence 生成唯一标识id = Column(Integer, Sequence('user_id_seq'), primary_key=True)# 定义列：name，是字符串类型，最大长度为50name = Column(String(50))# 定义列：age，是整数类型age = Column(Integer)

2.4. 创建表

通过在代码中调用 create_all 方法，可以根据定义的模型创建数据库表。

Base.metadata.create_all(engine)

2.5. 插入数据

使用 SQLAlchemy 进行插入数据的操作，首先需要创建一个会话（Session）对象，然后使用该对象添加数据并提交。

# 导入创建会话的模块
from sqlalchemy.orm import sessionmaker# 使用 sessionmaker 创建一个会话类 Session，并绑定到数据库引擎（bind=engine）
Session = sessionmaker(bind=engine)# 创建一个实例化的会话对象 session
session = Session()# 创建一个新的 User 实例，即要插入到数据库中的新用户
new_user = User(name='John Doe', age=30)# 将新用户添加到会话中，即将其添加到数据库操作队列中
session.add(new_user)# 提交会话，将所有在此会话中的数据库操作提交到数据库
session.commit()

2.6. 使用事务添加数据

如果添加过程中发生任何错误，我们将回滚事务，确保数据库的一致性。

try:# 开始一个新的事务session.begin()# 创建新用户对象user1 = User(name='Alice', email='alice@example.com')user2 = User(name='Bob', email='bob@example.com')# 添加到会话中session.add(user1)session.add(user2)# 提交事务，将所有更改保存到数据库session.commit()print("Users added successfully.")
except Exception as e:# 如果在添加用户过程中发生错误，则回滚事务session.rollback()print(f"An error occurred: {e}")
finally:# 关闭会话session.close()

在这个示例中，我们使用 session.begin() 显式地开始了一个新的事务。然后，我们尝试添加两个新用户到会话中。如果在这个过程中没有发生任何错误，我们使用 session.commit() 提交事务，将所有更改保存到数据库中。但是，如果在添加用户的过程中发生了任何异常（例如，由于重复的电子邮件地址或数据库连接问题），我们将使用 session.rollback() 回滚事务，确保数据库的一致性。

2.7. 查询数据

使用 SQLAlchemy 进行查询数据的操作，可以通过查询语句或使用 ORM 查询接口。

# 使用查询语句
result = engine.execute('SELECT * FROM users')# 使用 ORM 查询接口
users = session.query(User).all()

3. 复杂查询条件

3.1. 连接查询（Join）

假设我们有两个模型， User 和 Order ，并且一个用户可以有多个订单。

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String)orders = relationship("Order", back_populates="user")class Order(Base):__tablename__ = 'orders'id = Column(Integer, primary_key=True)user_id = Column(Integer, ForeignKey('users.id'))product = Column(String)quantity = Column(Integer)user = relationship("User", back_populates="orders")

现在，如果我们想要查询所有下过订单的用户及其订单信息，我们可以进行连接查询：

from sqlalchemy.orm import joinedload
# 加载所有用户的订单信息
users_with_orders = session.query(User).options(joinedload(User.orders)).all()
for user in users_with_orders:print(f"User: {user.name}")for order in user.orders:print(f"  Order: {order.product}, Quantity: {order.quantity}")

3.2. 分组和聚合（Grouping and Aggregation）

假设我们想要统计每个用户下的订单总数。

from sqlalchemy import func
# 按用户分组，并计算每个用户的订单数量
order_count_by_user = session.query(User.id, User.name, func.count(Order.id).label('order_count')).\join(Order).group_by(User.id, User.name).all()
for user_id, user_name, order_count in order_count_by_user:print(f"User ID: {user_id}, Name: {user_name}, Order Count: {order_count}")

3.3. 子查询（Subquery）

如果我们想要找出订单数量超过平均订单数量的用户，我们可以使用子查询。

from sqlalchemy import func, select
# 计算平均订单数量作为子查询
avg_order_quantity = select([func.avg(Order.quantity).label('avg_quantity')]).select_from(Order).alias()
# 查询订单数量超过平均值的用户及其订单信息
users_above_avg = session.query(User, Order.product, Order.quantity).\join(Order).filter(Order.quantity > avg_order_quantity.c.avg_quantity).all()
for user, product, quantity in users_above_avg:print(f"User: {user.name}, Product: {product}, Quantity: {quantity}")

3.4. 复杂筛选条件（Complex Filtering）

假设我们想要找到名字以 “A” 开头的用户，并且他们的订单中包含 “apple” 这个产品。

# 查询名字以“A”开头的用户，且订单中包含“apple”产品的用户信息
users_with_apple = session.query(User).join(Order).\filter(User.name.startswith('A')).\filter(Order.product.contains('apple')).\distinct().all()  # 使用distinct() 确保结果中的用户不重复
for user in users_with_apple:print(f"User: {user.name}")

3.5. 分页查询

3.5.1. limit

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine# 假设已经有了一个定义好的模型和数据库引擎
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()# 分页查询函数
def paginate_query(page, page_size):# 计算跳过的记录数offset = (page - 1) * page_size# 执行分页查询results = session.query(YourModel).order_by(YourModel.id).offset(offset).limit(page_size).all()return results# 使用分页查询
page = 1
page_size = 10
records = paginate_query(page, page_size)

3.5.2. slice

total = session.query(YourModel).with_entities(func.count(YourModel.id)).scalar()
db_query = session.query(YourModel).slice(page_start, page_end)

4. SQLAlchemy 实践1

4.1. 定义数据模型类

我们定义三个数据模型类：User（用户）、Post（文章）和Comment（评论）。这些类之间通过外键和关系进行关联。

# 导入 SQLAlchemy 中所需的模块
from sqlalchemy import create_engine, Column, Integer, String, Text, ForeignKey
from sqlalchemy.orm import declarative_base, relationship# 创建一个基类，用于定义数据模型的基本结构
Base = declarative_base()# 定义数据模型类 User，对应数据库中的 'users' 表
class User(Base):__tablename__ = 'users'# 定义列：id，是整数类型，作为主键id = Column(Integer, primary_key=True)# 定义列：username，是字符串类型，最大长度为50，唯一且不可为空username = Column(String(50), unique=True, nullable=False)# 定义列：email，是字符串类型，最大长度为100，唯一且不可为空email = Column(String(100), unique=True, nullable=False)# 定义关系，与 Post 类的关系为一对多关系，通过 back_populates 指定反向关系属性名posts = relationship('Post', back_populates='author')# 定义数据模型类 Post，对应数据库中的 'posts' 表
class Post(Base):__tablename__ = 'posts'# 定义列：id，是整数类型，作为主键id = Column(Integer, primary_key=True)# 定义列：title，是字符串类型，最大长度为100，不可为空title = Column(String(100), nullable=False)# 定义列：content，是文本类型，不可为空content = Column(Text, nullable=False)# 定义列：user_id，是整数类型，外键关联到 'users' 表的 id 列user_id = Column(Integer, ForeignKey('users.id'))# 定义关系，与 User 类的关系为多对一关系，通过 back_populates 指定反向关系属性名author = relationship('User', back_populates='posts')# 定义关系，与 Comment 类的关系为一对多关系，通过 back_populates 指定反向关系属性名comments = relationship('Comment', back_populates='post')# 定义数据模型类 Comment，对应数据库中的 'comments' 表
class Comment(Base):__tablename__ = 'comments'# 定义列：id，是整数类型，作为主键id = Column(Integer, primary_key=True)# 定义列：text，是文本类型，不可为空text = Column(Text, nullable=False)# 定义列：user_id，是整数类型，外键关联到 'users' 表的 id 列user_id = Column(Integer, ForeignKey('users.id'))# 定义列：post_id，是整数类型，外键关联到 'posts' 表的 id 列post_id = Column(Integer, ForeignKey('posts.id'))# 定义关系，与 User 类的关系为多对一关系author = relationship('User')# 定义关系，与 Post 类的关系为多对一关系，通过 back_populates 指定反向关系属性名post = relationship('Post', back_populates='comments')

4.2. 创建数据库引擎和会话

这里我们选择了 SQLite 数据库，并使用 create_all 创建相应的表。

# 导入 SQLAlchemy 中所需的模块
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker# 创建一个 SQLite 数据库引擎，连接到名为 'blog.db' 的数据库文件
engine = create_engine('sqlite:///blog.db')# 使用 Base 对象的 metadata 属性，创建数据库中定义的所有表
Base.metadata.create_all(engine)# 使用 sessionmaker 创建一个会话类 Session，并将其绑定到上面创建的数据库引擎
Session = sessionmaker(bind=engine)# 创建一个实例化的会话对象 session，用于进行数据库操作
session = Session()

4.3. 进行数据库操作

这段代码演示了如何使用 SQLAlchemy 对数据库进行插入和查询操作。首先，创建了一个用户、一篇文章和一条评论，然后通过查询用户的方式，打印出该用户的所有文章及评论。

# 创建一个新用户对象并设置其属性
user1 = User(username='john_doe', email='john@example.com')# 将新用户对象添加到会话，表示要进行数据库插入操作
session.add(user1)# 提交会话，将所有在此会话中的数据库操作提交到数据库
session.commit()# 创建一篇新文章对象并设置其属性
post1 = Post(title='Introduction to SQLAlchemy', content='SQLAlchemy is a powerful ORM for Python.')# 将文章的作者关联到之前创建的用户
post1.author = user1# 将新文章对象添加到会话，表示要进行数据库插入操作
session.add(post1)# 提交会话，将所有在此会话中的数据库操作提交到数据库
session.commit()# 创建一条新评论对象并设置其属性
comment1 = Comment(text='Great article!', author=user1, post=post1)# 将评论对象添加到会话，表示要进行数据库插入操作
session.add(comment1)# 提交会话，将所有在此会话中的数据库操作提交到数据库
session.commit()# 查询用户名为 'john_doe' 的用户，并打印其所有文章及评论
user = session.query(User).filter_by(username='john_doe').first()
print(f"User: {user.username}")# 遍历用户的所有文章
for post in user.posts:print(f"Post: {post.title}")# 遍历文章的所有评论for comment in post.comments:print(f"Comment: {comment.text}")

5. SQLAlchemy 实践2 数据库初始化操作

main.py

from sqlalchemy import create_engine
from sqlalchemy_utils import create_database, database_existsfrom sqlalchemy import and_, or_, func
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import SingletonThreadPoolfrom database import Base
from database import TUsersdef is_db_exist(db_url: str):engine = create_engine(db_url, max_overflow=0, pool_size=16, pool_timeout=5, pool_recycle=-1)if not database_exists(engine.url):create_database(engine.url)return Falseelse:    return Truedef init_database(db_url: str):# SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:password@localhost:3306/location_db"# # SQLALCHEMY_DATABASE_URL = "sqlite:///./location_db.db"# 设置连接池的大小engine = create_engine(db_url, max_overflow=0,         # 超过连接池大小外最多创建的连接pool_size=16,           # 连接池大小pool_timeout=5,         # 池中没有线程最多等待的时间，否则报错pool_recycle=-1         # 多久之后对线程池中的线程进行一次连接的回收（重置）)# 创建数据库Base.metadata.create_all(engine)# # 创建数据库连接# _session = sessionmaker(autocommit=False, autoflush=False, bind=engine)()if __name__ == '__main__':# 数据库参考db_host = "127.0.0.1"db_user = "root"db_password = "********"db_url = '%s%s:%s@%s/%s' % ("mysql+pymysql://", db_user, db_password, db_host, "local_db")if not is_db_exist(db_url):init_database(db_url)from account import AlchemyToolalchemytool = AlchemyTool(db_url=db_url)alchemytool.create_user()

database.py

from sqlalchemy import Column, Integer, String, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime, timedelta# sqlalchemy
Base = declarative_base()class TUsers(Base):__tablename__ = 'users'user_id     = Column(String(32), primary_key=True, index=True)user_name   = Column(String(32), nullable=False)def to_dict(self):return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}

account.py

from sqlalchemy import and_, or_, func
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import SingletonThreadPoolfrom database import Base
from database import TUsersclass AlchemyTool(object):def __init__(self, db_url: str):print("AlchemyTool init")# 设置连接池的大小db_config = {"pool_size": 10}#engine = create_engine(SQLALCHEMY_DATABASE_URL, **db_config)# engine = create_engine(db_url, #         #poolclass = SingletonThreadPool,#         connect_args = {'check_same_thread': False}#     )engine = create_engine(db_url, max_overflow=0, pool_size=16, pool_timeout=5, pool_recycle=-1)# # 创建数据库# Base.metadata.create_all(engine)# 创建数据库连接self.__session = sessionmaker(autocommit=False, autoflush=False, bind=engine)()# ############################### User ###############################def create_user(self):try:# db_location = TUsers(**user_base.dict(), user_id=user_id, creator=creator, editor=editor, state=state)db_location = TUsers(user_id="id2222", user_name="name")self.__session.add(db_location)self.__session.commit()except SQLAlchemyError as e:print(f"Error.AlchemyTool.create_user SQLAlchemyError:{str(e)}")self.__session.rollback()db_location = Nonefinally:pass

http://www.mrgr.cn/news/66791.html

相关文章：

  • git撤销commit和add
  • 【android12】【AHandler】【3.AHandler原理篇AHandler类方法全解】
  • Cocos的安装使用
  • 基础大模型 vs 应用大模型
  • Android 依赖统一配置管理（Version Catalogs）
  • 基于LabVIEW应用ARINC 429板卡实现数据通讯——(下篇)
  • 《XGBoost算法的原理推导》12-2 t轮迭代中对样本i的预测值 公式解析
  • DICOM标准：DICOM图像核心属性概念详解——关于参考帧、病人位置、病人方位、图像位置和图像方向、切片位置、图像像素等重要概念解析
  • 双十一热销中 Witsbb健敏思小蓝条钙镁锌霸榜品类TOP1，行业引领者
  • 新版 UTS 2022 版：支持linux上访问同步sqlserver
  • 变脸、看完毒液3，我把自己的脸变毒液了，视频有点惊悚！请谨慎观看
  • Node.js-API 限流与日志优化
  • 线代的几何意义（2）——矩阵乘法，三维及更高维的解释，与非方阵的几何解释
  • FLINK单机版安装部署入门-1
  • 双十一买啥最划算？盘点2024年双十一必买好物！超全选购指南
  • 数据治理项目怎么做，3种推进思路可参考
  • 有哪些“极简风”页面设计的办公协同工具？再不怕眼花缭乱啦！
  • 同是正式编铁饭碗，央国企薪资待遇哪家高？
  • ChatGPT Search：AI 搜索离「谷歌杀手」还有多远？
  • 架构师之路-学渣到学霸历程-44
  • 快乐数算法
  • VC++获取指定进程的路径-支持32位和64位
  • 解决Android Studio 控制台中文乱码
  • R语言生物群落（生态）数据统计分析与绘图丨tidyverse数据清洗、多元统计分析、随机森林、回归及混合效应模型、结构方程模型等
  • 饱和限幅器MATLAB和CODESYS平台下的实现
  • Node.js简介以及安装部署 （基础介绍 一）