tjdify/api/services/external_application_service.py

import uuid
from datetime import datetime
from typing import Optional

from flask_login import current_user
from werkzeug.exceptions import NotFound

from models import db
from models.external_application import ExternalApplication
from services.errors.external_application import ExternalApplicationNameDuplicateError


class ExternalApplicationService:
    @staticmethod
    def get_external_applications(page, per_page, search=None, type=None, url=None, method=None):
        query = ExternalApplication.query.order_by(ExternalApplication.created_at.desc())

        if search:
            query = query.filter(ExternalApplication.name.ilike(f"%{search}%"))

        if type:
            query = query.filter(ExternalApplication.type == type)

        if url:
            query = query.filter(ExternalApplication.url == url)

        if method:
            query = query.filter(ExternalApplication.method == method)

        external_applications = query.paginate(page=page, per_page=per_page, max_per_page=100, error_out=False)
        return external_applications.items, external_applications.total

    @staticmethod
    def get_external_application(external_application_id: str) -> Optional[ExternalApplication]:
        external_application = ExternalApplication.query.filter(
            ExternalApplication.id == external_application_id
        ).first()
        return external_application

    @staticmethod
    def get_external_application_by_name(name: str) -> Optional[ExternalApplication]:
        external_application = ExternalApplication.query.filter_by(name=name).first()
        return external_application

    @staticmethod
    def save_external_application(args: dict) -> ExternalApplication:
        name = args["name"]
        external_application = ExternalApplicationService.get_external_application_by_name(name)
        if external_application:
            raise ExternalApplicationNameDuplicateError(f"ExternalApplication with name {name} already exists.")
        external_application = ExternalApplication(
            id=str(uuid.uuid4()),
            name=name,
            type=args["type"],
            url=args["url"],
            method=args["method"],
            created_by=current_user.id,
            updated_by=current_user.id,
        )
        db.session.add(external_application)
        db.session.commit()
        return external_application

    @staticmethod
    def update_external_application(args: dict, external_application_id: str) -> ExternalApplication:
        external_application = ExternalApplicationService.get_external_application(external_application_id)
        if not external_application:
            raise NotFound("ExternalApplication not found")
        external_application.name = args["name"]
        external_application.type = args["type"]
        external_application.url = args["url"]
        external_application.method = args["method"]
        external_application.updated_by = current_user.id
        external_application.updated_at = datetime.now()
        db.session.add(external_application)
        db.session.commit()
        return external_application

    @staticmethod
    def delete_external_application(external_application_id):
        external_application = ExternalApplicationService.get_external_application(external_application_id)
        if not external_application:
            raise NotFound("ExternalApplication not found")
        db.session.delete(external_application)
        db.session.commit()
        return external_application