1 sqlalchemy快速使用
2 sqlalchemy原生操作
3 sqlalchemy操作表
3.2 基本增删查改
4 一对多关系
4.1 关系建立
4.2 增加和基于对象的跨表查询
5 多对多关系
5.2 增加和基于对象跨表查
6 scoped线程安全
7 g对象
8 基本增查改
7 常用查询
1 sqlalchemy快速使用
1.1 介绍
# SQLAlchemy是一个基于Python实现的ORM框架,对象关系映射# orm 框架----》django orm--》只能用在django中,不能独立使用# python界的orm框架-peewee:小型-sqlalchemy:企业级,支持异步,用在异步框架中-djagno ormm-Tortoise ORM-GINO# go 界orm框架-gorm 国人写的-Xorm# java界orm框架-ssh 框架springmvc structs Hibernate(java的orm框架)-ssh spring springmvc Hibernate-ssm Spring SpringMVC MyBatis (orm框架)-springboot :sb框架 ---》java工程师就是spring工程师-spring cloud:dubbo。。。# 分层:
Engine,框架的引擎
Connection Pooling ,数据库连接池
Dialect,选择连接数据库的DB API种类(sqlite,mysql...)
Schema/Types,架构和类型
SQL Exprression Language,SQL表达式语言#操作不同数据库
MySQL-Pythonmysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>pymysqlmysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]MySQL-Connectormysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>cx_Oracleoracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]更多:http://docs.sqlalchemy.org/en/latest/dialects/index.html# 了解orm不能创建数据库---》只能创建表,删除表---》sqlalchemy不能增加删除字段--》借助于第三方插件实现(flask-migrate)django orm 不能创建数据库,创建,删除表,增加字段,删除字段可以独立使用,也可以集成到 web项目中-flask-fastapi:
2 sqlalchemy原生操作
# 1 导入
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
import pymysql# 2 创建engine对象
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/cnblogs",max_overflow=0, # 超过连接池大小外最多创建的连接pool_size=5, # 连接池大小pool_timeout=30, # 池中没有线程最多等待的时间,否则报错pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)# 3 通过engine获得conn,cursor
conn = engine.raw_connection() # 拿到连接对象
cursor = conn.cursor()
# 4 具体操作
cursor.execute('select * from article limit 10')
print(cursor.fetchall())
cursor.close()
conn.close()
3 sqlalchemy操作表
# 在py中写一个类,通过sqlalchemy,同步到数据库,生成一张表
# python中的一个对象,保存到数据库中,形成一条记录
3.2 基本增删查改
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Users# 1 增 ,删 ,查 ,该操作# 1 生成 engine
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/sqlalchemy01?charset=utf8",max_overflow=0, # 超过连接池大小外最多创建的连接pool_size=5, # 连接池大小pool_timeout=30, # 池中没有线程最多等待的时间,否则报错pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)# 2 orm操作,拿到一个 session 对象
Session = sessionmaker(bind=engine) # 得到Session类
session = Session() # 得到对象
# 3 以后都通过session操作
# 3.1 创建一个User对象,新增到数据库中
# user = Users(name='lq1z', email='302@qq.com')
# session.add(user)
# session.commit()# 3.2 查--->最多的,最复杂的
# select * from Users where name=lqz;
# user=session.query(Users).filter_by(name='lqz').first()
# user=session.query(Users).filter_by(id=1).first()
# print(user.name)
# print(user.email)# 3.3 删除
# res=session.query(Users).filter_by(name='lqz').delete() # 影响的行数
# print(res)
# session.commit()# user=session.query(Users).filter_by(name='lqz').first()
# # user.delete() # 单个对象,不能这么删:'Users' object has no attribute 'delete'
# session.delete(user) #需要使用session删
# session.commit()# 3.4 修改
# res = session.query(Users).filter_by(name='lqz').update({'email':'333@qq.com'})
# session.commit()
# print(res)user = session.query(Users).filter_by(name='lqz').first()
user.email='999@qq.com'
session.add(user) # 如果 有id,就是修改
session.commit()session.close()
4 一对多关系
4.1 关系建立
# 1 导入一些模块
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, \relationship # 新的变成它,老的会有 from sqlalchemy.ext.declarative import declarative_base Warningfrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index# django orm,所有类,都要继承一个父类,sqlalchemy中这个父类就是Base,有点特殊,正常导入一个类,继承即可,但是它不是# 2 得到所有表模型需要继承 基类
Base = declarative_base()#### 一对多关系 一个爱好,有很多人喜欢
class Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)caption = Column(String(50), default='篮球')def __str__(self):return self.captionclass Person(Base):__tablename__ = 'person'id = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)# hobby指的是tablename而不是类名,uselist=Falsehobby_id = Column(Integer, ForeignKey("hobby.id")) # 关联字段写在多的一方# 跟数据库无关,不会新增字段,只用于快速链表操作# 类名,backref用于反向查询hobby = relationship('Hobby', backref='pers') # 等同于django中的relate_name,反向查询替换表名小写的def __str__(self):return self.namedef __repr__(self):return self.name# 4 创建 engine,需要手动创建数据库
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/sqlalchemy01?charset=utf8",max_overflow=0, # 超过连接池大小外最多创建的连接pool_size=5, # 连接池大小pool_timeout=30, # 池中没有线程最多等待的时间,否则报错pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 5 通过 engine 把上面的表模型,创建到数据库中
Base.metadata.create_all(engine)# 6 删除被Base管理的所有表
# Base.metadata.drop_all(engine)
4.2 增加和基于对象的跨表查询
#### 一对多关系新增和查询
### 新增,笨办法
# hobby=Hobby(caption='足球')
# session.add(hobby)
# person=Person(name='lqz',hobby_id=1) # hobby_id=1 只能放数字
# session.add(person)# 新增:高级
# hobby=Hobby(caption='橄榄球')
# person=Person(name='彭于晏',hobby=hobby) # hobby 放对象,用对象做关联,但是必须用 relationship
# session.add(person)
# session.commit()
# session.close()# 查询:正向和反向---》基于对象的跨表查询### 正向查询,Person中必须有 hobby这个字段,使用relationship
# person=session.query(Person).filter_by(name='彭于晏').first()
# print(person.name)
# # 拿到这个人的爱好
# # person.hobby_id # 爱好id号---》通过id可以查出爱好
# print(person.hobby)## 反向 查出所有喜欢橄榄球的人
hobby=session.query(Hobby).filter_by(caption='橄榄球').first()
# pers=session.query(Person).filter_by(hobby_id=hobby.id).all()
pers=hobby.pers
print(pers)
4.3 一对一关系,就是一对多,只不过多的一方只有一条
...
5 多对多关系
# 1 导入一些模块
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, \relationship # 新的变成它,老的会有 from sqlalchemy.ext.declarative import declarative_base Warningfrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index# django orm,所有类,都要继承一个父类,sqlalchemy中这个父类就是Base,有点特殊,正常导入一个类,继承即可,但是它不是# 2 得到所有表模型需要继承 基类
Base = declarative_base()#### 多对多
class Boy2Girl(Base):__tablename__ = 'boy2girl'id = Column(Integer, primary_key=True, autoincrement=True)girl_id = Column(Integer, ForeignKey('girl.id'))boy_id = Column(Integer, ForeignKey('boy.id'))ctime = Column(DateTime, default=datetime.datetime.now) # 约会时间class Girl(Base):__tablename__ = 'girl'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)# boy = relationship('Boy', secondary='boy2girl', backref='girls')def __repr__(self):return self.name
class Boy(Base):__tablename__ = 'boy'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(64), unique=True, nullable=False)# 与生成表结构无关,仅用于查询方便,放在哪个单表中都可以,这个字段放boy或girl表都可以girls = relationship('Girl', secondary='boy2girl', backref='boys')def __repr__(self):return self.name# 4 创建 engine,需要手动创建数据库
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/sqlalchemy01?charset=utf8",max_overflow=0, # 超过连接池大小外最多创建的连接pool_size=5, # 连接池大小pool_timeout=30, # 池中没有线程最多等待的时间,否则报错pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 5 通过 engine 把上面的表模型,创建到数据库中
Base.metadata.create_all(engine)# 6 删除被Base管理的所有表
# Base.metadata.drop_all(engine)
5.2 增加和基于对象跨表查
### 多对多关系新增
# 笨办法,新增boy和girl,新增约会
# boy = Boy(name='张三')
# girl = Girl(name='小红')
# session.add_all([boy, girl])
# 新增约会
# b=Boy2Girl(girl_id=1,boy_id=1)
# session.add(b)
# session.commit()# 高级一点
# boy = Boy(name='张五')
# girl = Girl(name='大红红红')
# # 查出小红
# xh=session.query(Girl).filter_by(name='小红').first()
# boy.girls = [girl,xh ]
# session.add(boy)
# session.commit()# 查询
# 查询:张五跟谁约过
zw=session.query(Boy).filter_by(name='张五').first()
print(zw.girls) # 基于对象的跨表的正向查询# 反向查询:小红,跟谁约过
xh=session.query(Girl).filter_by(name='小红').first()
print(xh.boys) # 基于对象的跨表的正向查询
6 scoped线程安全
# session 对象,集成到flask中去,要把session对象做成全局(大家公用),还是每个视图函数独有一个(没次都要实例化得到这个session对象)# 每个视图函数独有一个---》每次都要实例化---》sqlalchemy提供了一种方式,让咱们使用全局的一个session,但是每个视图函数中使用的都是不同的 request,session都是这种实现机制# sqlalchemy提供了一种,在不同线程中,虽然使用全局 session,实际上每个线程自己独有一个session
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Users
from threading import local
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/sqlalchemy01", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)"""
# 线程安全,基于本地线程实现每个线程用同一个session
# 特殊的:scoped_session中有原来方法的Session中的一下方法:public_methods = ('__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested','close', 'commit', 'connection', 'delete', 'execute', 'expire','expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind','is_modified', 'bulk_save_objects', 'bulk_insert_mappings','bulk_update_mappings','merge', 'query', 'refresh', 'rollback','scalar'
)
"""
# session=Session() # 不是线程安全---》多线程并发情况下---》用的是同一个,就会出现混乱
#scoped_session类并没有继承Session,但是却又它的所有方法# 以后在多线程,协程情况下,使用这个session,就没有并发安全的问题
# 原理是什么? 内部使用了threading包下local对象, local.session 在不同线程下执行这个取值或赋值,使用的都是当前线程自己的# 线程1 local.a=100 后续的操作中--->取local.a--->永远是当时这条线程中放进去的a ,也就是100
# 线程2 local.a=999 --->取local.a--->永远是当时这条线程中放进去的a ,也就是999#local对象如何实现的线程安全呢? 内部维护了一个字典,字典的key是线程id号,value值是
# l=local() # {}
# # 线程1
# l.session=线程1的session # {线程1的id号:{session:新session}}
#
# # 线程2
# l.session# {线程1的id号:{session:新session},线程2的id号:{session:线程2的session}}# 本质就是,不同线程,使用自己线程的那个session
# scoped_session 不是Session类的对象,但是他有 Session类对象的所有方法----》通过装饰器,装进去的
session = scoped_session(Session)
# ############# 执行ORM操作 #############
obj1 = Users(name="lqz4",email='3533@qq.com')
session.add(obj1)# 提交事务
session.commit()
# 关闭session
session.close()
7 g对象
#flask 中的g对象,上下文对象, global 缩写,由于 global 是关键字---》于是它命名成了g
# django中,有个东西,可以完成它的效果 就是 request,但是使用request会造成数据污染
# flask中也可以用 request,但是也会造成数据污染,所以咱们使用g对象# g和session的区别from flask import Flask,g,request
# g 对象可以放入值,在当次请求下,这个值一直存在
# 它等同于django中 的request
# g和session有什么区别? g只针对于当前请求,session针对于这个用户的所有请求
app = Flask(__name__)@app.before_request
def before():if request.path=='/':g.name='lqz'
def add():print(g.name)@app.route('/')
def index():g.name='lqz'add()return 'inde'if __name__ == '__main__':app.run()
8 基本增查改
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Users
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/sqlalchemy01", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)# ## 1 添加, add add_all## 2 删除
# 1 session.query(Users).filter_by(id=1).delete()
# 2 session.delete(对象)## 3 修改
# 1 方式一:
# session.query(Users).filter_by(id=1).update({'name':'lqz'})
# 2 方式二 类名.属性名,作为要修改的key
# session.query(Users).filter_by(id=4).update({Users.name:'lqz',Users.email:'55@qq.com'})
# id为4的人的名字后+ _nb 类似于django的 F 查询
# session.query(Users).filter_by(id=4).update({'name':Users.name+'_nb'},synchronize_session=False)
# session.query(Users).filter_by(id=4).update({'id':Users.id+6}, synchronize_session="evaluate")
# session.commit()# 3 方式三:
# 对象.name='xxx'
#session.add(对象)### 4 查询---》基本查询
# 4.1 filter_by 写条件
# res=session.query(Users).filter_by(name='lqz').first()
# res=session.query(Users).filter_by(name='lqz').all() # 放在列表中 不是queryset对象# 4.2 filter 写表达式
# res=session.query(Users).filter(Users.name=='lqz').first()
# res=session.query(Users).filter(Users.id>=3).all()
# res=session.query(Users).filter(Users.name!='lqz').all()# 4.3 只查表中某几个字段,并重命名
# select name as xx,age from user;
# res=session.query(Users.name.label('xx'), Users.email).all()
# res=session.query(Users.name, Users.email).all()# 4.4 条件可以使用text自己拼凑
# select * from users where id< 224 and name=lqz order by id
# res = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='lqz').order_by(Users.id).all()## 4.5 直接原生sql
# SELECT * FROM users where name=lqz
# res = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='lqz').all()
# res = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='张三')
# print(res)# 提交事务
session.commit()
# 关闭session
session.close()
9 常用查询
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Users,Person,Hobby,Boy,Girl
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:lqz123?@127.0.0.1:3306/sqlalchemy01", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)#### 5 高级查询--只需要看懂
#5.1 filter_by写 条件 会了
# ret = session.query(Users).filter_by(name='lqz').all()#5.2 表达式,and条件连接 between
# ret = session.query(Users).filter(Users.id > 1, Users.name == 'lqz').all()# 5.3 between
# select * from users where user.id between 4 and 10 and name=lqz;
# ret = session.query(Users).filter(Users.id.between(4, 10), Users.name == 'lqz')# 5.4 in 条件
# ret = session.query(Users).filter(Users.id.in_([1,4,5])).all()# 5.5 ~非,除。。外
# ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()# 5.5 二次筛选
# select * from users where id in (select id from users where name = lqz);
# ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='lqz'))).all()## 5.6 and 和 or条件
from sqlalchemy import and_, or_
# #or_包裹的都是or条件,and_包裹的都是and条件
# ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'lqz')).all()
# ret = session.query(Users).filter(Users.id > 3, Users.name == 'lqz').all() # 根上面一样# select * from users where id<=2 or name =lqz;
# ret = session.query(Users).filter(or_(Users.id <= 2, Users.name == 'lqz')).all()# select * from users where id <2 or (name=lqz and id>3) or extra !='';
# ret = session.query(Users).filter(
# or_(
# Users.id < 2,
# and_(Users.name == 'lqz', Users.id > 3),
# Users.extra != ""
# ))# 5.7 like
# # 通配符,以e开头,不以e开头
# select * from users where name like l%;
# ret = session.query(Users).filter(Users.name.like('l%')).all()
# ret = session.query(Users).filter(~Users.name.like('l%')).all()#5.8 限制,用于分页,区间
# ret = session.query(Users)[1:2] # 去第一条到第二条 其实就是只取 第二条,从0开始# 5.9 排序,根据name降序排列(从大到小)
# ret = session.query(Users).order_by(Users.name.desc()).all()
# ret = session.query(Users).order_by(Users.name.asc())
# ret = session.query(Users).order_by(Users.name).all()# #第一个条件重复后,再按第二个条件升序排
# ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()#5.10 分组
from sqlalchemy.sql import func
# select name from users group by name; 一旦分组,只能查询 分组字段和 聚合函数的字段
# ret = session.query(Users.name).group_by(Users.name).all()# #分组之后取最大id,id之和,最小id 和名字
ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id),func.count(Users.id),Users.name).group_by(Users.name).all()# #haviing筛选
# select max(id),sum(id),min(id),count(id),name from users group by name where id >2 having min(id)>2;
# ret = session.query(
# func.max(Users.id),
# func.sum(Users.id),
# func.min(Users.id),func.count(Users.id),Users.name).filter(Users.id>2).group_by(Users.name).having(func.min(Users.id) >2)# 5.11 连表(默认用forinkey关联)
# select * from person,hobby where user.hobby_id=hobby.id;
# ret = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all()# #join表,默认是inner join
# select * from person inner join hobby on person.hobby_id=hobby.id;
# ret = session.query(Person).join(Hobby).all()# #isouter=True 外连,表示Person left join Hobby,没有右连接,反过来即可
# select * from person left join hobby on person.hobby_id=hobby.id;
# ret = session.query(Person).join(Hobby, isouter=True).all()
# ret = session.query(Hobby).join(Person, isouter=True).all() # 右链接# 没有指定链表条件,默认以外键关联
# # 自己指定on条件(连表条件),第二个参数,支持on多个条件,用and_,同上
# select * from Person left join hobby on person.id=hobby.id; # sql 没有意义,只是讲案例
# ret = session.query(Person).join(Hobby,Person.id==Hobby.id, isouter=True)# 5.12 union和union all
# # 组合(了解)UNION 操作符用于合并两个或多个 SELECT 语句的结果集
# #union和union all的区别?
# q1 = session.query(Boy.id,Boy.name).filter(Boy.id > 1)
# q2 = session.query(Girl.id,Girl.name).filter(Girl.id < 10)
# ret = q1.union(q2).all()
#
# q1 = session.query(Boy.name).filter(Boy.id > 1)
# q2 = session.query(Girl.name).filter(Girl.id < 10)
# ret = q1.union_all(q2).all()
# print(ret)# 提交事务
session.commit()
# 关闭session
session.close()