| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556 |
- 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]
|
