Pular para conteúdo

execucoes

Módulo responsável por encapsular as operações de Execuções no SDK PLDPro.

Cada metodo desta classe é um wrapper seguro e tipado para os endpoints de execução disponíveis na API Java.

Endpoints cobertos: - POST /casos/{uuid}/execution -> start_execution - DELETE /casos/{uuid}/execution -> stop_execution - GET /casos/{uuid}/execution/logs -> get_execution_logs - GET /casos/{uuid}/execution/logs/console -> get_execution_console_logs - GET /casos/{uuid}/execution/status -> get_execution_status - GET /host/{uuid}/execution -> list_executions

Execucoes

Source code in pldpro_sdk/core/proxies/execucoes.py 
 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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class Execucoes:
    def __init__(self, api_client: ApiClient) -> None:
        self._exec_api = ExecucoesApi(api_client)

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

        - Para hosts com modo NATIVE, o campo `versao_container` é opcional.
        - Para hosts com modo CONTAINER, o campo `versao_container` é obrigatório.

        Args:
            caso_id (UUID|str): Identificador do caso.
            execucao_start_request (ExecucaoStartRequest, optional): Requisição de início de execução.

        Returns:
            ExecucaoResponse | None
        """
        try:
            if isinstance(caso_id, str):
                caso_id = UUID(caso_id)

            if execucao_start_request is None:
                options_executa_modelo = ExecucaoStartRequestOptionsExecutaModelo(
                    localExecution=TypeNodeExecutionEnum.CLOUD,
                    projectId="",
                    reservationId="",
                    machineType="c2std60",
                )

                execucao_start_request = ExecucaoStartRequest(
                    version="7.0.0", tipoEntrada="DEFAULT",
                    listaUUIDAuthNodes=[],
                    optionsExecutaModelo=options_executa_modelo
                )

            return self._exec_api.start_caso_execution(
                uuid=caso_id, execucao_start_request=execucao_start_request
            )

        except ApiException as e:
            logger.exception(
                f"Erro ao iniciar execução: {get_error_message(e.body)}", exc_info=True
            )
            return None

    def get_execution_status(self, execucao_id: Union[str, UUID]):
        """
        Retorna o status da execução associada ao UUID informado.

        Args:
            execucao_id (UUID|str): Identificador da execução.

        Returns:
            ExecucaoResponse | None
        """
        try:
            if isinstance(execucao_id, str):
                execucao_id = UUID(execucao_id)
            return self._exec_api.get_caso_execution_status(uuid=execucao_id)
        except ApiException as e:
            logger.exception(
                f"Erro ao consultar status da execução: {get_error_message(e)}",
                exc_info=True,
            )
            return None

    def get_execution_logs(self, caso_id: Union[str, UUID]):
        """
        Obtém o conteúdo textual do log da execução (arquivo de log).
        Args:
            caso_id (UUID|str): Identificador do caso.

        Returns:
            str | None
        """
        try:
            if isinstance(caso_id, str):
                caso_id = UUID(caso_id)

            response = self._exec_api.get_caso_execution_logs_without_preload_content(
                caso_id
            )

            # Se a resposta veio como bytes, decodifica
            if hasattr(response, "data"):
                raw = response.data
            elif hasattr(response, "read"):
                raw = response.read()
            else:
                raw = response

            if isinstance(raw, bytes):
                return raw.decode("utf-8", errors="ignore")
            if isinstance(raw, str):
                return raw.strip()
            return str(raw)

        except ApiException as e:
            logger.exception(
                f"Erro ao obter logs da execução: {get_error_message(e)}", exc_info=True
            )
            return None

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

        Args:
            caso_id (UUID|str): Identificador do caso.

        Returns:
            str | None
        """
        try:
            if isinstance(caso_id, str):
                caso_id = UUID(caso_id)

            response = (
                self._exec_api.get_caso_execution_console_logs_without_preload_content(
                    caso_id
                )
            )

            if hasattr(response, "data"):
                raw = response.data
            elif hasattr(response, "read"):
                raw = response.read()
            else:
                raw = response

            if isinstance(raw, bytes):
                return raw.decode("utf-8", errors="ignore")
            if isinstance(raw, str):
                return raw.strip()

            return str(raw)

        except ApiException as e:
            logger.exception(
                f"Erro ao obter logs de console: {get_error_message(e)}", exc_info=True
            )
            return None

    def stop_execution(self, execucao_id: Union[str, UUID]):
        """
        Interrompe uma execução em andamento (idempotente).

        Args:
            execucao_id (UUID|str): Identificador da execução.

        Returns:
            str | None
        """
        try:
            if isinstance(execucao_id, str):
                execucao_id = UUID(execucao_id)
            return self._exec_api.stop_caso_execution(uuid=execucao_id)
        except ApiException as e:
            msg = get_error_message(e)
            if "não encontrada" in msg.lower() or "finalizada" in msg.lower():
                print("Execução já finalizada. Nenhuma ação necessária.")
                return None
            logger.exception(f"Erro ao interromper execução: {msg}", exc_info=True)
            return None

    def list_executions(self, auth_node_uuid: Union[UUID, str]) -> List[StatusExecucao]:
        """
        Lista as execuções associadas a um host específico.

        Args:
            auth_node_uuid (UUID|str): Identificador único do host.
        Return:
            str | Array []
        """
        try:
            if isinstance(auth_node_uuid, str):
                auth_node_uuid = UUID(auth_node_uuid)

            api_response = self._exec_api.list_executions_by_auth_node(auth_node_uuid)
            return api_response
        except (ApiException, ValueError) as e:
            error_msg = get_error_message(e) if isinstance(e, ApiException) else str(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar as execuções do host {auth_node_uuid}:\n\t {error_msg}"
            )
        return []

