Lottery/orm/db.py

from orm_db import History, BlueForecastFiveAll, BlueForecastThreeAll, RedForecastAll
from sqlalchemy import engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select, desc, asc

# 初始化数据库实例
DB_URI = 'mysql+pymysql://root:Sxzgx1209@home.rogersun.cn:3306/lottery'
db_engine = engine.create_engine(DB_URI, encoding='utf8', echo=True)
db_session = sessionmaker(bind=db_engine)()


def db_get_last_data():
    try:
        get_last_data_stmt = select(History).order_by(desc('id')).limit(1)
        last_data_result = db_session.execute(get_last_data_stmt).scalar()
        return history_scalar(last_data_result)
    except Exception as e:
        print(e)
        return False


def db_get_all_data():
    try:
        get_all_data_stmt = select(History).order_by(asc(History.id))
        all_data_result = db_session.execute(get_all_data_stmt).scalars()
        return history_scalars(all_data_result)
    except Exception as e:
        print(e)
        return False


def db_get_data_by_date_id(_date_id):
    try:
        get_by_open_date_stmt = select(History).where(History.dateId == _date_id)
        open_date_result = db_session.execute(get_by_open_date_stmt).scalar()
        return history_scalar(open_date_result)
    except Exception as e:
        print(e)
        return False


def history_scalar(_scalar):
    _data = dict()
    _data['id'] = _scalar.id
    _data['dateId'] = _scalar.dateId
    _data['openDate'] = _scalar.openDate
    _data['red'] = _scalar.red
    _data['blue'] = _scalar.blue
    return _data


def history_scalars(_scalars):
    _list = []
    for row in _scalars:
        _data = dict()
        _data['id'] = row.id
        _data['dateId'] = row.dateId
        _data['openDate'] = row.openDate
        _data['red'] = row.red
        _data['blue'] = row.blue
        _list.append(_data)
    return _list