【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

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/413224.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

5.3二叉树——二叉树链式结构实现

本篇博客梳理二叉树链式结构 明确:二叉树是递归定义的 递归的本质:当前问题子问题,返回条件是最小规模的子问题 一、二叉树的遍历 1.前序、中序与后序遍历 (1)前序:根->左子树->右子树…

书生大模型实战营(1)——InterStudio基础知识+Vscode SSH连接远程服务器+Linux基础指令

参加书生.浦江大模型实战训练营,学习大模型知识和微调技术,所有课程免费,通过闯关的形式学习,也比较有趣。一起来了解LLM的世界。邀请链接 产品简介 InternStudio 是大模型时代下的云端算力平台。基于 InternLM 组织下的诸多算法…

经典文献阅读之--ParkingE2E(基于摄像头的端到端停车网络:从图像到规划)

0. 简介 自动泊车是智能驾驶领域的一项关键任务。传统泊车算法通常采用基于规则的方案来实现。然而,由于算法设计的复杂性,这些方法在复杂的泊车场景中效果欠佳。相比之下,基于神经网络的方法往往比基于规则的方法更加直观且功能多样。通过收…

ORACLE 统计信息的备份与恢复

备份 --需要先创建统计信息基础表 exec dbms_stats.create_stat_table(USER1,STAT_TIMESTAMP); --导出某个用户的所有统计信息 exec dbms_stats.export_schema_stats(USER1,STAT_TIMESTAMP);--测试(插入100条,更新统计信息,略) select num_rows,last_ana…

基于STM32开发的简易自动驾驶系统

目录 引言环境准备工作 硬件准备软件安装与配置系统设计 系统架构硬件连接代码实现 系统初始化传感器数据采集与处理电机控制与转向OLED显示与状态提示Wi-Fi通信与远程监控应用场景 简易自动驾驶演示智能车模型开发与学习常见问题及解决方案 常见问题解决方案结论 1. 引言 随…

蜂鸣器奏乐

一、粗略了解简谱 拍号:如图,“2”表示一个小节有2拍,“4”表示4分音符为一拍 终止线表示歌曲结束 注意:以下音符都按以四分音符为一拍计算拍数 四分音符: 唱一拍 二分音符: 某一个音右边有一个小横线&…

中国招标投标平台JS逆向:DES加密与Python纯算还原

中国招标投标平台JS逆向:DES加密与Python纯算还原 目录 🔐 JS DES解密🧮 Python版本的纯算实现 🔐 JS DES解密 在中国招标投标公共服务平台的分析过程中,发现了数据加密采用了DES算法。DES(数据加密标准&…

github源码指引:C++嵌入式WEB服务器

初级代码游戏的专栏介绍与文章目录-CSDN博客 我的github:codetoys,所有代码都将会位于ctfc库中。已经放入库中我会指出在库中的位置。 这些代码大部分以Linux为目标但部分代码是纯C的,可以在任何平台上使用。 相关专题: C嵌入式…

Broker服务器模块

一.Broker模块介绍 二.Broker模块具体实现 1. 类的成员变量与构造函数 成员变量 事件循环和TCP服务器: muduo::net::EventLoop _baseloop;muduo::net::TcpServer _server; 这些是muduo库提供的核心组件,负责处理网络事件和管理TCP连接。 消息分发和编码: muduo::…

Spring Security 认证源码超详细分析

Spring Security 认证源码超详细分析 认证(Authentication)是系统确认用户信息的重要途径,用户通过认证之后,系统才能明确用户的身份,进而才可以为该用户分配一定的权限,这个过程也叫授权(Auth…

项目实战--Sa-Token详细方案笔记

Sa-Token权限系统设计 一、前言二、认证授权的概念三、Sa-Token简介3.1 Sa-Token使用方式3.2 踢人下线3.3 全局异常处理3.4 二级认证3.5 同端互斥登录3.6 Http Basic/Digest 认证3.6.1 HttpBasic认证3.6.2 Http Digest 认证 四、Sa-Token授权(鉴权)4.1 权…

详说 类和对象

类怎么定义 类是什么呢?类就是我们上篇文说的命名空间,单独创建一个域,自己有自己的生命空间,那么类怎么定义呢?C规定,假设 stack就是他的类名,那么前面要加个class,换行之后就是他…

汽车乘客热舒适度大挑战,如何利用仿真技术提高汽车环境舒适度

舒适性在人们选择汽车的决定性方面占比越来越重,而汽车乘员舱环境的舒适性是指为乘员提供舒适愉快便利的乘坐环境与条件,包括良好的平顺性、车内的低噪声、适宜的空气环境以及良好的驾驶操作性能。 舒适性 经济性 安全性、动力性 典型的乘员舱热舒适性模…

laravel的队列的使用

laravel队列 laravel的特性:laravel队列可以基于不同的后台存储服务提供统一的api,后台存储服务包括 Redis MySQL等。队列实现了业务解耦,异步处理,错误重试的功能。比如调用第三方api,无法保证api的可靠性&#xff0…

Transformer 与传统模型Informer

Transformer 与传统模型:Informer 如何改变时间序列预测的规则 Transformers 是那些聪明的注意力构建者,它们在机器学习的各个领域掀起了波澜。但在时间序列预测领域,它们才真正大显身手。你可能会问,为什么?想象一下,有一个水晶球,它不仅能看到未来,还能理解导致未来的…

TCP协议 配合 Wireshark 分析数据

在TCP连接中,无论是客户端还是服务端,都有可能成为发送端或接收端,这是因为TCP是一个全双工协议,允许数据在同一连接中双向流动 客户端(Client):通常是指主动发起连接请求的一方。例如&#xf…

宠物空气净化器有用吗?为什么养宠家庭要买宠物空气净化器?

身为一个鼻炎患者,却喜欢猫咪,所以毅然决然的养了两只宠物,而且还是长毛猫,不要问为什么鼻炎还买两只猫咪,因为怕一只猫咪孤单,所以养了两只。对于很多人来说,猫咪就像焦虑不安时的精神搭子&…

如何让私域服务赢得用户的心?

私域流量的概念在当今的商业环境中已经变得极为重要,许多品牌和企业都投入大量资源尝试通过各种策略吸引并保留用户。然而,单纯的流量积累并不足以确保商业成功。当面对用户的沉默、缺乏活跃度以及无法变现的困境时,我们必须重新审视私域流量…

语音控制开关的语音识别ic芯片方案

语音控制开关是一种基于语音识别技术的设备,它通过内置的语音识别芯片,将用户的语音指令转化为电信号,从而实现对设备的控制。例如在智能家居设备上的应用,通常需要连接到家庭的Wi-Fi网络上,以便与智能手机或智能音箱等…

Java之初始泛型

1 包装类 在Java中,由于基本类型不是继承自Object,为了在泛型代码中可以支持基本类型,Java给每个基本类型都对应了一个包装类型。 1.1 基本数据类型和对应的包装类 基本数据类型包装类byteByteshortShortintIntegerlongLongfloatFloatdoub…