1.数据表定义和创建表

from sqlalchemy.orm import sessionmaker
from config import setting
from sqlalchemy import Column, String, Integer, Float, DECIMAL, Boolean, DateTime, create_engine,BigInteger
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import funcengine = create_engine("mysql+pymysql://{}:{}@{}/{}".format(setting.User, setting.Passwd, setting.Host,setting.DB),encoding='utf8',echo=False, pool_recycle = 7200)
Base = declarative_base()
session = sessionmaker(bind=engine)()a = [{'name': '1', 'item_id': '284440448086', 'year': '2007', 'make': 'Honda', 'model': 'Accord', 'trim': 'Special Edition Sedan 4-Door', 'engine': '2.4L 2354CC 144Cu. In. l4 GAS DOHC Naturally Aspirated', 'notes': 'All Trim Levels'},
{'name': '21', 'item_id': '284440448086', 'year': '2007', 'make': 'Honda', 'model': 'Accord', 'trim': 'Special Edition Sedan 4-Door', 'engine': '3.0L 2997CC V6 GAS SOHC Naturally Aspirated', 'notes': 'All Trim Levels'}]class YtTest(Base):__tablename__ = 'yt_test'id = Column(Integer, primary_key=True)name = Column(String(30), comment='名字')item_id = Column(String(50), comment='item_id')year = Column(String(50), comment='year')make = Column(String(100), comment='make')model = Column(String(100), comment='model')trim = Column(String(255), comment='trim')engine = Column(String(255), comment='engine')notes = Column(String(255), comment='notes')create_time = Column(DateTime(timezone=True), server_default=func.now())# update_time = Column(DateTime, comment='更新时间')def __init__(self, **items):for key in items:if hasattr(self, key):setattr(self, key, items[key])addre = [{'name':'21','address':'盘龙区','city':'四圣市','e_num':'10111','t_id':'12'},{'name':'21','address':'盘龙区','city':'四圣市','e_num':'10111','t_id':'12'},
]
class Address(Base):__tablename__ = 'yt_address'id = Column(Integer, primary_key=True)name = Column(String(30), comment='name')address = Column(String(30), comment='Address')city = Column(String(50), comment='city')e_num = Column(String(50), comment='e_num')t_id = Column(String(50), comment='t_id')def __init__(self, **items):for key in items:if hasattr(self, key):setattr(self, key, items[key])# 创建表
# Base.metadata.create_all(engine)

2.新增数据（add_all()/add()）

# # 一 新增数据  
# # 新增数据 多条数据
# save_data = []
# for i in a:
#     report_to_sql = eval('YtTest')(**i)
#     save_data.append(report_to_sql)
# session.add_all(save_data)
# session.commit()# save_data = []
# for i in addre:
#     report_to_sql = eval('Address')(**i)
#     save_data.append(report_to_sql)
# session.add_all(save_data)
# session.commit()# # #新增数据 单条数据
# role2 = YtTest(name="李明")
# session.add(role2)
# session.commit()
# print(role2.id) #返回id

3.删除数据

# # 二, 删除 session.delete(i)/session.query(YtTest).delete()
# # 循环删除
user = session.query(YtTest).filter(YtTest.name == "22").all()
for i in user:session.delete(i)
session.commit()
session.close()# # 全量删除
session.query(YtTest).filter(YtTest.name=='21r').delete()
session.commit()

4.更新数据

# 三. 更新操作  session.query(YtTest).update()/查询-commit()
# update 更新
session.query(YtTest).filter(YtTest.name == 'stuff').update({YtTest.name: '李明'})
# 3：提交与关闭
session.commit()# # 更新二
test = session.query(YtTest).filter(YtTest.name == '李明').all()
for i in test:i.name = '李明1'
# 3：提交与关闭
session.commit()
test = session.query(YtTest).filter(YtTest.name == '李明').first()
test.name = '李明1'
# 3：提交与关闭
session.commit()

