表模型建立
from sqlalchemy import Boolean,Column,Integer,String,DateTime,VARCHAR,CHAR
from sqlalchemy.ext.declaratice import declaratice_base
from sqlalchemy.sql import func
Base = declaratice_base()
class TableModel(Base):
tablename = '表名'
id = Column(Integer,primary_key=True)
value = Column(VARCHAR(255),default='')
created_at = Column(DateTime,server_default=func.now())
常用引用
from sqlalchemy.orm import Session
from sqlalchemy.sql import or_, and, func, desc
from sqlalchemy.dialects.mysql import insert
增
# 普通添加
session.add(TableModel(**value_dict))
# 批量添加
obj_list = [TableModel(**value_dict1),...]
session.bulk_save_objects(obj_list)
# 主键冲突则覆盖
# 单条记录
insert(TableModel).values(
**value_dict).on_duplicate_key_update(
updated_at=now()
)
# 多条记录
value_list = [value_dict1,value_dict2]
insert_stmt = insert(TableModel).values(value_list)
on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
key=insert_stmt.inserted.key
)
删
session.query(TableModel).filter(
key=value
).delete()
改
# 统一更改
session.query(TableModel).filter(
key=value
).update({
TableModel.updated_at:now()
})
# map映射更改,必须含主键
value_list = [value_dict1,value_dict2]
session.bulk_update_mappings(TableModel,value_list)
查
session.query(字段).join(需要连接的表,条件).filter(
赛选条件).group_by(聚合key).order_by(排序字段).offset(偏移).limit(数量).all()
# 以上即是查询优先级的各种连接
# func 常用函数
func.count() 记数
func.min() 最小值
func.max() 最大值
func.IF() if函数
func.greatest(字段1，字段2) 比较两个字段取出最大值
func.now() 数据库当前时间
func.hour(DateTime) 获取时
func.dayofweek(DateTime) 获取星期几
func.length(字符串) 返回字节长度
func.replace(原字符串，需要替换的值，替换成什么值)
# 可用比较出key字段中与value值的相似度，用于排优先级
# 先用条件过滤出 key.like(f'%{value}%')
func.length(func.replace(TableModel.key,value,''))
