flask---》更多查询方式/连表查询/原生sql(django-orm如何执行原生sql)/flask-sqlalchemy_flask sqlalchemy execute select原生sql

更多查询方式

#1  查询： filer:写条件     filter_by：等于的值
#  查询所有  是list对象
res = session.query(User).all()  # 是个普通列表
print(type(res))
print(len(res))
 
# 2 只查询某几个字段
# select name as xx,email from user;
res = session.query(User.name.label('xx'), User.email)
# print(res)  # 打出原生sql
# # print(res.all())
# for item in res.all():
#     print(item[0])
 
 
# 3 filter传的是表达式，filter_by传的是参数
res = session.query(User).filter(User.name == "lqz").all()
res = session.query(User).filter(User.name != "lqz").all()
res = session.query(User).filter(User.name != "lqz", User.email == '3@qq.com').all() 
 
#4 django 中使用 Q (与  或   非)   如果是， 就是and条件
res = session.query(User).filter_by(name='lqz099').all()
res = session.query(User).filter_by(name='lqz099',email='47@qq.com').all()
 
 
# 5 取一个 all了后是list，list 没有first方法
res = session.query(User).first()
 
 
 
# 6查询所有，使用占位符(了解)  :value     :name
# select * from user where id <20 or name=lqz
res = session.query(User).filter(text("id<:value or name=:name")).params(value=20, name='lqz').all()
 
 
# 7 自定义查询（了解）
# from_statement 写纯原生sql
 
res=session.query(User).from_statement(text("SELECT * FROM users where email=:email")).params(email='3@qq.com').all()
print(type(res[0]))  # 是book的对象，但是查的是User表   不要这样写
print(res[0].name)  #
 
# 8 高级查询
# 　条件
# 表达式，and条件连接
res = session.query(User).filter(User.id > 1, User.name == 'lqz099').all() # and条件
 
# between
res = session.query(User).filter(User.id.between(1, 9), User.name == 'lqz099').all()
res = session.query(User).filter(User.id.between(1, 9)).all()
 
# in
res = session.query(User).filter(User.id.in_([1,3,4])).all()
res = session.query(User).filter(User.email.in_(['3@qq.com','r@qq.com'])).all()
 
# ~非，除。。外
res = session.query(User).filter(~User.id.in_([1,3,4])).all()
print(res)
 
# 二次筛选
res = session.query(User).filter(~User.id.in_(session.query(User.id).filter_by(name='lqz'))).all()
print(res)
 
 
# and or条件
from sqlalchemy import and_, or_
 
# or_包裹的都是or条件，and_包裹的都是and条件
res = session.query(User).filter(and_(User.id >= 3, User.name == 'lqz099')).all()  #  and条件
res = session.query(User).filter(User.id < 3, User.name == 'lqz099').all()  #  等同于上面
res = session.query(User).filter(or_(User.id < 2, User.name == 'eric')).all()
res = session.query(User).filter(or_(User.id < 2,and_(User.name == 'lqz099', User.id > 3),User.extra != ""))
 
 
# 通配符，以e开头，不以e开头
res = session.query(User).filter(User.email.like('%@%')).all()
# select user.id from user where  user.name not like e%;
res = session.query(User.id).filter(~User.name.like('e%'))
res = session.query(User).filter(~User.name.like('e%')).all()
 
# 分页
# 一页2条，查第5页
res = session.query(User)[2*5:2*5+2]
 
# 排序，根据name降序排列（从大到小）
res = session.query(User).order_by(User.email.desc()).all()
res = session.query(Book).order_by(Book.price.desc()).all()
res = session.query(Book).order_by(Book.price.asc()).all()
# 第一个条件重复后，再按第二个条件升序排
res = session.query(User).order_by(User.name.desc(), User.id.asc())
 
 
 
# 分组查询  5个聚合函数
from sqlalchemy.sql import func
# 分组后，只能拿分组字段和聚合函数字典，如果拿别的，是严格模式，会报错
res = session.query(User).group_by(User.extra)  # 如果是严格模式，就报错
# 分组之后取最大id，id之和，最小id  和分组的字段
from sqlalchemy.sql import func
res = session.query(
    User.name,
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id),
    func.avg(User.id)).group_by(User.name).all()
for item in res:
    print(item)
 
