from decorator.singleton import singleton
|
from db.db_conn import db_conn
|
from db.models.tables import DustSiteInfo
|
|
|
@singleton
|
class SiteInfoRepository(object):
|
"""站点基本信息相关数据库操作"""
|
|
def __init__(self) -> None:
|
pass
|
|
def save_site_info(self, info_list) -> bool:
|
"""保存站点基本信息"""
|
def f(session):
|
# 获取已有站点信息
|
site_list = session.query(DustSiteInfo).all()
|
for i in info_list:
|
# 更新已有站点
|
find = False
|
for s in site_list:
|
if i["ID"] == s.id:
|
s.update(i)
|
find = True
|
break
|
# 插入新站点
|
if not find:
|
new_site_info = DustSiteInfo(i)
|
session.add(new_site_info)
|
session.commit()
|
return db_conn.query_sql(f, new_session=True)
|
|
def query_by_mncode(self, mn_code) -> DustSiteInfo:
|
"""根据mncode查询站点基本信息"""
|
return db_conn.query_sql(lambda session: session.query(DustSiteInfo).filter_by(mn_code=mn_code).first())
|
