python---sqlalchemy(一)

2024-08-31 22:18
文章标签 python sqlalchemy

本文主要是介绍python---sqlalchemy(一)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

      • 执行原生SQL语句
      • 创建删除表
      • 操作数据库表
      • scoped_session
      • 增删改查
      • 其他

执行原生SQL语句

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engineengine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)def task(arg):# 方式一# conn = engine.raw_connection()# cursor = conn.cursor()# cursor.execute(#     "select * from user "# )# result = cursor.fetchall()# cursor.close()# conn.close()# print(result)# print("------ # 方式二----------------")# cur = engine.execute("select * from user")# result = cur.fetchall()# cur.close()# print(result)# print("------ # 方式三----------------")conn = engine.contextual_connect()with conn:cur = conn.execute("select * from user")result = cur.fetchall()print(result)for i in range(5):t = threading.Thread(target=task, args=(i,))t.start()

方式一输出如下：

(('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3))
(('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3))
(('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3))
(('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3))
(('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3))

方式二如下：

[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]

方式三如下：

[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]
[('safly', '123', 1), ('saf', '123', 2), ('alex', '123', 3)]

创建删除表

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationshipBase = declarative_base()# ##################### 单表示例 #########################
class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (# UniqueConstraint('id', 'name', name='uix_id_name'),# Index('ix_id_name', 'name', 'extra'),)class Hosts(Base):__tablename__ = 'hosts'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)ctime = Column(DateTime, default=datetime.datetime.now)# ##################### 一对多示例 #########################
class Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)caption = Column(String(50), default='篮球')class Person(Base):__tablename__ = 'person'nid = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)hobby_id = Column(Integer, ForeignKey("hobby.id"))# 与生成表结构无关，仅用于查询方便hobby = relationship("Hobby", backref='pers')# ##################### 多对多示例 #########################class Server2Group(Base):__tablename__ = 'server2group'id = Column(Integer, primary_key=True, autoincrement=True)server_id = Column(Integer, ForeignKey('server.id'))group_id = Column(Integer, ForeignKey('group.id'))class Group(Base):__tablename__ = 'group'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)# 与生成表结构无关，仅用于查询方便servers = relationship('Server', secondary='server2group', backref='groups')class Server(Base):__tablename__ = 'server'id = Column(Integer, primary_key=True, autoincrement=True)hostname = Column(String(64), unique=True, nullable=False)def init_db():"""根据类创建数据库表:return:"""engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)Base.metadata.create_all(engine)def drop_db():"""根据类删除数据库表:return:"""engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)Base.metadata.drop_all(engine)if __name__ == '__main__':drop_db()init_db()

操作数据库表

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engineBase = declarative_base()
class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (# UniqueConstraint('id', 'name', name='uix_id_name'),# Index('ix_id_name', 'name', 'extra'),)engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)# 每次执行数据库操作时，都需要创建一个session
session = Session()# ############# 执行ORM操作 #############
obj1 = Users(name="safly1")
session.add(obj1)# 提交事务
session.commit()
# 关闭session
session.close()

scoped_session

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_sessionimport datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationshipBase = declarative_base()class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (# UniqueConstraint('id', 'name', name='uix_id_name'),# Index('ix_id_name', 'name', 'extra'),)engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)"""
# 线程安全，基于本地线程实现每个线程用同一个session
# 特殊的：scoped_session中有原来方法的Session中的一下方法："""
session = scoped_session(Session)# ############# 执行ORM操作 #############
obj1 = Users(name="safly")
session.add(obj1)# 提交事务
session.commit()
# 关闭session
session.close()

总结：

SQLAlchemy两种创建session的方式：方式一：import modelsfrom threading import Threadfrom sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_engineengine =create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8day128db?charset=utf8",pool_size=2,max_overflow=0)XXXXXX = sessionmaker(bind=engine)def task():from sqlalchemy.orm.session import Sessionsession = XXXXXX()data = session.query(models.Classes).all()print(data)session.close()for i in range(10):t = Thread(target=task)t.start()方式二（推荐）：import modelsfrom threading import Threadfrom sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom sqlalchemy.orm import scoped_sessionengine =create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8day128db?charset=utf8",pool_size=2,max_overflow=0)XXXXXX = sessionmaker(bind=engine)session = scoped_session(XXXXXX)def task():# 1. 原来的session对象 = 执行session.registry()# 2. 原来session对象.querydata = session.query(models.Classes).all()print(data)session.remove()for i in range(10):t = Thread(target=task)t.start()flask-session默认也是使用的第二种方式：scoped_session

增删改查

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationshipBase = declarative_base()class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (# UniqueConstraint('id', 'name', name='uix_id_name'),# Index('ix_id_name', 'name', 'extra'),)from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import textengine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)session = Session()# ################ 添加add\add_all ################
"""
obj1 = Users(name="wupeiqi")
session.add(obj1)session.add_all([Users(name="wupeiqi"),Users(name="alex"),])
session.commit()
"""# ################ 删除 ################
"""
session.query(Users).filter(Users.id > 2).delete()
session.commit()"""# ################ 修改 ################"""
session.query(Users).filter(Users.id > 0).update({"name" : "099"})
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
session.commit()
"""
# ################ 查询 ################
"""
"""print("-------r1----------")
r1 = session.query(Users).all()for i in r1:print(i.name)print("-------r2----------")r2 = session.query(Users.name, Users.age).all()
for i in r2:print(i)print("-------r3----------")r3 = session.query(Users).filter(Users.name == "099099").all()for i in r3:print(i.name)print("-------r4----------")
r4 = session.query(Users).filter_by(name='099099').all()for i in r4:print(i.name)print("-------r5----------")
r5 = session.query(Users).filter_by(name='099099').first()print(r5.name)print("-------r6----------")
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=5, name='099099').order_by(Users.id).all()
for i in r6:print(i.name)print("-------r7----------")
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='099099').all()
for i in r7:print(i.name)session.close()

其他

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationshipBase = declarative_base()class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (# UniqueConstraint('id', 'name', name='uix_id_name'),# Index('ix_id_name', 'name', 'extra'),)class Hosts(Base):__tablename__ = 'hosts'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)ctime = Column(DateTime, default=datetime.datetime.now)from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import textengine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/test", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)session = Session()################################  　条件###############################
ret = session.query(Users).filter(Users.id > 1, Users.name == 'safly').all()ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == '099099').all()ret = session.query(Users).filter(Users.id.in_([1, 3, 4, 10])).all()ret = session.query(Users).filter(~Users.id.in_([1, 3, 4])).all()ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='099099'))).all()from sqlalchemy import and_, or_ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'xiaowan')).all()ret = session.query(Users).filter(or_(Users.id > 11, Users.name == 'eric')).all()ret = session.query(Users).filter(or_(Users.id > 2,and_(Users.name == 'eric', Users.id > 3),Users.extra != "")).all()# ################################  通配符###############################
ret = session.query(Users).filter(Users.name.like('09%')).all()ret = session.query(Users).filter(~Users.name.like('09%')).all()#
# ############################### 切片###############################
ret = session.query(Users)[1:4]################################  排序###############################
ret = session.query(Users).order_by(Users.name.desc()).all()ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()# ############################### 分组###############################
from sqlalchemy.sql import funcret = session.query(Users).group_by(Users.age).all()ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).all()ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) > 2).all()############################### 连表###############################
# (<__main__.Users object at 0x05934670>, <__main__.Hosts object at 0x05934530>)
# (<__main__.Users object at 0x05934610>, <__main__.Hosts object at 0x05934650>)
ret = session.query(Users, Hosts).filter(Users.id == Hosts.id).all()# for i in ret:
#     print(i)# ??
ret = session.query(Users).join(Hosts).all()
print(ret)
# ret = session.query(Users).join(Hosts, isouter=True).all()################################  组合###############################
# [('safly',), ('xiaowan',), ('sdfsf',), ('werwrw',), ('www',), ('wwerwerw',)]
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Hosts.name).filter(Hosts.id < 4)
ret = q1.union(q2).all()
print(ret)# [('safly',), ('xiaowan',), ('sdfsf',), ('werwrw',), ('www',), ('wwerwerw',), ('safly',)]
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Hosts.name).filter(Hosts.id < 4)
ret = q1.union_all(q2).all()
print(ret)session.close()

这篇关于python---sqlalchemy(一)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

一文带你搞懂Python中__init__.py到底是什么

一文带你搞懂Python中__init__.py到底是什么

《一文带你搞懂Python中__init__.py到底是什么》朋友们,今天我们来聊聊Python里一个低调却至关重要的文件——__init__.py,有些人可能听说过它是“包的标志”,也有人觉得它“没... 目录先搞懂 python 模块（module）Python 包（package）是啥？那么 __in
阅读更多...
使用Python实现图像LBP特征提取的操作方法

使用Python实现图像LBP特征提取的操作方法

《使用Python实现图像LBP特征提取的操作方法》LBP特征叫做局部二值模式,常用于纹理特征提取,并在纹理分类中具有较强的区分能力,本文给大家介绍了如何使用Python实现图像LBP特征提取的操作方... 目录一、LBP特征介绍二、LBP特征描述三、一些改进版本的LBP1.圆形LBP算子2.旋转不变的LB
阅读更多...
Python中__init__方法使用的深度解析

Python中__init__方法使用的深度解析

《Python中__init__方法使用的深度解析》在Python的面向对象编程（OOP）体系中,__init__方法如同建造房屋时的奠基仪式——它定义了对象诞生时的初始状态,下面我们就来深入了解下_... 目录一、__init__的基因图谱二、初始化过程的魔法时刻继承链中的初始化顺序self参数的奥秘默认
阅读更多...
Python实现特殊字符判断并去掉非字母和数字的特殊字符

Python实现特殊字符判断并去掉非字母和数字的特殊字符

《Python实现特殊字符判断并去掉非字母和数字的特殊字符》在Python中,可以通过多种方法来判断字符串中是否包含非字母、数字的特殊字符,并将这些特殊字符去掉,本文为大家整理了一些常用的,希望对大家... 目录1. 使用正则表达式判断字符串中是否包含特殊字符去掉字符串中的特殊字符2. 使用 str.isa
阅读更多...
python中各种常见文件的读写操作与类型转换详细指南

python中各种常见文件的读写操作与类型转换详细指南

《python中各种常见文件的读写操作与类型转换详细指南》这篇文章主要为大家详细介绍了python中各种常见文件（txt,xls,csv,sql,二进制文件）的读写操作与类型转换,感兴趣的小伙伴可以跟... 目录1.文件txt读写标准用法1.1写入文件1.2读取文件2. 二进制文件读取3. 大文件读取3.1
阅读更多...
使用Python实现一个优雅的异步定时器

使用Python实现一个优雅的异步定时器

《使用Python实现一个优雅的异步定时器》在Python中实现定时器功能是一个常见需求,尤其是在需要周期性执行任务的场景下,本文给大家介绍了基于asyncio和threading模块,可扩展的异步定... 目录需求背景代码1. 单例事件循环的实现2. 事件循环的运行与关闭3. 定时器核心逻辑4. 启动与停
阅读更多...
基于Python实现读取嵌套压缩包下文件的方法

基于Python实现读取嵌套压缩包下文件的方法

《基于Python实现读取嵌套压缩包下文件的方法》工作中遇到的问题,需要用Python实现嵌套压缩包下文件读取,本文给大家介绍了详细的解决方法,并有相关的代码示例供大家参考,需要的朋友可以参考下... 目录思路完整代码代码优化思路打开外层zip压缩包并遍历文件：使用with zipfile.ZipFil
阅读更多...
Python处理函数调用超时的四种方法

Python处理函数调用超时的四种方法

《Python处理函数调用超时的四种方法》在实际开发过程中,我们可能会遇到一些场景,需要对函数的执行时间进行限制,例如,当一个函数执行时间过长时,可能会导致程序卡顿、资源占用过高,因此,在某些情况下,... 目录前言func-timeout1. 安装 func-timeout2. 基本用法自定义进程subp
阅读更多...
Python实现word文档内容智能提取以及合成

Python实现word文档内容智能提取以及合成

《Python实现word文档内容智能提取以及合成》这篇文章主要为大家详细介绍了如何使用Python实现从10个左右的docx文档中抽取内容,再调整语言风格后生成新的文档,感兴趣的小伙伴可以了解一下... 目录核心思路技术路径实现步骤阶段一：准备工作阶段二：内容提取 (python 脚本)阶段三：语言风格调
阅读更多...
Python结合PyWebView库打造跨平台桌面应用

Python结合PyWebView库打造跨平台桌面应用

《Python结合PyWebView库打造跨平台桌面应用》随着Web技术的发展,将HTML/CSS/JavaScript与Python结合构建桌面应用成为可能,本文将系统讲解如何使用PyWebView... 目录一、技术原理与优势分析1.1 架构原理1.2 核心优势二、开发环境搭建2.1 安装依赖2.2 验
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2