【Python系列】SQLAlchemy 基本介绍

2024-09-01 00:20
文章标签 python 介绍 系列 基本 sqlalchemy

本文主要是介绍【Python系列】SQLAlchemy 基本介绍，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。
img

  • 推荐:kwan 的首页,持续学习,不断总结,共同进步,活到老学到老
  • 导航
    • 檀越剑指大厂系列:全面总结 java 核心技术,jvm,并发编程 redis,kafka,Spring,微服务等
    • 常用开发工具系列:常用的开发工具,IDEA,Mac,Alfred,Git,typora 等
    • 数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
    • 新空间代码工作室:提供各种软件服务,承接各种毕业设计,毕业论文等
    • 懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作
    • 数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

博客目录

    • 一.基础介绍
      • 1. SQLAlchemy 的起源
      • 2. SQLAlchemy 的核心组件
        • 2.1 核心 SQL 工具包
        • 2.2 ORM 层
      • 3. SQLAlchemy 的优势
        • 3.1 灵活性
        • 3.2 跨数据库支持
        • 3.3 强大的社区支持
    • 二.实战步骤
      • 1.数据库配置
      • 2.model
      • 3.连接配置
      • 4.调用 SQL

一.基础介绍

SQLAlchemy 是一个 Python 的 SQL 工具包和对象关系映射（ORM）工具，它提供了一个高层的 ORM 以及底层的 SQL 表达式语言。SQLAlchemy 是开源的，并且可以在商业和非商业项目中免费使用。它支持多种数据库系统，包括 PostgreSQL、MySQL、SQLite 等。
在这里插入图片描述

1. SQLAlchemy 的起源

SQLAlchemy 最初由 Michael Bayer 在 2005 年创建，目的是提供一个全面的 SQL 工具包和 ORM 解决方案，以满足 Python 社区的需求。随着时间的推移，SQLAlchemy 不断发展和完善，成为了 Python 数据库编程领域中最受欢迎的库之一。

2. SQLAlchemy 的核心组件

2.1 核心 SQL 工具包

SQLAlchemy 的核心 SQL 工具包提供了构建 SQL 查询的功能，它允许开发者以 Pythonic 的方式编写 SQL 语句。这包括了对数据库表的创建、数据的增删改查等操作。

2.2 ORM 层

ORM（Object-Relational Mapping）层是 SQLAlchemy 的另一个重要组成部分，它允许开发者使用 Python 类和对象来表示数据库中的表和行。ORM 层抽象了数据库操作，使得开发者可以不必编写 SQL 语句，而是通过操作 Python 对象来间接地与数据库交互。

3. SQLAlchemy 的优势

3.1 灵活性

SQLAlchemy 提供了灵活的 SQL 构建工具，开发者可以自由地编写 SQL 语句，同时也可以利用 ORM 层提供的抽象来简化数据库操作。

3.2 跨数据库支持

SQLAlchemy 支持多种数据库系统，这意味着开发者可以使用相同的代码库来操作不同的数据库，而不需要为每种数据库编写特定的代码。

3.3 强大的社区支持

由于 SQLAlchemy 的流行，它拥有一个活跃的社区，开发者可以在社区中找到大量的资源和帮助，包括文档、教程和第三方库。

二.实战步骤

1.数据库配置

# 数据库
database:TYPE: mysqlDATABASE_URL: mysql://root:xxx@xxxx:9306/test?serverTimezone=Asia/ShanghaiUSERNAME: rootPASSWORD: xxxHOST: xxxxPORT: 9306DBNAME: testMAX_OVERFLOW: 60POOL_TIMEOUT: 120POOL_SIZE: 30URL_PROPERTY: ?charset=utf8ECHO: True

2.model

from datetime import datetimeimport pytz
from sqlalchemy import String, Column, Text, DateTime, JSON
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, attributesdef get_beijing_now():# 获取当前系统时区return datetime.now(pytz.timezone('Asia/Shanghai'))# 基类
class Base(AsyncAttrs, DeclarativeBase):id: Mapped[int] = mapped_column(primary_key=True)create_time = Column(DateTime, default=get_beijing_now, nullable=False)update_time = Column(DateTime, default=get_beijing_now, onupdate=get_beijing_now, nullable=False)def to_dict(self):"""转为字典输出:return:"""return {c.name: getattr(self, c.name) for c in self.__table__.columns}@repr_generator
class AlchemyEntitySchemas(Base):__tablename__ = "entity_schemas"name = Column(String(255), nullable=False, comment='名称')

3.连接配置

