Pular para conteúdo

execucoes

Proxy para operacoes de execucao de casos no SDK do PLDPro.

Execucoes

Operações para iniciar, consultar e interromper execuções de casos.

Source code in pldpro_sdk/core/proxies/execucoes.py 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class Execucoes:
    """
    Operações para iniciar, consultar e interromper execuções de casos.
    """

    def __init__(self, api_client: ApiClient) -> None:
        self._cases_api = CasesApi(api_client)
        self._auth_nodes_api = AuthNodesApi(api_client)

    @staticmethod
    def _as_uuid(value: Union[str, UUID]) -> UUID:
        return value if isinstance(value, UUID) else UUID(str(value))

    def start_execution(
        self,
        caso_id: Union[str, UUID],
        execution_start_request: ExecutionStartRequest | None = None,
    ):
        """
        Inicia a execução de um caso.

        :param caso_id: Identificador único do caso a ser executado.
        :type caso_id: UUID ou str
        :param execution_start_request: Configuração da execução. Quando não informada, usa a configuração padrão do SDK.
        :type execution_start_request: ExecutionStartRequest | None
        :return: Status da execução iniciada ou None caso a operação falhe.
        :rtype: ExecutionStatus | None
        """
        try:
            case_uuid = self._as_uuid(caso_id)
            if execution_start_request is None:
                execution_start_request = ExecutionStartRequest(
                    version="7.0.0",
                    tipoScript="COMPLETE",
                    listaUUIDAuthNodes=[],
                    optionsExecutaModelo=OptionsExecutaModelo(
                        localExecution="CLOUD",
                        projectId="",
                        reservationId="",
                        machineType="c2std60",
                    ),
                )
            return self._cases_api.start_caso_execution(
                uuid=case_uuid,
                execution_start_request=execution_start_request,
            )
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível iniciar a execução: %s",
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def get_execution_status(self, caso_id: str | UUID):
        """
        Consulta o status da execução de um caso.

        :param caso_id: Identificador único do caso.
        :type caso_id: UUID ou str
        :return: Status da execução ou None caso a consulta falhe.
        :rtype: ExecutionStatus | None
        """
        try:
            return self._cases_api.get_caso_execution_status(uuid=self._as_uuid(caso_id))
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível consultar o status da execução: %s",
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def get_execution_logs(self, caso_id: str | UUID) -> str | None:
        """
        Obtém o conteúdo do log da execução.

        :param caso_id: Identificador único do caso.
        :type caso_id: UUID ou str
        :return: Conteúdo textual do log ou None caso a consulta falhe.
        :rtype: str | None
        """
        return self._get_execution_text(caso_id, self._cases_api.get_caso_execution_logs_without_preload_content)

    def get_execution_console_logs(self, caso_id: Union[str, UUID]) -> str | None:
        """
        Obtém a saída do console da execução.

        :param caso_id: Identificador único do caso.
        :type caso_id: UUID ou str
        :return: Conteúdo textual do console ou None caso a consulta falhe.
        :rtype: str | None
        """
        return self._get_execution_text(caso_id, self._cases_api.get_caso_execution_console_logs_without_preload_content)

    def _get_execution_text(self, caso_id: Union[str, UUID], operation) -> str | None:
        try:
            response = operation(self._as_uuid(caso_id))
            raw = response.data if hasattr(response, "data") else response.read() if hasattr(response, "read") else response
            return raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else str(raw)
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível obter logs da execução: %s",
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def stop_execution(self, caso_id: Union[str, UUID]) -> str | None:
        """
        Interrompe a execução de um caso.

        :param caso_id: Identificador único do caso.
        :type caso_id: UUID ou str
        :return: Confirmação textual da parada ou None caso a operação falhe.
        :rtype: str | None
        """
        try:
            return self._cases_api.stop_caso_execution(uuid=self._as_uuid(caso_id), _headers={"Accept": "text/plain"})
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível interromper a execução: %s",
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def list_executions(self, auth_node_uuid: Union[UUID, str]) -> List[ExecutionStatus]:
        """
        Lista as execuções associadas a um AuthNode.

        :param auth_node_uuid: Identificador único do AuthNode.
        :type auth_node_uuid: UUID ou str
        :return: Execuções encontradas para o AuthNode.
        :rtype: List[ExecutionStatus]
        """
        try:
            return list(
                self._auth_nodes_api.list_executions_by_auth_node(
                    self._as_uuid(auth_node_uuid)
                )
                or []
            )
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível listar execuções do AuthNode %s: %s",
                auth_node_uuid,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return []

get_execution_console_logs(caso_id)

Obtém a saída do console da execução.

Parameters:

Name Type Description Default
caso_id Union[str, UUID]

Identificador único do caso.

 required

Returns:

Type Description
str | None

Conteúdo textual do console ou None caso a consulta falhe.
Source code in pldpro_sdk/core/proxies/execucoes.py 
100
101
102
103
104
105
106
107
108
109
def get_execution_console_logs(self, caso_id: Union[str, UUID]) -> str | None:
    """
    Obtém a saída do console da execução.

    :param caso_id: Identificador único do caso.
    :type caso_id: UUID ou str
    :return: Conteúdo textual do console ou None caso a consulta falhe.
    :rtype: str | None
    """
    return self._get_execution_text(caso_id, self._cases_api.get_caso_execution_console_logs_without_preload_content)