get_execution_console_logs(caso_id)

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

Args: caso_id (UUID|str): Identificador do caso.

Returns: str | None

Source code in pldpro_sdk/core/proxies/execucoes.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def get_execution_console_logs(self, caso_id: Union[str, UUID]):
    """
    Obtém a saída do console (stdout) da execução.

    Args:
        caso_id (UUID|str): Identificador do caso.

    Returns:
        str | None
    """
    try:
        if isinstance(caso_id, str):
            caso_id = UUID(caso_id)

        response = (
            self._exec_api.get_caso_execution_console_logs_without_preload_content(
                caso_id
            )
        )

        if hasattr(response, "data"):
            raw = response.data
        elif hasattr(response, "read"):
            raw = response.read()
        else:
            raw = response

        if isinstance(raw, bytes):
            return raw.decode("utf-8", errors="ignore")
        if isinstance(raw, str):
            return raw.strip()

        return str(raw)

    except ApiException as e:
        logger.exception(
            f"Erro ao obter logs de console: {get_error_message(e)}", exc_info=True
        )
        return None

get_execution_logs(caso_id)

Obtém o conteúdo textual do log da execução (arquivo de log). Args: caso_id (UUID|str): Identificador do caso.

Returns: str | None

Source code in pldpro_sdk/core/proxies/execucoes.py 
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
def get_execution_logs(self, caso_id: Union[str, UUID]):
    """
    Obtém o conteúdo textual do log da execução (arquivo de log).
    Args:
        caso_id (UUID|str): Identificador do caso.

    Returns:
        str | None
    """
    try:
        if isinstance(caso_id, str):
            caso_id = UUID(caso_id)

        response = self._exec_api.get_caso_execution_logs_without_preload_content(
            caso_id
        )

        # Se a resposta veio como bytes, decodifica
        if hasattr(response, "data"):
            raw = response.data
        elif hasattr(response, "read"):
            raw = response.read()
        else:
            raw = response

        if isinstance(raw, bytes):
            return raw.decode("utf-8", errors="ignore")
        if isinstance(raw, str):
            return raw.strip()
        return str(raw)

    except ApiException as e:
        logger.exception(
            f"Erro ao obter logs da execução: {get_error_message(e)}", exc_info=True
        )
        return None

get_execution_status(execucao_id)

Retorna o status da execução associada ao UUID informado.

Args: execucao_id (UUID|str): Identificador da execução.

Returns: ExecucaoResponse | None