from sqlalchemy.pool import QueuePool
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.sql import text
from base.config import get_config_key
from urllib.parse import quote_plus as urlquote
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, AsyncEngine, async_sessionmakerclass Database:def __init__(self, url, pool_size=30, pool_timeout=1200, max_overflow=60, echo=False):try:self.engine = create_engine(url, poolclass=QueuePool, pool_size=pool_size,max_overflow=max_overflow, pool_timeout=pool_timeout,echo=echo, pool_recycle=7200, pool_pre_ping=True, echo_pool=echo)self.Session = sessionmaker(bind=self.engine, expire_on_commit=False, autocommit=False, autoflush=False)print("Database connected successfully.")except SQLAlchemyError as e:print(f"Error connecting to the database: {e}")def get_session(self):return self.Session()@staticmethoddef close_session(_session):_session.close()@staticmethoddef execute_query(query, _session):try:result = _session.execute(query)return result.fetchall()except SQLAlchemyError as e:print(f"Error executing query: {e}")return Nonefinally:Database.close_session(_session)class SyncDatabase:async_engine: AsyncEngine = Noneasync_session = Nonedef __init__(self, url, pool_size=30, pool_timeout=1200, max_overflow=60, echo=False):self.url = urlself.max_overflow = max_overflowself.pool_timeout = pool_timeoutself.pool_size = pool_sizeself.echo = echoself.connect()def connect(self):"""创建数据库引擎和会话类"""try:self.async_engine = create_async_engine(self.url, echo=self.echo, pool_size=pool_size,max_overflow=max_overflow, pool_timeout=pool_timeout,pool_recycle=7200,pool_pre_ping=True, echo_pool=self.echo)self.async_session = async_sessionmaker(bind=self.async_engine, class_=AsyncSession, expire_on_commit=False,autocommit=False, autoflush=False)print("Database connected successfully.")except SQLAlchemyError as e:print(f"Error connecting to the database: {e}")def get_db_url():userName = get_config_key("database", "USERNAME")password = get_config_key("database", "PASSWORD")dbHost = get_config_key("database", "HOST")dbPort = get_config_key("database", "PORT")dbName = get_config_key("database", "DBNAME")urlProperty = get_config_key("database", "URL_PROPERTY")if dbName is None:return f'mysql+pymysql://{userName}:{urlquote(password)}@{dbHost}:{dbPort}{urlProperty}'else:return f'mysql+pymysql://{userName}:{urlquote(password)}@{dbHost}:{dbPort}/{dbName}{urlProperty}'def get_sync_db_url():userName = get_config_key("database", "USERNAME")password = get_config_key("database", "PASSWORD")dbHost = get_config_key("database", "HOST")dbPort = get_config_key("database", "PORT")dbName = get_config_key("database", "DBNAME")urlProperty = get_config_key("database", "URL_PROPERTY")if dbName is None:return f'mysql+aiomysql://{userName}:{urlquote(password)}@{dbHost}:{dbPort}{urlProperty}'else:return f'mysql+aiomysql://{userName}:{urlquote(password)}@{dbHost}:{dbPort}/{dbName}{urlProperty}'url = get_db_url()
max_overflow = get_config_key("database", "MAX_OVERFLOW")
pool_timeout = get_config_key("database", "POOL_TIMEOUT")
pool_size = get_config_key("database", "POOL_SIZE")
echo = get_config_key("database", "ECHO")# sqlalchemy实际操作对象，导入的时候应该导入这个对象
get_sqlalchemy_db = Database(url, pool_size=pool_size, pool_timeout=pool_timeout, max_overflow=max_overflow, echo=echo)# 异步的
SYNC_DB_URI = get_sync_db_url()
_async_engine = create_async_engine(SYNC_DB_URI, echo=echo, pool_size=pool_size,max_overflow=max_overflow, pool_timeout=pool_timeout, pool_recycle=7200,pool_pre_ping=True, echo_pool=echo)
# 异步IO的 sqlalchemy实际操作对象，导入的时候应该导入这个对象
async_session_factory = async_sessionmaker(bind=_async_engine, class_=AsyncSession, expire_on_commit=False,autocommit=False, autoflush=False)

在这里插入图片描述

4.调用 SQL

@staticmethod
async def find_by_name(name: str):"""根据名称查询"""db = get_sqlalchemy_dbtry:with Session(db.engine) as session:stmt = select(AlchemySchemas)if name:stmt = stmt.where(AlchemySchemas.name == name)schemas_infos = session.scalars(stmt).all()return [schemas_info.to_dict() for schemas_info in schemas_infos] if schemas_infos else Noneexcept SQLAlchemyError as e:print(f"An error occurred: {e}")return Nonefinally:db.close_session(session)

觉得有用的话点个赞 👍🏻 呗。
❤️❤️❤️本人水平有限，如有纰漏，欢迎各位大佬评论批评指正！😄😄😄

💘💘💘如果觉得这篇文对你有帮助的话，也请给个点赞、收藏下吧，非常感谢!👍 👍 👍