get_execution_logs(caso_id)

Obtém o conteúdo do log da execução.

Parameters:

Name Type Description Default
caso_id str | UUID

Identificador único do caso.

 required

Returns:

Type Description
str | None

Conteúdo textual do log ou None caso a consulta falhe.
Source code in pldpro_sdk/core/proxies/execucoes.py 
89
90
91
92
93
94
95
96
97
98
def get_execution_logs(self, caso_id: str | UUID) -> str | None:
    """
    Obtém o conteúdo do log da execução.

    :param caso_id: Identificador único do caso.
    :type caso_id: UUID ou str
    :return: Conteúdo textual do log ou None caso a consulta falhe.
    :rtype: str | None
    """
    return self._get_execution_text(caso_id, self._cases_api.get_caso_execution_logs_without_preload_content)

get_execution_status(caso_id)

Consulta o status da execução de um caso.

Parameters:

Name Type Description Default
caso_id str | UUID

Identificador único do caso.

 required

Returns:

Type Description
ExecutionStatus | None

Status da execução ou None caso a consulta falhe.
Source code in pldpro_sdk/core/proxies/execucoes.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def get_execution_status(self, caso_id: str | UUID):
    """
    Consulta o status da execução de um caso.

    :param caso_id: Identificador único do caso.
    :type caso_id: UUID ou str
    :return: Status da execução ou None caso a consulta falhe.
    :rtype: ExecutionStatus | None
    """
    try:
        return self._cases_api.get_caso_execution_status(uuid=self._as_uuid(caso_id))
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível consultar o status da execução: %s",
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

list_executions(auth_node_uuid)

Lista as execuções associadas a um AuthNode.

Parameters:

Name Type Description Default
auth_node_uuid Union[UUID, str]

Identificador único do AuthNode.

 required

Returns:

Type Description
List[ExecutionStatus]

Execuções encontradas para o AuthNode.
Source code in pldpro_sdk/core/proxies/execucoes.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def list_executions(self, auth_node_uuid: Union[UUID, str]) -> List[ExecutionStatus]:
    """
    Lista as execuções associadas a um AuthNode.

    :param auth_node_uuid: Identificador único do AuthNode.
    :type auth_node_uuid: UUID ou str
    :return: Execuções encontradas para o AuthNode.
    :rtype: List[ExecutionStatus]
    """
    try:
        return list(
            self._auth_nodes_api.list_executions_by_auth_node(
                self._as_uuid(auth_node_uuid)
            )
            or []
        )
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível listar execuções do AuthNode %s: %s",
            auth_node_uuid,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return []

start_execution(caso_id, execution_start_request=None)

Inicia a execução de um caso.

Parameters:

Name Type Description Default
caso_id Union[str, UUID]

Identificador único do caso a ser executado.

 required
execution_start_request ExecutionStartRequest | None

Configuração da execução. Quando não informada, usa a configuração padrão do SDK.

 None

Returns:

Type Description
ExecutionStatus | None

Status da execução iniciada ou None caso a operação falhe.
Source code in pldpro_sdk/core/proxies/execucoes.py 
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
def start_execution(
    self,
    caso_id: Union[str, UUID],
    execution_start_request: ExecutionStartRequest | None = None,
):
    """
    Inicia a execução de um caso.

    :param caso_id: Identificador único do caso a ser executado.
    :type caso_id: UUID ou str
    :param execution_start_request: Configuração da execução. Quando não informada, usa a configuração padrão do SDK.
    :type execution_start_request: ExecutionStartRequest | None
    :return: Status da execução iniciada ou None caso a operação falhe.
    :rtype: ExecutionStatus | None
    """
    try:
        case_uuid = self._as_uuid(caso_id)
        if execution_start_request is None:
            execution_start_request = ExecutionStartRequest(
                version="7.0.0",
                tipoScript="COMPLETE",
                listaUUIDAuthNodes=[],
                optionsExecutaModelo=OptionsExecutaModelo(
                    localExecution="CLOUD",
                    projectId="",
                    reservationId="",
                    machineType="c2std60",
                ),
            )
        return self._cases_api.start_caso_execution(
            uuid=case_uuid,
            execution_start_request=execution_start_request,
        )
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível iniciar a execução: %s",
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

stop_execution(caso_id)

Interrompe a execução de um caso.

Parameters:

Name Type Description Default
caso_id Union[str, UUID]

Identificador único do caso.

 required

Returns:

Type Description
str | None

Confirmação textual da parada ou None caso a operação falhe.
Source code in pldpro_sdk/core/proxies/execucoes.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def stop_execution(self, caso_id: Union[str, UUID]) -> str | None:
    """
    Interrompe a execução de um caso.

    :param caso_id: Identificador único do caso.
    :type caso_id: UUID ou str
    :return: Confirmação textual da parada ou None caso a operação falhe.
    :rtype: str | None
    """
    try:
        return self._cases_api.stop_caso_execution(uuid=self._as_uuid(caso_id), _headers={"Accept": "text/plain"})
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível interromper a execução: %s",
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None