Source code in pldpro_sdk/core/proxies/execucoes.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_execution_status(self, execucao_id: Union[str, UUID]):
    """
    Retorna o status da execução associada ao UUID informado.

    Args:
        execucao_id (UUID|str): Identificador da execução.

    Returns:
        ExecucaoResponse | None
    """
    try:
        if isinstance(execucao_id, str):
            execucao_id = UUID(execucao_id)
        return self._exec_api.get_caso_execution_status(uuid=execucao_id)
    except ApiException as e:
        logger.exception(
            f"Erro ao consultar status da execução: {get_error_message(e)}",
            exc_info=True,
        )
        return None

list_executions(auth_node_uuid)

Lista as execuções associadas a um host específico.

Args: auth_node_uuid (UUID|str): Identificador único do host. Return: str | Array []

Source code in pldpro_sdk/core/proxies/execucoes.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def list_executions(self, auth_node_uuid: Union[UUID, str]) -> List[StatusExecucao]:
    """
    Lista as execuções associadas a um host específico.

    Args:
        auth_node_uuid (UUID|str): Identificador único do host.
    Return:
        str | Array []
    """
    try:
        if isinstance(auth_node_uuid, str):
            auth_node_uuid = UUID(auth_node_uuid)

        api_response = self._exec_api.list_executions_by_auth_node(auth_node_uuid)
        return api_response
    except (ApiException, ValueError) as e:
        error_msg = get_error_message(e) if isinstance(e, ApiException) else str(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar as execuções do host {auth_node_uuid}:\n\t {error_msg}"
        )
    return []

start_execution(caso_id, execucao_start_request=None)

Inicia a execução de um caso.

  • Para hosts com modo NATIVE, o campo versao_container é opcional.
  • Para hosts com modo CONTAINER, o campo versao_container é obrigatório.

Args: caso_id (UUID|str): Identificador do caso. execucao_start_request (ExecucaoStartRequest, optional): Requisição de início de execução.

Returns: ExecucaoResponse | None

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

    - Para hosts com modo NATIVE, o campo `versao_container` é opcional.
    - Para hosts com modo CONTAINER, o campo `versao_container` é obrigatório.

    Args:
        caso_id (UUID|str): Identificador do caso.
        execucao_start_request (ExecucaoStartRequest, optional): Requisição de início de execução.

    Returns:
        ExecucaoResponse | None
    """
    try:
        if isinstance(caso_id, str):
            caso_id = UUID(caso_id)

        if execucao_start_request is None:
            options_executa_modelo = ExecucaoStartRequestOptionsExecutaModelo(
                localExecution=TypeNodeExecutionEnum.CLOUD,
                projectId="",
                reservationId="",
                machineType="c2std60",
            )

            execucao_start_request = ExecucaoStartRequest(
                version="7.0.0", tipoEntrada="DEFAULT",
                listaUUIDAuthNodes=[],
                optionsExecutaModelo=options_executa_modelo
            )

        return self._exec_api.start_caso_execution(
            uuid=caso_id, execucao_start_request=execucao_start_request
        )

    except ApiException as e:
        logger.exception(
            f"Erro ao iniciar execução: {get_error_message(e.body)}", exc_info=True
        )
        return None

stop_execution(execucao_id)

Interrompe uma execução em andamento (idempotente).

Args: execucao_id (UUID|str): Identificador da execução.

Returns: str | None

Source code in pldpro_sdk/core/proxies/execucoes.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def stop_execution(self, execucao_id: Union[str, UUID]):
    """
    Interrompe uma execução em andamento (idempotente).

    Args:
        execucao_id (UUID|str): Identificador da execução.

    Returns:
        str | None
    """
    try:
        if isinstance(execucao_id, str):
            execucao_id = UUID(execucao_id)
        return self._exec_api.stop_caso_execution(uuid=execucao_id)
    except ApiException as e:
        msg = get_error_message(e)
        if "não encontrada" in msg.lower() or "finalizada" in msg.lower():
            print("Execução já finalizada. Nenhuma ação necessária.")
            return None
        logger.exception(f"Erro ao interromper execução: {msg}", exc_info=True)
        return None