🔥🔥🔥Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧！🌙🌙🌙

img

这篇关于【Python系列】SQLAlchemy 基本介绍的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1125414。

相关文章

Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
性能测试介绍

性能测试介绍

性能测试是一种测试方法，旨在评估系统、应用程序或组件在现实场景中的性能表现和可靠性。它通常用于衡量系统在不同负载条件下的响应时间、吞吐量、资源利用率、稳定性和可扩展性等关键指标。 为什么要进行性能测试 通过性能测试，可以确定系统是否能够满足预期的性能要求，找出性能瓶颈和潜在的问题，并进行优化和调整。 发现性能瓶颈：性能测试可以帮助发现系统的性能瓶颈，即系统在高负载或高并发情况下可能出现的问题
阅读更多...
水位雨量在线监测系统概述及应用介绍

水位雨量在线监测系统概述及应用介绍

在当今社会，随着科技的飞速发展，各种智能监测系统已成为保障公共安全、促进资源管理和环境保护的重要工具。其中，水位雨量在线监测系统作为自然灾害预警、水资源管理及水利工程运行的关键技术，其重要性不言而喻。 一、水位雨量在线监测系统的基本原理 水位雨量在线监测系统主要由数据采集单元、数据传输网络、数据处理中心及用户终端四大部分构成，形成了一个完整的闭环系统。 数据采集单元：这是系统的“眼睛”，
阅读更多...
python: 多模块(.py)中全局变量的导入

python: 多模块(.py)中全局变量的导入

文章目录 global关键字可变类型和不可变类型数据的内存地址单模块(单个py文件)的全局变量示例总结 多模块(多个py文件)的全局变量from x import x导入全局变量示例 import x导入全局变量示例 总结 global关键字 global 的作用范围是模块(.py)级别： 当你在一个模块（文件）中使用 global 声明变量时，这个变量只在该模块的全局命名空
阅读更多...
Hadoop数据压缩使用介绍

Hadoop数据压缩使用介绍

一、压缩原则 （1）运算密集型的Job，少用压缩 （2）IO密集型的Job，多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器 2）要在Hadoop中启用压缩，可以配置如下参数
阅读更多...
基本知识点

基本知识点

1、c++的输入加上ios::sync_with_stdio(false);  等价于 c的输入，读取速度会加快（但是在字符串的题里面和容易出现问题） 2、lower_bound()和upper_bound() iterator lower_bound( const key_type &key ): 返回一个迭代器，指向键值>= key的第一个元素。 iterator upper_bou
阅读更多...
【Python编程】Linux创建虚拟环境并配置与notebook相连接

【Python编程】Linux创建虚拟环境并配置与notebook相连接

1.创建 使用 venv 创建虚拟环境。例如，在当前目录下创建一个名为 myenv 的虚拟环境： python3 -m venv myenv 2.激活 激活虚拟环境使其成为当前终端会话的活动环境。运行： source myenv/bin/activate 3.与notebook连接 在虚拟环境中，使用 pip 安装 Jupyter 和 ipykernel： pip instal
阅读更多...
科研绘图系列：R语言扩展物种堆积图（Extended Stacked Barplot）

科研绘图系列：R语言扩展物种堆积图（Extended Stacked Barplot）

介绍 R语言的扩展物种堆积图是一种数据可视化工具，它不仅展示了物种的堆积结果，还整合了不同样本分组之间的差异性分析结果。这种图形表示方法能够直观地比较不同物种在各个分组中的显著性差异，为研究者提供了一种有效的数据解读方式。 加载R包 knitr::opts_chunk$set(warning = F, message = F)library(tidyverse)library(phyl
阅读更多...
【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

引言 高斯过程（Gaussian Process，简称GP）是一种概率模型，用于描述一组随机变量的联合概率分布，其中任何一个有限维度的子集都具有高斯分布 文章目录 引言一、高斯过程1.1 基本定义1.1.1 随机过程1.1.2 高斯分布 1.2 高斯过程的特性1.2.1 联合高斯性1.2.2 均值函数1.2.3 协方差函数（或核函数） 1.3 核函数1.4 高斯过程回归（Gauss
阅读更多...
图神经网络模型介绍（1）

图神经网络模型介绍（1）

我们将图神经网络分为基于谱域的模型和基于空域的模型，并按照发展顺序详解每个类别中的重要模型。 1.1基于谱域的图神经网络         谱域上的图卷积在图学习迈向深度学习的发展历程中起到了关键的作用。本节主要介绍三个具有代表性的谱域图神经网络：谱图卷积网络、切比雪夫网络和图卷积网络。 （1）谱图卷积网络 卷积定理：函数卷积的傅里叶变换是函数傅里叶变换的乘积，即F{f*g}
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2