from decorator.singleton import singleton
|
from db.db_conn import db_conn
|
from db.models.tables import RequestTaskSetting, LoginInfo
|
|
|
@singleton
|
class ConfigRepository(object):
|
"""配置参数相关数据库操作"""
|
|
def __init__(self) -> None:
|
pass
|
|
def read_request_config(self) -> list:
|
"""获取定时任务相关配置参数"""
|
|
return db_conn.query_sql(lambda session: session
|
.query(RequestTaskSetting)
|
.filter_by(region="金山区")
|
.order_by(RequestTaskSetting.update_time.desc())
|
.first(), new_session=True)
|
|
def get_login_info(self) -> LoginInfo:
|
"""获取登录参数"""
|
|
return db_conn.query_sql(lambda session: session
|
.query(LoginInfo)
|
.filter_by(id=1)
|
.one(), new_session=True)
|
|
def update_login_info(self, obj: LoginInfo):
|
"""更新登录参数"""
|
|
if isinstance(obj, LoginInfo):
|
def f(session):
|
record = session.query(LoginInfo).filter_by(id=obj.id).one()
|
record.cookie = obj.cookie
|
record.cookie_timestamp = obj.cookie_timestamp
|
session.commit()
|
return db_conn.query_sql(f)
|