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)