rag-gateway/app/service/common/app_register.py

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):

        logger.error("更新数据: app register---------------------------")
        try:
            self.db.query(AppRegisterModel).filter(AppRegisterModel.id==app_id).update({"status":status, "updated_at": current_time()})
            self.db.commit()
        except Exception as e:
            logger.error(e)
            self.db.rollback()
            raise Exception("更新失败！")

    def insert_app(self, app_id: str, status: int):
        logger.error("新增数据: app register---------------------------")
        new_session = AppRegisterModel(
            id=app_id,
            status=status
        )
        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):

        logger.error("更新或者添加数据: app register---------------------------")
        token_boj = self.get_app_by_id(app_id)
        if token_boj:
            self.update_app(app_id, status)
        else:
            self.insert_app(app_id, status)


    def get_apps(self):
        app_list = self.db.query(AppRegisterModel).filter_by(status=1).all()

        return [i.to_dict() for i in app_list]