Skip to content

BPMN Parser

Implement a Python API to a BPMN. With this, is possible to get and list some attributes of BPMN, such as: service tasks, user tasks, sequence flows, etc.

Parameters:

Name Type Description Default
file_path Union[Path, str]

path of bpmn file.

 required
Source code in bpmn_parser/parser.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class BPMNParser:
    """Implement a Python API to a BPMN. With this, is possible to get and list some
    attributes of BPMN, such as: service tasks, user tasks, sequence flows, etc.


    Args:
        file_path (Union[Path, str]): path of bpmn file.
    """

    def __init__(self, file_path: Union[Path, str]):
        if isinstance(file_path, str):
            file_path = Path(file_path)

        if file_path.suffix != '.bpmn':
            raise NotBPMNFile()

        self.file_path = file_path
        self.xml_parser = XMLParser(remove_blank_text=True)
        self.tree: _ElementTree = parse(file_path, self.xml_parser)
        self.root: _Element = self.tree.getroot()

        self._service_task: Optional[ServiceTask] = None
        self._user_task: Optional[UserTask] = None
        self._sequence_flow: Optional[SequenceFlow] = None
        self._intermediate_catch_event: Optional[IntermediateCatchEvent] = None
        self._exclusive_gateway: Optional[ExclusiveGateway] = None

    @property
    def service_task(self):
        """Parse all Service Tasks from this BPMN."""
        if self._service_task is not None:
            return self._service_task

        self._service_task = ServiceTask(self.root)
        return self._service_task

    @property
    def user_task(self):
        """Parse all User Task from this BPMN."""
        if self._user_task is not None:
            return self._user_task

        self._user_task = UserTask(self.root)
        return self._user_task

    @property
    def sequence_flow(self):
        """Parse all Sequence Flow from this BPMN."""
        if self._sequence_flow is not None:
            return self._sequence_flow

        self._sequence_flow = SequenceFlow(self.root)
        return self._sequence_flow

    @property
    def intermediate_catch_event(self):
        """Parse all Intermediate Catch Event from this BPMN."""
        if self._intermediate_catch_event is not None:
            return self._intermediate_catch_event

        self._intermediate_catch_event = IntermediateCatchEvent(self.root)
        return self._intermediate_catch_event

    @property
    def exclusive_gateway(self):
        """Parse all Exclusive Gateway from this BPMN."""
        if self._exclusive_gateway is not None:
            return self._exclusive_gateway

        self._exclusive_gateway = ExclusiveGateway(self.root)
        return self._exclusive_gateway

    @property
    def refresh(self):
        """Refresh the BPMN if the file get some modify."""
        self.tree = parse(self.file_path, self.xml_parser)
        self.root = self.tree.getroot()

        del self._service_task
        self._service_task = None

        del self._user_task
        self._user_task = None

        del self._sequence_flow
        self._sequence_flow = None

        del self._intermediate_catch_event
        self._intermediate_catch_event = None

        del self._exclusive_gateway
        self._exclusive_gateway = None

    def __repr__(self) -> str:
        return (
            f'{self.__class__.__name__}('
            f'service_task={len(self.service_task)}, '
            f'user_task={len(self.user_task)}, '
            f'sequence_flow={len(self.sequence_flow)}, '
            f'intermediate_catch_event={len(self.intermediate_catch_event)}, '
            f'exclusive_gateway={len(self.exclusive_gateway)}, '
            f'file_path={self.root.base}'
            ')'
        )

exclusive_gateway property

Parse all Exclusive Gateway from this BPMN.

intermediate_catch_event property

Parse all Intermediate Catch Event from this BPMN.

refresh property

Refresh the BPMN if the file get some modify.

sequence_flow property

Parse all Sequence Flow from this BPMN.

service_task property

Parse all Service Tasks from this BPMN.

user_task property

Parse all User Task from this BPMN.

1
bpmn_parser
Output
BPMNParser(
    service_task=4,
    user_task=2,
    sequence_flow=24,
    intermediate_catch_event=2,
    exclusive_gateway=10,
    file_path=resources/flow.bpmn
)