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

相关文章

Conda与Python venv虚拟环境的区别与使用方法详解

《Conda与Pythonvenv虚拟环境的区别与使用方法详解》随着Python社区的成长,虚拟环境的概念和技术也在不断发展,:本文主要介绍Conda与Pythonvenv虚拟环境的区别与使用... 目录前言一、Conda 与 python venv 的核心区别1. Conda 的特点2. Python v

Python使用python-can实现合并BLF文件

《Python使用python-can实现合并BLF文件》python-can库是Python生态中专注于CAN总线通信与数据处理的强大工具,本文将使用python-can为BLF文件合并提供高效灵活... 目录一、python-can 库:CAN 数据处理的利器二、BLF 文件合并核心代码解析1. 基础合

Python使用OpenCV实现获取视频时长的小工具

《Python使用OpenCV实现获取视频时长的小工具》在处理视频数据时,获取视频的时长是一项常见且基础的需求,本文将详细介绍如何使用Python和OpenCV获取视频时长,并对每一行代码进行深入解析... 目录一、代码实现二、代码解析1. 导入 OpenCV 库2. 定义获取视频时长的函数3. 打开视频文

Python中你不知道的gzip高级用法分享

《Python中你不知道的gzip高级用法分享》在当今大数据时代,数据存储和传输成本已成为每个开发者必须考虑的问题,Python内置的gzip模块提供了一种简单高效的解决方案,下面小编就来和大家详细讲... 目录前言:为什么数据压缩如此重要1. gzip 模块基础介绍2. 基本压缩与解压缩操作2.1 压缩文

Python设置Cookie永不超时的详细指南

《Python设置Cookie永不超时的详细指南》Cookie是一种存储在用户浏览器中的小型数据片段,用于记录用户的登录状态、偏好设置等信息,下面小编就来和大家详细讲讲Python如何设置Cookie... 目录一、Cookie的作用与重要性二、Cookie过期的原因三、实现Cookie永不超时的方法(一)

Python内置函数之classmethod函数使用详解

《Python内置函数之classmethod函数使用详解》:本文主要介绍Python内置函数之classmethod函数使用方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地... 目录1. 类方法定义与基本语法2. 类方法 vs 实例方法 vs 静态方法3. 核心特性与用法(1编程客

Python函数作用域示例详解

《Python函数作用域示例详解》本文介绍了Python中的LEGB作用域规则,详细解析了变量查找的四个层级,通过具体代码示例,展示了各层级的变量访问规则和特性,对python函数作用域相关知识感兴趣... 目录一、LEGB 规则二、作用域实例2.1 局部作用域(Local)2.2 闭包作用域(Enclos

Python实现对阿里云OSS对象存储的操作详解

《Python实现对阿里云OSS对象存储的操作详解》这篇文章主要为大家详细介绍了Python实现对阿里云OSS对象存储的操作相关知识,包括连接,上传,下载,列举等功能,感兴趣的小伙伴可以了解下... 目录一、直接使用代码二、详细使用1. 环境准备2. 初始化配置3. bucket配置创建4. 文件上传到os

使用Python实现可恢复式多线程下载器

《使用Python实现可恢复式多线程下载器》在数字时代,大文件下载已成为日常操作,本文将手把手教你用Python打造专业级下载器,实现断点续传,多线程加速,速度限制等功能,感兴趣的小伙伴可以了解下... 目录一、智能续传:从崩溃边缘抢救进度二、多线程加速:榨干网络带宽三、速度控制:做网络的好邻居四、终端交互

Python中注释使用方法举例详解

《Python中注释使用方法举例详解》在Python编程语言中注释是必不可少的一部分,它有助于提高代码的可读性和维护性,:本文主要介绍Python中注释使用方法的相关资料,需要的朋友可以参考下... 目录一、前言二、什么是注释?示例:三、单行注释语法:以 China编程# 开头,后面的内容为注释内容示例:示例:四