1. SQLAlchemy的使用

文中讲解时的表结构

# app/models/_base.py

from datetime import datetime, timedelta
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Column, TIMESTAMP, BigInteger, func
from app.common.cache import FlaskRedis

db = SQLAlchemy()
db_redis = FlaskRedis()


class BaseColumns:
    id = Column(BigInteger(), primary_key=True, nullable=False,
                autoincrement=True, unique=True)
    create_time = Column(TIMESTAMP, server_default=func.current_timestamp(), nullable=False)
    update_time = Column(TIMESTAMP, server_default=func.current_timestamp(),
                         onupdate=func.current_timestamp(), index=True)

def format_time():
    return datetime.now().strftime('%Y-%m-%d')

def y_format_time():
    return (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
# app/models/learn.py
from sqlalchemy import Column, Integer, Float, String, JSON, TEXT
from ._base import db, BaseColumns, format_time, y_format_time


class Learn(db.Model, BaseColumns):
    __tablename__ = "learn"    
    ip = Column(String(100), server_default='', index=True, unique=True, nullable=True)
    ip2num = Column(BigInteger, unique=True, index=True, nullable=True)

    whitewash_deadline = Column(Integer, server_default="30", index=True)  # 从黑IP转化灰色IP时间,初始为30天
    identity_color = Column(String(10), server_default='Black', index=True)  # 身份,默认为黑
    switch_count = Column(Integer, server_default="0")  # 历史上曾经洗白次数
    whitewash_time = Column(TIMESTAMP)  # 最近一次洗白的时间

    region = Column(String(100), server_default="")  # 区域
    country = Column(String(100), server_default="")  # 国家
    province = Column(String(100), server_default="")  # 省会
    city = Column(String(100), server_default="")  # 都市
    county = Column(String(100), server_default="")  # 乡县
    isp = Column(String(100), server_default="")  # 运营商

1.1. 语句篇

1.1.1. 数量统计

from sqlalchemy import func
get_num = db.session(func.count(Learn.id)).first()

1.1.2. 分组数量统计

from sqlalchemy import func
get_num = db.session(func.count(Learn.id)).group_by(Learn.identity_color == "Black").first()

1.1.3. 分组数量统计并排序

province_top = db.session.query(Learn.province, func.count(Learn.province)).filter(Learn.country == "中国",Learn.identity_color == "Black").group_by(Learn.province).order_by(func.count(Learn.province).desc()).all()

1.1.4. 分页用法

limit限制表查询数量;offset设置查询偏移起始值;slice(起始值, 结束值)切片操作。

data = db.session.query(Learn).limit(100).all()   # 查询前100条
data2 = db.session.query(Learn).limit(100).offset(200).all()  # 从第201起算,查询100条
data = db.session.query(Learn).slice(100, 200).all()   # 查询第101至第200条

1.1.5. 匹配一个列表 in_

need_query_ip_list = ["1.1.1.1"]
query_data = db.session.query(Learn.id).filter(Learn.ip.in_(need_query_ip_list)).all()

1.1.6. 获取固定时间内的数据

from datetime import datetime, timedelta
today = datetime.today()
yestorday = today + timedelta(-1)
# 统计昨天的所有数据量
info = db.session.query(func.count(IP.id)).filter(db.cast(IP.update_time, db.DATE) == db.cast(yestorday, db.DATE)).first()
# 统计11天前的所有数据量
info2 = db.session.query(func.count(IP.id)).filter(db.cast(yestorday, db.DATE)-db.cast(IP.update_time, db.DATE) > 10).first()

1.1.7. 批量更新或删除可能遇到的问题

Could not evaluate current criteria in Python: "xxxxxxxxxx". Specify 'fetch' or False for the synchronize_session parameter.

目前遇到过这种情况的场景有:

  1. in_delete结合使用
  2. castupdate 结合使用

[info] 解决方法 在delete或update方法中加参数: synchronize_session=False

范例:

db.session.query(Learn).filter(Learn.identity_color == "Black"). \
        filter(Learn.whitewash_deadline <
               (db.cast(today, db.DATE) - db.cast(Learn.update_time, db.DATE))).\
        update({"identity_color": "Gray"}, synchronize_session=False)
db.session.commit()

1.2. 优化篇

1.2.1. 查询优化

  1. 建立索引

    # index=True就是建立索引
    ip2num = Column(BigInteger, unique=True, index=True, nullable=True)
    
  2. 尽量通过SQLAlchemy来使用SQL支持的原生语法
      例如当你需要查询一个列表中的元素是否都在数据库中时,最快的方法是:

    db.session.query(Learn).filter(Learn.ip.in_(["1.1.1.1"])).all()
    
  3. 逻辑优化
      数据库中,对字符型数据的查找要远高于数字型的查找,所以在本例子中,表中看似冗余的ip2num字段其本质是对搜索的一种优化。这点在数据量较小的时候可能无关紧要,但如果可预见的未来数据超过百万级,那么优势将会逐渐凸显。

results matching ""

    No results matching ""

    results matching ""

      No results matching ""