nodes

Proxy para operações de Node e AuthNode do SDK do PLDPro.

Nodes

Operações para cadastrar e consultar Nodes e AuthNodes.

Source code in pldpro_sdk/core/proxies/nodes.py

class Nodes:
    """
    Operações para cadastrar e consultar Nodes e AuthNodes.
    """

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

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

    @staticmethod
    def _save_public_key(
        public_key: str,
        usuario: str,
        output_path: Optional[str],
    ) -> str:
        destination = Path(output_path or Path.cwd()) / f"{usuario}_key.pub"
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_text(
            public_key if public_key.endswith("\n") else f"{public_key}\n",
            encoding="utf-8",
        )
        return str(destination)
    def register(
        self,
        registration: FullNodeRegistrationDTO | NodeRequestDTO | NodeRequest,
    ) -> Optional[NodeResponseDTO]:
        """
        Registra um Node ou um Node com AuthNode.

        :param registration: Dados do cadastro do Node. Pode incluir as credenciais remotas do AuthNode.
        :type registration: FullNodeRegistrationDTO, NodeRequestDTO ou NodeRequest (Evitar usar. Deverá ser descontinuado
        quando evoluirmos o sdk para a próxima major)
        :return: Node registrado ou None caso a operação falhe.
        :rtype: NodeResponseDTO | None
        """
        try:
            if isinstance(registration, FullNodeRegistrationDTO):
                return self._nodes_api.register_full_node(registration)

            if isinstance(registration, NodeRequest):
                registration = NodeRequestDTO(
                    name=registration.nome,
                    ip=registration.ip,
                    dns=registration.dns,
                    port=registration.porta,
                    lockToServer=registration.lock_to_server,
                )

            if isinstance(registration, NodeRequestDTO):
                return self._nodes_api.create_node(registration)

            raise TypeError("Use FullNodeRegistrationDTO, NodeRequestDTO ou NodeRequest.")
        except (ApiException, TypeError, ValueError) as error:
            logger.error(
                "Não foi possível registrar o Node: %s",
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def create_node(self, node_request: NodeRequestDTO) -> Optional[NodeResponseDTO]:
        """
        Cria um Node.

        :param node_request: Dados do Node a ser criado.
        :type node_request: NodeRequestDTO
        :return: Node criado ou None caso a operação falhe.
        :rtype: NodeResponseDTO | None
        """
        try:
            return self._nodes_api.create_node(node_request)
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível registrar o Node: %s",
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def unregister(self, node_id: NodeId) -> bool:
        """
        Remove um Node.

        :param node_id: Identificador único do Node.
        :type node_id: UUID ou str
        :return: True quando o Node é removido; False caso a operação falhe.
        :rtype: bool
        """
        try:
            node_uuid = self._as_uuid(node_id)
            self._nodes_api.delete_node(node_uuid)
            logger.info("Node %s removido com sucesso.", node_uuid)
            return True
        except (ApiException, ValueError) as error:
            logger.error(
                "O Node %s não pode ser removido: %s",
                node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return False

    def list_nodes(self) -> List[NodeResponseDTO]:
        """
        Lista os Nodes disponíveis para o usuário autenticado.

        :return: Nodes encontrados.
        :rtype: List[NodeResponseDTO]
        """
        try:
            return list(self._nodes_api.list_nodes() or [])
        except ApiException as error:
            logger.error("Não foi possível listar Nodes: %s", get_error_message(error))
            return []

    def get_node(self, node_id: NodeId) -> Optional[NodeResponseDTO]:
        """
        Consulta um Node pelo identificador.

        :param node_id: Identificador único do Node.
        :type node_id: UUID ou str
        :return: Node encontrado ou None caso não exista ou a consulta falhe.
        :rtype: NodeResponseDTO | None
        """
        try:
            return self._nodes_api.get_node_by_uuid(self._as_uuid(node_id))
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível consultar o Node %s: %s",
                node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def get_node_from_name(self, node_name: str) -> Optional[NodeResponseDTO]:
        """
        Consulta um Node pelo nome.

        :param node_name: Nome do Node procurado.
        :type node_name: str
        :return: Node encontrado ou None caso não exista.
        :rtype: NodeResponseDTO | None
        """
        normalized_name = node_name.strip().casefold()
        for node in self.list_nodes():
            if node.name and node.name.casefold() == normalized_name:
                return node

        logger.warning("Node não cadastrado ou não autorizado: %s", node_name)
        return None

    def list_node_casos(self, node_id: NodeId) -> List[CaseResponse]:
        """
        Lista os casos associados a um Node.

        :param node_id: Identificador único do Node.
        :type node_id: UUID ou str
        :return: Casos associados ao Node.
        :rtype: List[CaseResponse]
        """
        try:
            return list(self._nodes_api.get_casos_node(self._as_uuid(node_id)) or [])
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível listar casos do Node %s: %s",
                node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return []

    def list_auth_nodes(self, node_id: NodeId) -> List[AuthNodeResponseDTO]:
        """
        Lista os AuthNodes associados a um Node.

        :param node_id: Identificador único do Node.
        :type node_id: UUID ou str
        :return: AuthNodes associados ao Node.
        :rtype: List[AuthNodeResponseDTO]
        """
        try:
            node_uuid = self._as_uuid(node_id)
            auth_nodes = self._auth_nodes_api.list_auth_nodes() or []
            return [auth_node for auth_node in auth_nodes if auth_node.node_id == node_uuid]
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível listar AuthNodes do Node %s: %s",
                node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return []

    def create_auth_node(
        self,
        node_id: NodeId,
        user_id: NodeId,
        login: str,
        work_dir: str,
        password: Optional[str] = None,
        private_key: Optional[str] = None,
        token: Optional[str] = None,
        role_node: Optional[str] = None,
    ) -> Optional[AuthNodeResponseDTO]:
        """
        Cria um AuthNode associado a um Node.

        :param node_id: Identificador único do Node associado ao AuthNode.
        :type node_id: UUID ou str
        :param user_id: Identificador único do usuário proprietário do AuthNode.
        :type user_id: UUID ou str
        :param login: Nome de usuário usado na autenticação remota.
        :type login: str
        :param work_dir: Diretório de trabalho do usuário no servidor remoto.
        :type work_dir: str
        :param password: Senha usada na autenticação remota, quando aplicável.
        :type password: Optional[str]
        :param private_key: Chave privada usada na autenticação remota, quando aplicável.
        :type private_key: Optional[str]
        :param token: Token usado na autenticação remota, quando aplicável.
        :type token: Optional[str]
        :param role_node: Papel atribuído ao AuthNode no Node.
        :type role_node: Optional[str]
        :return: AuthNode criado ou None caso a operação falhe.
        :rtype: AuthNodeResponseDTO | None
        """
        try:
            auth_node_pass = (
                AuthNodePassRequestDTO(login=login, password=password)
                if login
                else None
            )
            request = AuthNodeRequestDTO(
                node_id=self._as_uuid(node_id),
                user_id=self._as_uuid(user_id),
                work_dir=work_dir,
                role_node=role_node,
                auth_node_pass=auth_node_pass,
                auth_node_key=AuthNodeKeyRequestDTO(private_key=private_key)
                if private_key
                else None,
                auth_node_token=AuthNodeTokenRequestDTO(token=token) if token else None,
            )
            return self._auth_nodes_api.create_auth_node(request)
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível criar AuthNode para o Node %s: %s",
                node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def get_auth_node(self, auth_node_id: NodeId) -> Optional[AuthNodeResponseDTO]:
        """
        Consulta um AuthNode pelo identificador.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :return: AuthNode encontrado ou None caso não exista ou a consulta falhe.
        :rtype: AuthNodeResponseDTO | None
        """
        try:
            return self._auth_nodes_api.get_auth_node_by_uuid(self._as_uuid(auth_node_id))
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível consultar AuthNode %s: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def get_auth_node_from_user(
        self,
        node_id: NodeId,
        login: str,
    ) -> Optional[AuthNodeResponseDTO]:
        """
        Consulta um AuthNode pelo Node e pelo usuário remoto.

        :param node_id: Identificador único do Node.
        :type node_id: UUID ou str
        :param login: Nome de usuário remoto associado ao AuthNode.
        :type login: str
        :return: AuthNode encontrado ou None caso não exista.
        :rtype: AuthNodeResponseDTO | None
        """
        normalized_login = login.strip().casefold()
        for auth_node in self.list_auth_nodes(node_id):
            configured_login = (
                auth_node.auth_node_pass.login
                if auth_node.auth_node_pass is not None
                else None
            )
            if configured_login and configured_login.casefold() == normalized_login:
                return auth_node

        logger.warning("AuthNode não encontrado para login %s no Node %s.", login, node_id)
        return None

    def delete_auth_node(self, auth_node_id: NodeId) -> bool:
        """
        Remove um AuthNode.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :return: True quando o AuthNode é removido; False caso a operação falhe.
        :rtype: bool
        """
        try:
            auth_node_uuid = self._as_uuid(auth_node_id)
            self._auth_nodes_api.delete_auth_node(auth_node_uuid)
            logger.info("AuthNode %s removido com sucesso.", auth_node_uuid)
            return True
        except (ApiException, ValueError) as error:
            logger.error(
                "O AuthNode %s não pode ser removido: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return False

    def create_managed_ssh_auth_node_access(
        self,
        node_id: NodeId,
        login: str,
        work_dir: str,
        password: Optional[str] = None,
    ) -> Optional[AuthNodeSshAccessResponseDTO]:
        """Cria um AuthNode com par de chaves SSH gerenciado pela API.

        :param node_id: Identificador único do Node associado ao AuthNode.
        :type node_id: UUID ou str
        :param login: Usuário remoto usado na conexão SSH.
        :type login: str
        :param work_dir: Diretório de trabalho remoto do usuário.
        :type work_dir: str
        :param password: Senha usada para instalar a chave pública no host remoto.
        :type password: Optional[str]
        :return: AuthNode criado, chave pública e o resultado do provisionamento.
        :rtype: AuthNodeSshAccessResponseDTO | None
        """
        try:
            request = AuthNodeSshAccessRequestDTO(
                node_id=self._as_uuid(node_id),
                login=login,
                work_dir=work_dir,
                password=password,
            )
            return self._auth_nodes_api.create_managed_ssh_auth_node_access(request)
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível criar o acesso SSH gerenciado para o Node %s: %s",
                node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def create_auth_node_and_download_key(
        self,
        node_uuid: NodeId,
        usuario: str,
        workdir: str,
        output_path: Optional[str] = None,
        password: Optional[str] = None,
    ) -> Optional[str]:
        """Cria um AuthNode e salva sua chave pública.

        :param node_uuid: Identificador único do Node associado ao AuthNode.
        :type node_uuid: UUID ou str
        :param usuario: Nome do usuário remoto.
        :type usuario: str
        :param workdir: Diretório de trabalho do usuário remoto.
        :type workdir: str
        :param output_path: Diretório onde a chave pública será salva.
        :type output_path: Optional[str]
        :param password: Senha usada para instalar a chave pública no host remoto.
        :type password: Optional[str]
        :return: Caminho da chave pública salva ou None caso a operação falhe.
        :rtype: str | None
        """
        try:
            response = self.create_managed_ssh_auth_node_access(
                node_id=node_uuid,
                login=usuario,
                work_dir=workdir,
                password=password,
            )
            if response is None or not response.public_key:
                return None
            return self._save_public_key(response.public_key, usuario, output_path)
        except OSError as error:
            logger.error(
                "Não foi possível criar a chave pública para o Node %s: %s",
                node_uuid,
                error,
            )
            return None

    def get_auth_node_file_public_key(
        self,
        node_name: str,
        usuario: str,
        output_path: Optional[str] = None,
    ) -> Optional[str]:
        """Baixa a chave pública de um AuthNode.

        :param node_name: Nome do Node associado ao AuthNode.
        :type node_name: str
        :param usuario: Nome do usuário remoto associado ao AuthNode.
        :type usuario: str
        :param output_path: Diretório onde a chave pública será salva.
        :type output_path: Optional[str]
        :return: Caminho da chave pública salva ou None caso a operação falhe.
        :rtype: str | None
        """
        try:
            node = self.get_node_from_name(node_name)
            if node is None or node.id is None:
                return None

            auth_node = self.get_auth_node_from_user(node.id, usuario)
            if auth_node is None or auth_node.id is None:
                return None

            public_key = self.get_auth_node_public_key(auth_node.id)
            if not public_key:
                return None
            return self._save_public_key(public_key, usuario, output_path)
        except (ApiException, OSError, ValueError) as error:
            logger.error(
                "Não foi possível baixar a chave pública de %s no Node %s: %s",
                usuario,
                node_name,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def test_node_connection(self, auth_node_id: NodeId) -> Optional[HostCheckResult]:
        """Verifica a conectividade de um AuthNode.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :return: Resultado da verificação ou None caso a consulta falhe.
        :rtype: HostCheckResult | None
        """
        try:
            return self._auth_nodes_api.check_auth_node_health(self._as_uuid(auth_node_id))
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível verificar o AuthNode %s: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return None

    def list_versions(self, auth_node_id: NodeId) -> List[str]:
        """
        Lista as versões disponíveis em um AuthNode.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :return: Versões disponíveis.
        :rtype: List[str]
        """
        try:
            return list(
                self._auth_nodes_api.get_versoes_container(self._as_uuid(auth_node_id))
                or []
            )
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível listar versões do AuthNode %s: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return []

    def provision_managed_ssh_auth_node_public_key(
        self,
        auth_node_id: NodeId,
    ) -> bool:
        """Instala no host a chave pública gerenciada de um AuthNode.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :return: True quando a chave foi instalada; False caso a operação falhe.
        :rtype: bool
        """
        try:
            self._auth_nodes_api.provision_managed_ssh_auth_node_public_key(
                self._as_uuid(auth_node_id)
            )
            return True
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível instalar a chave pública do AuthNode %s: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return False

    def rotate_managed_ssh_auth_node_key_pair(
        self,
        auth_node_id: NodeId,
        password: str,
    ) -> str:
        """Rotaciona o par de chaves SSH gerenciado de um AuthNode.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :param password: Senha usada para instalar a nova chave pública no host remoto.
        :type password: str
        :return: Nova chave pública ou uma string vazia caso a operação falhe.
        :rtype: str
        """
        try:
            response = self._auth_nodes_api.rotate_managed_ssh_auth_node_key_pair(
                self._as_uuid(auth_node_id),
                AuthNodeSshKeyRotationRequestDTO(
                    password=password
                ),
            )
            return response.public_key or ""
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível rotacionar as chaves do AuthNode %s: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return ""

    def get_auth_node_public_key(self, auth_node_id: NodeId) -> str:
        """Obtém a chave pública SSH de um AuthNode.

        :param auth_node_id: Identificador único do AuthNode.
        :type auth_node_id: UUID ou str
        :return: Conteúdo da chave pública ou uma string vazia caso a consulta falhe.
        :rtype: str
        """
        try:
            response = self._auth_nodes_api.get_managed_ssh_auth_node_public_key(
                self._as_uuid(auth_node_id)
            )
            return response.public_key or ""
        except (ApiException, ValueError) as error:
            logger.error(
                "Não foi possível obter chave pública do AuthNode %s: %s",
                auth_node_id,
                get_error_message(error) if isinstance(error, ApiException) else error,
            )
            return ""

create_auth_node(node_id, user_id, login, work_dir, password=None, private_key=None, token=None, role_node=None)

Cria um AuthNode associado a um Node.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node associado ao AuthNode.	required
user_id	NodeId	Identificador único do usuário proprietário do AuthNode.	required
login	str	Nome de usuário usado na autenticação remota.	required
work_dir	str	Diretório de trabalho do usuário no servidor remoto.	required
password	Optional[str]	Senha usada na autenticação remota, quando aplicável.	None
private_key	Optional[str]	Chave privada usada na autenticação remota, quando aplicável.	None
token	Optional[str]	Token usado na autenticação remota, quando aplicável.	None
role_node	Optional[str]	Papel atribuído ao AuthNode no Node.	None

Returns:

Type	Description
AuthNodeResponseDTO | None	AuthNode criado ou None caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def create_auth_node(
    self,
    node_id: NodeId,
    user_id: NodeId,
    login: str,
    work_dir: str,
    password: Optional[str] = None,
    private_key: Optional[str] = None,
    token: Optional[str] = None,
    role_node: Optional[str] = None,
) -> Optional[AuthNodeResponseDTO]:
    """
    Cria um AuthNode associado a um Node.

    :param node_id: Identificador único do Node associado ao AuthNode.
    :type node_id: UUID ou str
    :param user_id: Identificador único do usuário proprietário do AuthNode.
    :type user_id: UUID ou str
    :param login: Nome de usuário usado na autenticação remota.
    :type login: str
    :param work_dir: Diretório de trabalho do usuário no servidor remoto.
    :type work_dir: str
    :param password: Senha usada na autenticação remota, quando aplicável.
    :type password: Optional[str]
    :param private_key: Chave privada usada na autenticação remota, quando aplicável.
    :type private_key: Optional[str]
    :param token: Token usado na autenticação remota, quando aplicável.
    :type token: Optional[str]
    :param role_node: Papel atribuído ao AuthNode no Node.
    :type role_node: Optional[str]
    :return: AuthNode criado ou None caso a operação falhe.
    :rtype: AuthNodeResponseDTO | None
    """
    try:
        auth_node_pass = (
            AuthNodePassRequestDTO(login=login, password=password)
            if login
            else None
        )
        request = AuthNodeRequestDTO(
            node_id=self._as_uuid(node_id),
            user_id=self._as_uuid(user_id),
            work_dir=work_dir,
            role_node=role_node,
            auth_node_pass=auth_node_pass,
            auth_node_key=AuthNodeKeyRequestDTO(private_key=private_key)
            if private_key
            else None,
            auth_node_token=AuthNodeTokenRequestDTO(token=token) if token else None,
        )
        return self._auth_nodes_api.create_auth_node(request)
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível criar AuthNode para o Node %s: %s",
            node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

create_auth_node_and_download_key(node_uuid, usuario, workdir, output_path=None, password=None)

Cria um AuthNode e salva sua chave pública.

Parameters:

Name	Type	Description	Default
node_uuid	NodeId	Identificador único do Node associado ao AuthNode.	required
usuario	str	Nome do usuário remoto.	required
workdir	str	Diretório de trabalho do usuário remoto.	required
output_path	Optional[str]	Diretório onde a chave pública será salva.	None
password	Optional[str]	Senha usada para instalar a chave pública no host remoto.	None

Returns:

Type	Description
str | None	Caminho da chave pública salva ou None caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def create_auth_node_and_download_key(
    self,
    node_uuid: NodeId,
    usuario: str,
    workdir: str,
    output_path: Optional[str] = None,
    password: Optional[str] = None,
) -> Optional[str]:
    """Cria um AuthNode e salva sua chave pública.

    :param node_uuid: Identificador único do Node associado ao AuthNode.
    :type node_uuid: UUID ou str
    :param usuario: Nome do usuário remoto.
    :type usuario: str
    :param workdir: Diretório de trabalho do usuário remoto.
    :type workdir: str
    :param output_path: Diretório onde a chave pública será salva.
    :type output_path: Optional[str]
    :param password: Senha usada para instalar a chave pública no host remoto.
    :type password: Optional[str]
    :return: Caminho da chave pública salva ou None caso a operação falhe.
    :rtype: str | None
    """
    try:
        response = self.create_managed_ssh_auth_node_access(
            node_id=node_uuid,
            login=usuario,
            work_dir=workdir,
            password=password,
        )
        if response is None or not response.public_key:
            return None
        return self._save_public_key(response.public_key, usuario, output_path)
    except OSError as error:
        logger.error(
            "Não foi possível criar a chave pública para o Node %s: %s",
            node_uuid,
            error,
        )
        return None

create_managed_ssh_auth_node_access(node_id, login, work_dir, password=None)

Cria um AuthNode com par de chaves SSH gerenciado pela API.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node associado ao AuthNode.	required
login	str	Usuário remoto usado na conexão SSH.	required
work_dir	str	Diretório de trabalho remoto do usuário.	required
password	Optional[str]	Senha usada para instalar a chave pública no host remoto.	None

Returns:

Type	Description
AuthNodeSshAccessResponseDTO | None	AuthNode criado, chave pública e o resultado do provisionamento.

Source code in pldpro_sdk/core/proxies/nodes.py

def create_managed_ssh_auth_node_access(
    self,
    node_id: NodeId,
    login: str,
    work_dir: str,
    password: Optional[str] = None,
) -> Optional[AuthNodeSshAccessResponseDTO]:
    """Cria um AuthNode com par de chaves SSH gerenciado pela API.

    :param node_id: Identificador único do Node associado ao AuthNode.
    :type node_id: UUID ou str
    :param login: Usuário remoto usado na conexão SSH.
    :type login: str
    :param work_dir: Diretório de trabalho remoto do usuário.
    :type work_dir: str
    :param password: Senha usada para instalar a chave pública no host remoto.
    :type password: Optional[str]
    :return: AuthNode criado, chave pública e o resultado do provisionamento.
    :rtype: AuthNodeSshAccessResponseDTO | None
    """
    try:
        request = AuthNodeSshAccessRequestDTO(
            node_id=self._as_uuid(node_id),
            login=login,
            work_dir=work_dir,
            password=password,
        )
        return self._auth_nodes_api.create_managed_ssh_auth_node_access(request)
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível criar o acesso SSH gerenciado para o Node %s: %s",
            node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

create_node(node_request)

Cria um Node.

Parameters:

Name	Type	Description	Default
node_request	NodeRequestDTO	Dados do Node a ser criado.	required

Returns:

Type	Description
NodeResponseDTO | None	Node criado ou None caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def create_node(self, node_request: NodeRequestDTO) -> Optional[NodeResponseDTO]:
    """
    Cria um Node.

    :param node_request: Dados do Node a ser criado.
    :type node_request: NodeRequestDTO
    :return: Node criado ou None caso a operação falhe.
    :rtype: NodeResponseDTO | None
    """
    try:
        return self._nodes_api.create_node(node_request)
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível registrar o Node: %s",
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

delete_auth_node(auth_node_id)

Remove um AuthNode.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required

Returns:

Type	Description
bool	True quando o AuthNode é removido; False caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def delete_auth_node(self, auth_node_id: NodeId) -> bool:
    """
    Remove um AuthNode.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :return: True quando o AuthNode é removido; False caso a operação falhe.
    :rtype: bool
    """
    try:
        auth_node_uuid = self._as_uuid(auth_node_id)
        self._auth_nodes_api.delete_auth_node(auth_node_uuid)
        logger.info("AuthNode %s removido com sucesso.", auth_node_uuid)
        return True
    except (ApiException, ValueError) as error:
        logger.error(
            "O AuthNode %s não pode ser removido: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return False

get_auth_node(auth_node_id)

Consulta um AuthNode pelo identificador.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required

Returns:

Type	Description
AuthNodeResponseDTO | None	AuthNode encontrado ou None caso não exista ou a consulta falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def get_auth_node(self, auth_node_id: NodeId) -> Optional[AuthNodeResponseDTO]:
    """
    Consulta um AuthNode pelo identificador.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :return: AuthNode encontrado ou None caso não exista ou a consulta falhe.
    :rtype: AuthNodeResponseDTO | None
    """
    try:
        return self._auth_nodes_api.get_auth_node_by_uuid(self._as_uuid(auth_node_id))
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível consultar AuthNode %s: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

get_auth_node_file_public_key(node_name, usuario, output_path=None)

Baixa a chave pública de um AuthNode.

Parameters:

Name	Type	Description	Default
node_name	str	Nome do Node associado ao AuthNode.	required
usuario	str	Nome do usuário remoto associado ao AuthNode.	required
output_path	Optional[str]	Diretório onde a chave pública será salva.	None

Returns:

Type	Description
str | None	Caminho da chave pública salva ou None caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def get_auth_node_file_public_key(
    self,
    node_name: str,
    usuario: str,
    output_path: Optional[str] = None,
) -> Optional[str]:
    """Baixa a chave pública de um AuthNode.

    :param node_name: Nome do Node associado ao AuthNode.
    :type node_name: str
    :param usuario: Nome do usuário remoto associado ao AuthNode.
    :type usuario: str
    :param output_path: Diretório onde a chave pública será salva.
    :type output_path: Optional[str]
    :return: Caminho da chave pública salva ou None caso a operação falhe.
    :rtype: str | None
    """
    try:
        node = self.get_node_from_name(node_name)
        if node is None or node.id is None:
            return None

        auth_node = self.get_auth_node_from_user(node.id, usuario)
        if auth_node is None or auth_node.id is None:
            return None

        public_key = self.get_auth_node_public_key(auth_node.id)
        if not public_key:
            return None
        return self._save_public_key(public_key, usuario, output_path)
    except (ApiException, OSError, ValueError) as error:
        logger.error(
            "Não foi possível baixar a chave pública de %s no Node %s: %s",
            usuario,
            node_name,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

get_auth_node_from_user(node_id, login)

Consulta um AuthNode pelo Node e pelo usuário remoto.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node.	required
login	str	Nome de usuário remoto associado ao AuthNode.	required

Returns:

Type	Description
AuthNodeResponseDTO | None	AuthNode encontrado ou None caso não exista.

Source code in pldpro_sdk/core/proxies/nodes.py

def get_auth_node_from_user(
    self,
    node_id: NodeId,
    login: str,
) -> Optional[AuthNodeResponseDTO]:
    """
    Consulta um AuthNode pelo Node e pelo usuário remoto.

    :param node_id: Identificador único do Node.
    :type node_id: UUID ou str
    :param login: Nome de usuário remoto associado ao AuthNode.
    :type login: str
    :return: AuthNode encontrado ou None caso não exista.
    :rtype: AuthNodeResponseDTO | None
    """
    normalized_login = login.strip().casefold()
    for auth_node in self.list_auth_nodes(node_id):
        configured_login = (
            auth_node.auth_node_pass.login
            if auth_node.auth_node_pass is not None
            else None
        )
        if configured_login and configured_login.casefold() == normalized_login:
            return auth_node

    logger.warning("AuthNode não encontrado para login %s no Node %s.", login, node_id)
    return None

get_auth_node_public_key(auth_node_id)

Obtém a chave pública SSH de um AuthNode.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required

Returns:

Type	Description
str	Conteúdo da chave pública ou uma string vazia caso a consulta falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def get_auth_node_public_key(self, auth_node_id: NodeId) -> str:
    """Obtém a chave pública SSH de um AuthNode.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :return: Conteúdo da chave pública ou uma string vazia caso a consulta falhe.
    :rtype: str
    """
    try:
        response = self._auth_nodes_api.get_managed_ssh_auth_node_public_key(
            self._as_uuid(auth_node_id)
        )
        return response.public_key or ""
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível obter chave pública do AuthNode %s: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return ""

get_node(node_id)

Consulta um Node pelo identificador.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node.	required

Returns:

Type	Description
NodeResponseDTO | None	Node encontrado ou None caso não exista ou a consulta falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def get_node(self, node_id: NodeId) -> Optional[NodeResponseDTO]:
    """
    Consulta um Node pelo identificador.

    :param node_id: Identificador único do Node.
    :type node_id: UUID ou str
    :return: Node encontrado ou None caso não exista ou a consulta falhe.
    :rtype: NodeResponseDTO | None
    """
    try:
        return self._nodes_api.get_node_by_uuid(self._as_uuid(node_id))
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível consultar o Node %s: %s",
            node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

get_node_from_name(node_name)

Consulta um Node pelo nome.

Parameters:

Name	Type	Description	Default
node_name	str	Nome do Node procurado.	required

Returns:

Type	Description
NodeResponseDTO | None	Node encontrado ou None caso não exista.

Source code in pldpro_sdk/core/proxies/nodes.py

def get_node_from_name(self, node_name: str) -> Optional[NodeResponseDTO]:
    """
    Consulta um Node pelo nome.

    :param node_name: Nome do Node procurado.
    :type node_name: str
    :return: Node encontrado ou None caso não exista.
    :rtype: NodeResponseDTO | None
    """
    normalized_name = node_name.strip().casefold()
    for node in self.list_nodes():
        if node.name and node.name.casefold() == normalized_name:
            return node

    logger.warning("Node não cadastrado ou não autorizado: %s", node_name)
    return None

list_auth_nodes(node_id)

Lista os AuthNodes associados a um Node.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node.	required

Returns:

Type	Description
List[AuthNodeResponseDTO]	AuthNodes associados ao Node.

Source code in pldpro_sdk/core/proxies/nodes.py

def list_auth_nodes(self, node_id: NodeId) -> List[AuthNodeResponseDTO]:
    """
    Lista os AuthNodes associados a um Node.

    :param node_id: Identificador único do Node.
    :type node_id: UUID ou str
    :return: AuthNodes associados ao Node.
    :rtype: List[AuthNodeResponseDTO]
    """
    try:
        node_uuid = self._as_uuid(node_id)
        auth_nodes = self._auth_nodes_api.list_auth_nodes() or []
        return [auth_node for auth_node in auth_nodes if auth_node.node_id == node_uuid]
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível listar AuthNodes do Node %s: %s",
            node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return []

list_node_casos(node_id)

Lista os casos associados a um Node.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node.	required

Returns:

Type	Description
List[CaseResponse]	Casos associados ao Node.

Source code in pldpro_sdk/core/proxies/nodes.py

def list_node_casos(self, node_id: NodeId) -> List[CaseResponse]:
    """
    Lista os casos associados a um Node.

    :param node_id: Identificador único do Node.
    :type node_id: UUID ou str
    :return: Casos associados ao Node.
    :rtype: List[CaseResponse]
    """
    try:
        return list(self._nodes_api.get_casos_node(self._as_uuid(node_id)) or [])
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível listar casos do Node %s: %s",
            node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return []

list_nodes()

Lista os Nodes disponíveis para o usuário autenticado.

Returns:

Type	Description
List[NodeResponseDTO]	Nodes encontrados.

Source code in pldpro_sdk/core/proxies/nodes.py

def list_nodes(self) -> List[NodeResponseDTO]:
    """
    Lista os Nodes disponíveis para o usuário autenticado.

    :return: Nodes encontrados.
    :rtype: List[NodeResponseDTO]
    """
    try:
        return list(self._nodes_api.list_nodes() or [])
    except ApiException as error:
        logger.error("Não foi possível listar Nodes: %s", get_error_message(error))
        return []

list_versions(auth_node_id)

Lista as versões disponíveis em um AuthNode.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required

Returns:

Type	Description
List[str]	Versões disponíveis.

Source code in pldpro_sdk/core/proxies/nodes.py

def list_versions(self, auth_node_id: NodeId) -> List[str]:
    """
    Lista as versões disponíveis em um AuthNode.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :return: Versões disponíveis.
    :rtype: List[str]
    """
    try:
        return list(
            self._auth_nodes_api.get_versoes_container(self._as_uuid(auth_node_id))
            or []
        )
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível listar versões do AuthNode %s: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return []

provision_managed_ssh_auth_node_public_key(auth_node_id)

Instala no host a chave pública gerenciada de um AuthNode.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required

Returns:

Type	Description
bool	True quando a chave foi instalada; False caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def provision_managed_ssh_auth_node_public_key(
    self,
    auth_node_id: NodeId,
) -> bool:
    """Instala no host a chave pública gerenciada de um AuthNode.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :return: True quando a chave foi instalada; False caso a operação falhe.
    :rtype: bool
    """
    try:
        self._auth_nodes_api.provision_managed_ssh_auth_node_public_key(
            self._as_uuid(auth_node_id)
        )
        return True
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível instalar a chave pública do AuthNode %s: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return False

register(registration)

Registra um Node ou um Node com AuthNode.

Parameters:

Name	Type	Description	Default
registration	FullNodeRegistrationDTO | NodeRequestDTO | NodeRequest	Dados do cadastro do Node. Pode incluir as credenciais remotas do AuthNode.	required

Returns:

Type	Description
NodeResponseDTO | None	Node registrado ou None caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def register(
    self,
    registration: FullNodeRegistrationDTO | NodeRequestDTO | NodeRequest,
) -> Optional[NodeResponseDTO]:
    """
    Registra um Node ou um Node com AuthNode.

    :param registration: Dados do cadastro do Node. Pode incluir as credenciais remotas do AuthNode.
    :type registration: FullNodeRegistrationDTO, NodeRequestDTO ou NodeRequest (Evitar usar. Deverá ser descontinuado
    quando evoluirmos o sdk para a próxima major)
    :return: Node registrado ou None caso a operação falhe.
    :rtype: NodeResponseDTO | None
    """
    try:
        if isinstance(registration, FullNodeRegistrationDTO):
            return self._nodes_api.register_full_node(registration)

        if isinstance(registration, NodeRequest):
            registration = NodeRequestDTO(
                name=registration.nome,
                ip=registration.ip,
                dns=registration.dns,
                port=registration.porta,
                lockToServer=registration.lock_to_server,
            )

        if isinstance(registration, NodeRequestDTO):
            return self._nodes_api.create_node(registration)

        raise TypeError("Use FullNodeRegistrationDTO, NodeRequestDTO ou NodeRequest.")
    except (ApiException, TypeError, ValueError) as error:
        logger.error(
            "Não foi possível registrar o Node: %s",
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

rotate_managed_ssh_auth_node_key_pair(auth_node_id, password)

Rotaciona o par de chaves SSH gerenciado de um AuthNode.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required
password	str	Senha usada para instalar a nova chave pública no host remoto.	required

Returns:

Type	Description
str	Nova chave pública ou uma string vazia caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def rotate_managed_ssh_auth_node_key_pair(
    self,
    auth_node_id: NodeId,
    password: str,
) -> str:
    """Rotaciona o par de chaves SSH gerenciado de um AuthNode.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :param password: Senha usada para instalar a nova chave pública no host remoto.
    :type password: str
    :return: Nova chave pública ou uma string vazia caso a operação falhe.
    :rtype: str
    """
    try:
        response = self._auth_nodes_api.rotate_managed_ssh_auth_node_key_pair(
            self._as_uuid(auth_node_id),
            AuthNodeSshKeyRotationRequestDTO(
                password=password
            ),
        )
        return response.public_key or ""
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível rotacionar as chaves do AuthNode %s: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return ""

test_node_connection(auth_node_id)

Verifica a conectividade de um AuthNode.

Parameters:

Name	Type	Description	Default
auth_node_id	NodeId	Identificador único do AuthNode.	required

Returns:

Type	Description
HostCheckResult | None	Resultado da verificação ou None caso a consulta falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def test_node_connection(self, auth_node_id: NodeId) -> Optional[HostCheckResult]:
    """Verifica a conectividade de um AuthNode.

    :param auth_node_id: Identificador único do AuthNode.
    :type auth_node_id: UUID ou str
    :return: Resultado da verificação ou None caso a consulta falhe.
    :rtype: HostCheckResult | None
    """
    try:
        return self._auth_nodes_api.check_auth_node_health(self._as_uuid(auth_node_id))
    except (ApiException, ValueError) as error:
        logger.error(
            "Não foi possível verificar o AuthNode %s: %s",
            auth_node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return None

unregister(node_id)

Remove um Node.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Identificador único do Node.	required

Returns:

Type	Description
bool	True quando o Node é removido; False caso a operação falhe.

Source code in pldpro_sdk/core/proxies/nodes.py

def unregister(self, node_id: NodeId) -> bool:
    """
    Remove um Node.

    :param node_id: Identificador único do Node.
    :type node_id: UUID ou str
    :return: True quando o Node é removido; False caso a operação falhe.
    :rtype: bool
    """
    try:
        node_uuid = self._as_uuid(node_id)
        self._nodes_api.delete_node(node_uuid)
        logger.info("Node %s removido com sucesso.", node_uuid)
        return True
    except (ApiException, ValueError) as error:
        logger.error(
            "O Node %s não pode ser removido: %s",
            node_id,
            get_error_message(error) if isinstance(error, ApiException) else error,
        )
        return False