5.查询数据

# 四，查询
# filter/filter_by ：对应 WHERE ，fillter 可以进行比较运算(==, >, < ...)来对条件进行灵活的运用，不同的条件用逗号分割，
# fillter_by 只能指定参数传参来获取查询结果。
users = session.query(YtTest).filter(YtTest.name == "李明").all()
users = session.query(YtTest).filter_by(name = "李明").all()print([i.name for i in users])

filter_by()和 filter()的最主要区别：

filter_by()
       查询 user 表里面名字等于 Tom 的：
       db.session.query(User).filter_by(name='Tom').all()
       查询 user 表里面名字等于 Tom 并且年龄等于 18：
       db.session.query(User).filter_by(name='Tom', age=18).all()
       比如新的需求，查询 user 表里面名字等于 Tom 或者年龄等于 18 的用户，那么 filter_by() 就满足不了要求了

filter()
       查询 user 表里面名字等于 Tom 的：
       db.session.query(User).filter(User.name == 'Tom').all()
       查询 user 表里面名字等于 Tom 并且年龄等于 18：
       db.session.query(User).filter(User.name == 'Tom', User.age == 18).all()
       也可以这样：
       db.session.query(User).filter(User.name == 'Tom'）.filter(User.age == 18).all()
       如果想使用 and 拼接需要用以下方式：
       db.session.query(User).filter(and_(User.name == 'Tom', User.age == 18)).all()
       以下的方式 and 后面的 User.age == 18 不会生效：
       db.session.query(User).filter(User.name == 'Tom' and User.age == 18).all()
       查询 user 表里面名字等于 Tom 的或者年龄等于 18：
       db.session.query(User).filter(or_(User.name == 'Tom', User.age == 18)).all()
       查询 user 表里面名字等于 Tom 的并且年龄大于 18
       db.session.query(User).filter(User.name == 'Tom', User.age > 18).all()
       查询 name 中包含字母 a 的所有数据(模糊查询)
       db.session.query(User).filter(User.name.like('%{0}%'.format("a"))).all()

以上的例子都是查询中使用比较多的，使用方面看大家喜好，filter_by() 对组合查询等等支持的不是很好，但是语法相对 filter() 简洁一些； 另外 filter() 还有很多其他的查询，大家可以自己去多多探讨。。。

笔记：

现在有 mysql 的 user 数据库表，存储内容如下图，下面通过例子来说明两种查询方式的用法：

在一个用户的模型类中, 要查询id为5的一个用户;

User.query.filter(User.id==5).all()  #  使用filter 中必须指定那个 模型类.id==5; 

User.query.filter(User.id==5).first()  # all() , first() ,都是查询执行器 , 一个返回列表包含对象, 一个直接返回对象;

User.query.filter_by(id=5).first()  # filter_by默认就是使用id  进行查询 , 写法为id=5; 不可以使用==; 也不用模型类.id的方式; 它是一种更加精确的查询;

# users = User.query.filter_by(isdelete=False).all()        用等值符号

# users = User.query.filter(User.isdelete == False,

                   User.phone.startswith('150')).order_by(-User.rdatetime)  用关系符号 升序降序正负号
users = User.query.filter(or_(User.username==search, User.phone == search)).all()  # select * from user where username=search or phone=search
number = User.query.filter(or_(User.username==search,User.phone == search)).count()
users = User.query.filter(or_(User.username.like('%' + search + '%'), User.phone == search)).all()  #返回的是列表
number = User.query.filter(or_(User.username.like(search), User.phone == search)).count() 

查询一些方法

为了更好的理解 SQL 与 SQLalchemy 的写法区别，可以参照以下内容：

• query ：对应 SELECT xxx FROM xxx
• filter/filter_by ：对应 WHERE ，fillter 可以进行比较运算(==, >, < ...)来对条件进行灵活的运用，不同的条件用逗号分割，fillter_by 只能指定参数传参来获取查询结果。
• limit ：对应 limit()
• order by ：对应 order_by()

