Engine，框架的引擎
Connection Pooling ，数据库连接池
Dialect，选择连接数据库的DB API种类（sqlite,mysql...）
Schema/Types，架构和类型
SQL Exprression Language，SQL表达式语言
操作不同数据库
MySQL-Python mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> pymysql mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
例如：
mysql+pymysql://root:lqz123?@127.0.0.1:3306/cnblogs MySQL-Connector mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> cx_Oracle oracle+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
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()
sqlalchemy操作
在py中写一个类，通过sqlalchemy，同步到数据库，生成一张表
python中的一个对象，保存到数据库中，形成一条记录
基本增删改查
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 查--->最多的，最复杂的
原生sql：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()
一对多关系
关系建立
1、导入一些模块
import datetime from sqlalchemy import create_engine from sqlalchemy.orm import declarative_base, \ relationship # 新的变成它，老的会有 from sqlalchemy.ext.declarative import declarative_base Warning from 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.caption class Person(Base): __tablename__ = 'person' id = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) # hobby指的是tablename而不是类名，uselist=False hobby_id = Column(Integer, ForeignKey("hobby.id")) # 关联字段写在多的一方 # 跟数据库无关，不会新增字段，只用于快速链表操作 # 类名，backref用于反向查询 hobby = relationship('Hobby', backref='pers') # 等同于django中的relate_name，反向查询替换表名小写的 def __str__(self): return self.name def __repr__(self): return self.name
类的 __repr__() 方法定义了实例化对象的输出信息，重写该方法，可以输出我们想要的信息，对类的实例化对象有更好的了解。
3、创建 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 # 多久之后对线程池中的线程进行一次连接的回收（重置） )
4、通过 engine 把上面的表模型，创建到数据库中
Base.metadata.create_all(engine)
5、 删除被Base管理的所有表
Base.metadata.drop_all(engine)
增加和基于对象的跨表查询
一对多关系新增和查询
新增，笨办法
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)
一对一
一对一关系，就是一对多，只不过多的一方只要一条
多对多关系
1、导入一些模块
import datetime from sqlalchemy import create_engine from sqlalchemy.orm import declarative_base, \ relationship # 新的变成它，老的会有 from sqlalchemy.ext.declarative import declarative_base Warning from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index # django orm，所有类，都要继承一个父类，sqlalchemy中这个父类就是Base，有点特殊，正常导入一个类，继承即可，但是它不是
2、得到所有表模型需要继承基类
Base = declarative_base()
3、多对多
models.py
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、增加和基于对象跨表查
多对多关系新增
笨办法，新增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) # 基于对象的跨表的正向查询
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()
g对象
flask 中的g对象，上下文对象， global 缩写，由于 global 是关键字---》于是它命名成了g
django中，有个东西，可以完成它的效果 就是 request，但是使用request会造成数据污染
flask中也可以用 request，但是也会造成数据污染，所以咱们使用g对象
g和session的区别
from flask import Flask,g,reques
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()
基本增改查
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") #修改数字要加上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()
常用查询
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()