# 分组后having
# select name,max(id),sum(id),min(id) from  user group by  user.name   having id_max>2;
 
 
from sqlalchemy.sql import func
res = session.query(
    User.name,
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.name).having(func.max(User.id) > 2).all()
 
print(res)
 
 

连表查询

### 关联关系，基于连表的跨表查询
from models1 import Person,Hobby
# 链表操作
select * from person,hobby where person.hobby_id=hobby.id;
res = 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;
# res = session.query(Person).join(Hobby).all()
 
#isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
# select * from Person left join Hobby on Person.hobby_id=Hobby.id;
# res = session.query(Person).join(Hobby, isouter=True).all()
# 没有right join，通过这个实现
# res = session.query(Hobby).join(Person, isouter=True).all()
 
# # 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
# select * from Person left join Hobby on Person.id=Hobby.id;
# res = session.query(Person).join(Hobby, Person.hobby_id == Hobby.id, isouter=True) #  sql本身有问题，只是给你讲， 自己指定链接字段
# 右链接
# print(res)
 
 
 
# 多对多关系连表
# 多对多关系，基于链表的跨表查
# 多表链接
#方式一：直接连
#select * FROM boy, girl, boy2girl WHERE boy.id = boy2girl.boy_id AND girl.id = boy2girl.girl_id
# res = session.query(Boy, Girl,Boy2Girl).filter(Boy.id == Boy2Girl.boy_id,Girl.id == Boy2Girl.girl_id)
 
# 方式二：join连
# SELECT* FROM boy INNER JOIN boy2girl ON boy.id = boy2girl.boy_id INNER JOIN girl ON girl.id = boy2girl.girl_id WHERE boy.id >= %(id_1)s
res = session.query(Boy).join(Boy2Girl).join(Girl).filter(Boy.id>=2)
print(res)

原生sql(django-orm如何执行原生sql)

sqlalchemy执行原生sql

# 有的复杂sql 用orm写不出来---》用原生sql查询
 
# 原生sql查询，查出的结果是对象
# 原生sql查询，查询结果列表套元组
 
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/db001", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
#### 执行原生sql方式一：
# 查询方式一：
# cursor = session.execute('select * from users')
# result = cursor.fetchall()
# print(result) #列表套元组
 
# 添加
# cursor = session.execute('insert into users(name,email) values(:name,:email)',
#                          params={"name": 'lqz', 'email': '3333@qq.com'})
# session.commit()
# print(cursor.lastrowid)
 
 
###执行原生sql方式二（以后都用session操作---》socpe_session线程安全）一般不用
# conn = engine.raw_connection()
# cursor = conn.cursor()
# cursor.execute(
#     "select * from app01_book"
# )
# result = cursor.fetchall()
 
 
# 执行原生sql方式三：
# res = session.query(User).from_statement(text("SELECT * FROM boy where name=:name")).params(name='lqz').all()
 
 
session.close()

django执行原生sql

# 执行完的结果映射到对象中---》上面讲的  方式三：
from model import Book
books_obj_list = Book.objects.raw('select distinct id, book_name from test_book')
for book_obj in books_obj_list:
	print(book_obj.id, book_obj.book_name)
 
 
    
# 纯原生sql 
from django.db import connection
cur=connection.cursor() 
cur.execute('select distinct id, book_name from test_book')
print(cur.fetch_all())
cur.close()
 
 
with connection.cursor() as cur:
    cur.execute('select distinct id, book_name from test_book')

flask-sqlalchemy

# sqlalchemy 集成到flask中
 
 
# 第三方: flask-sqlalchemy 封装了用起来，更简洁
#  使用flask-sqlalchemy集成
	1 导入 from flask_sqlalchemy import SQLAlchemy
    2 实例化得到对象
    	db = SQLAlchemy()
    3  将db注册到app中
    	db.init_app(app)
    4 视图函数中使用session
    	全局的db.session  # 线程安全的
    5 models.py 中继承Model
    	db.Model
    6 写字段 
    	username = db.Column(db.String(80), unique=True, nullable=False)
    7 配置文件中加入
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root@127.0.0.1:3306/ddd?charset=utf8"
    SQLALCHEMY_POOL_SIZE = 5
    SQLALCHEMY_POOL_TIMEOUT = 30
    SQLALCHEMY_POOL_RECYCLE = -1
    # 追踪对象的修改并且发送信号
    SQLALCHEMY_TRACK_MODIFICATIONS = False