• group by ：对应 group_by(）

like查询

data_like = session.query(YtTest).filter(YtTest.name.like('%李%')).first()
data_like = session.query(YtTest).filter(YtTest.name.notlike('%李%')).first()

is查询

# is_ 相当于 ==
result = session.query(YtTest).filter(YtTest.item_id.is_(None)).all()
result = session.query(YtTest).filter(YtTest.item_id == 'None').all()
# isnot 相当于 !=
result = session.query(YtTest).filter(YtTest.item_id.isnot(None)).all()
result = session.query(YtTest).filter(YtTest.item_id != None).all()
print([i.name for i in result])

正则查询

data_regexp = session.query(YtTest).filter(YtTest.name.op("regexp")(r"^[\u4e00-\u9fa5]+")).all()
print([i.name for i in data_regexp])

统计数量 .count()

data_like_count = session.query(YtTest).filter(YtTest.name.like("李%")).count()

in查询

more_person = session.query(YtTest).filter(YtTest.name.in_(['李明', '21'])).all()

notin 查询

# ~代表取反,转换成sql就是关键字not
more_person = session.query(YtTest).filter(~YtTest.name.in_(['李明', '21'])).all()
# 或 notin_
more_person = session.query(YtTest).filter(~YtTest.name.notin_(['李明', '21'])).all()

and 查询

from sqlalchemy import and_
# and 作用
more_person = session.query(YtTest).filter_by(name = '李明' , year = "2006").all()
more_person = session.query(YtTest).filter(and_(YtTest.name == '李明', YtTest.item_id == "21")).all()more_person = session.query(YtTest).filter(YtTest.name == '李明' and YtTest.year == "2006").all()  #and后面的不起作用
more_person = session.query(YtTest).filter(YtTest.name == '李明' , YtTest.year == "2006").all()

OR 查询

from sqlalchemy import or_more_person = session.query(YtTest).filter(or_(YtTest.name == '李明' , YtTest.year == "2006")).all()

分组查询

std_group_by = session.query(YtTest).group_by(YtTest.year).all()
# 或是
from sqlalchemy.sql import func
res = session.query(YtTest.year,func.count(YtTest.year),).group_by(YtTest.year).all()# 遍历查看，已无ed用户记录
for person in std_group_by:print(person.name)

排序函数  limit 查询  偏移量查询  聚合函数

# 排序查询
std_order_by = session.query(YtTest).order_by(YtTest.year.desc()).all()
# limit 限制数量查询, limit里传入一个整型来约束查看的数量, 当limit里面的参数大于实例表中的数量时，会返回所有的查询结果
data_limit = session.query(YtTest).filter(YtTest.name.notlike("李%")).limit(1).all()
# offset 偏移量查询,offset中传入一个整型，从表中的该位置开始查询，offset可以和limit混用来进行限制
data_like = session.query(YtTest).filter(YtTest.name.like("李%")).offset(1).all()
result = session.query(YtTest).offset(1).limit(6).all()print([i.name for i in result])# 聚合函数
from sqlalchemy import func, extract
# count
result = session.query(YtTest.name, func.count(YtTest.id)).group_by(YtTest.name).all()
# sum
result = session.query(YtTest.name, func.sum(YtTest.id)).group_by(YtTest.name).all()
# max
result = session.query(YtTest.name, func.max(YtTest.id)).group_by(YtTest.name).all()
# min
result = session.query(YtTest.name, func.min(YtTest.id)).group_by(YtTest.name).all()
# having
result = session.query(YtTest.name, func.count(YtTest.id)).group_by(YtTest.name).having(func.count(YtTest.id) > 1).all()
print(result)

传参

# 传参
filter = (YtTest.name == '李明')
our_user = session.query(YtTest).filter(filter).first()
print(our_user.name)