from datetime import datetime
|
|
from app.models.public_api_model import AppRegisterModel
|
from Log import logger
|
# from app.models import current_time
|
from sqlalchemy.orm import Session
|
from typing import Type
|
|
|
class AppRegisterDao:
|
def __init__(self, db: Session):
|
self.db = db
|
|
def get_app_by_id(self, api_id: str) -> Type[AppRegisterModel] | None:
|
session = self.db.query(AppRegisterModel).filter_by(id=api_id).first()
|
|
return session
|
|
def update_app(self, app_id: str, status: int, name):
|
|
logger.error("更新数据: app register---------------------------")
|
try:
|
self.db.query(AppRegisterModel).filter(AppRegisterModel.id==app_id).update({"status":status, "updated_at": datetime.now(), "name": name})
|
self.db.commit()
|
except Exception as e:
|
logger.error(e)
|
self.db.rollback()
|
raise Exception("更新失败！")
|
|
def insert_app(self, app_id: str, status: int, name: str):
|
logger.error("新增数据: app register---------------------------")
|
new_session = AppRegisterModel(
|
id=app_id,
|
status=status,
|
name=name
|
)
|
self.db.add(new_session)
|
self.db.commit()
|
self.db.refresh(new_session)
|
return new_session
|
|
|
def update_and_insert_app(self, app_id: str, status: int, name=""):
|
|
logger.error("更新或者添加数据: app register---------------------------")
|
token_boj = self.get_app_by_id(app_id)
|
if token_boj:
|
self.update_app(app_id, status, name)
|
else:
|
self.insert_app(app_id, status, name)
|
|
|
def get_apps(self):
|
app_list = self.db.query(AppRegisterModel).filter_by(status=1).all()
|
|
return [i.to_dict() for i in app_list]
|
