SQLAlchemy快速使用

介绍

# 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-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
    
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
    
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：

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:[email protected]: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中的一个对象，保存到数据库中，形成一条记录

基本增删查改

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Users

# 1 增 ，删 ，查 ，该操作

# 1 生成 engine
engine = create_engine(
    "mysql+pymysql://root:[email protected]: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='[email protected]')
# 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':'[email protected]'})
# session.commit()
# print(res)


user = session.query(Users).filter_by(name='lqz').first()
user.email='[email protected]'
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


# 4 创建 engine,需要手动创建数据库
engine = create_engine(
    "mysql+pymysql://root:[email protected]: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)

增加和基于对象的跨表查询

####  一对多关系新增和查询
###  新增，笨办法
# 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()




####  多对多
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:[email protected]: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)

增加和基于对象跨表查

### 多对多关系新增
# 笨办法，新增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)  # 基于对象的跨表的正向查询

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:[email protected]: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='[email protected]')
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,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()

基本增查改

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:[email protected]: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:'[email protected]'})
# 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()

常用查询

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:[email protected]: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()