shanxia/bx/bx-backend/app/services/workflow.py

from sqlalchemy.orm import Session

from app.models.workflow import WorkflowDef, WorkflowNode
from app.services.base import CrudService


class WorkflowService:

    @staticmethod
    def get_by_id(db: Session, id: int) -> WorkflowDef | None:
        return CrudService.get_by_id(WorkflowDef, db, id)

    @staticmethod
    def get_paged(db: Session, page: int, page_size: int, filters: list | None = None):
        return CrudService.get_paged(WorkflowDef, db, page, page_size, filters)

    @staticmethod
    def create(db: Session, obj_in: dict) -> WorkflowDef:
        return CrudService.create(WorkflowDef, db, obj_in)

    @staticmethod
    def update(db: Session, id: int, obj_in: dict) -> WorkflowDef | None:
        return CrudService.update(WorkflowDef, db, id, obj_in)

    @staticmethod
    def soft_delete(db: Session, id: int) -> bool:
        return CrudService.soft_delete(WorkflowDef, db, id)

    @staticmethod
    def get_by_biz_type(db: Session, biz_type: str) -> WorkflowDef | None:
        return (
            db.query(WorkflowDef)
            .filter(WorkflowDef.biz_type == biz_type, WorkflowDef.del_flag == 0)
            .first()
        )

    @staticmethod
    def save_nodes(db: Session, wf_id: int, nodes: list[dict]) -> list[WorkflowNode]:
        db.query(WorkflowNode).filter(WorkflowNode.wf_id == wf_id).delete()
        created: list[WorkflowNode] = []
        for node_data in nodes:
            node_data["wf_id"] = wf_id
            node = WorkflowNode(**node_data)
            db.add(node)
            created.append(node)
        db.flush()
        return created