helpers

Helpers para operações comuns ao SDK PLDPro.

config_api_key(client_config, api_key_env, tenant_code)

Configura API key e tenant conforme o contrato público da API.

Source code in pldpro_sdk/core/helpers.py

def config_api_key(
    client_config: Configuration,
    api_key_env: str,
    tenant_code: str,
) -> Configuration:
    """Configura API key e tenant conforme o contrato público da API."""
    if not api_key_env or not tenant_code:
        raise ValueError("API key e código do tenant são obrigatórios para criar o client.")

    if not getattr(client_config, "api_key", None):
        client_config.api_key = {}

    api_key = api_key_env.strip()
    if not api_key:
        raise ValueError("API key informada esta vazia.")

    client_config.access_token = None
    _clear_development_headers(client_config)
    client_config.api_key["ApiKeyAuth"] = api_key
    client_config.api_key["1_tenant"] = tenant_code
    return client_config

config_development_auth(client_config, tenant_code, dev_user=None, dev_user_id=None, dev_roles=None)

Configura headers de desenvolvimento sem enviar Authorization.

Source code in pldpro_sdk/core/helpers.py

def config_development_auth(
    client_config: Configuration,
    tenant_code: str,
    dev_user: str | None = None,
    dev_user_id: str | None = None,
    dev_roles: str | None = None,
) -> Configuration:
    """Configura headers de desenvolvimento sem enviar Authorization."""
    if not tenant_code:
        raise ValueError("Código do tenant é obrigatório para criar o client.")

    if not getattr(client_config, "api_key", None):
        client_config.api_key = {}

    client_config.access_token = None
    client_config.api_key.pop("ApiKeyAuth", None)
    client_config.api_key.pop("TenantIdAuth", None)
    client_config.api_key["1_tenant"] = tenant_code
    _configure_development_headers(
        client_config,
        dev_user=dev_user,
        dev_user_id=dev_user_id,
        dev_roles=dev_roles,
    )
    return client_config

download(method, caso_id, filename, save_path='', specific_file=False, as_zip=False)

Realiza o download de arquivos de entrada de um caso.

Parameters:

Name	Type	Description	Default
method	Callable	Função de download a ser chamada.	required
caso_id	Union[UUID, str]	Identificador único do caso.	required
filename	Union[Path, str]	Nome do arquivo a ser baixado.	required
save_path	Union[Path, str]	Caminho onde o arquivo será salvo.	''
specific_file	bool	Indica se o download é de um arquivo específico.	False
as_zip	bool	Indica se o arquivo deve ser baixado como zip.	False

Returns:

Type	Description
str	Caminho do arquivo salvo ou string vazia em caso de erro.

Source code in pldpro_sdk/core/helpers.py

def download(
    method: Callable,
    caso_id: Union[UUID, str],
    filename: Union[Path, str],
    save_path: Union[Path, str] = "",
    specific_file: bool = False,
    as_zip: bool = False,
) -> str:
    """
    Realiza o download de arquivos de entrada de um caso.

    :param method: Função de download a ser chamada.
    :type method: Callable
    :param caso_id: Identificador único do caso.
    :type caso_id: Union[UUID, str]
    :param filename: Nome do arquivo a ser baixado.
    :type filename: Union[Path, str]
    :param save_path: Caminho onde o arquivo será salvo.
    :type save_path: Union[Path, str]
    :param specific_file: Indica se o download é de um arquivo específico.
    :type specific_file: bool
    :param as_zip: Indica se o arquivo deve ser baixado como zip.
    :type as_zip: bool
    :return: Caminho do arquivo salvo ou string vazia em caso de erro.
    :rtype: str
    """
    try:
        case_uuid = UUID(str(caso_id))
        output_name = Path(filename)
        destination = Path(save_path) if save_path else Path.cwd()
        destination.mkdir(parents=True, exist_ok=True)

        response = method(case_uuid, output_name.name) if specific_file else method(case_uuid)
        content = response.data if hasattr(response, "data") else response.read() if hasattr(response, "read") else response
        if isinstance(content, str):
            content = content.encode("utf-8")

        target = destination / output_name
        if as_zip:
            target = target.with_suffix(".zip")
        target.write_bytes(content)
        return str(target)
    except (ApiException, ValueError, FileNotFoundError, OSError) as error:
        error_msg = get_error_message(error) if isinstance(error, ApiException) else str(error)
        logger.exception(
            "Ocorreu o seguinte erro ao baixar arquivo(s) do caso %s:\n\t%s",
            caso_id,
            error_msg,
        )
        return ""

get_error_message(api_exception)

Extrai a mensagem mais relevante da resposta de erro da API.

Source code in pldpro_sdk/core/helpers.py

def get_error_message(api_exception: ApiException) -> str:
    """Extrai a mensagem mais relevante da resposta de erro da API."""
    body = getattr(api_exception, "body", None)
    if isinstance(body, bytes):
        body = body.decode("utf-8", errors="replace")

    if isinstance(body, str):
        try:
            body = json.loads(body)
        except json.JSONDecodeError:
            return body

    if isinstance(body, dict):
        return str(
            body.get("details")
            or body.get("message")
            or body.get("menssage")
            or body.get("title")
            or body
        )

    return str(body or api_exception)

get_logger(name='pldpro_sdk')

Retorna uma instância de logger configurada para o SDK. Essa função cria loggers padronizados, reutilizáveis e idempotentes.

Source code in pldpro_sdk/core/helpers.py

def get_logger(name: str = "pldpro_sdk") -> logging.Logger:
    """
    Retorna uma instância de logger configurada para o SDK.
    Essa função cria loggers padronizados, reutilizáveis e idempotentes.
    """
    logger = logging.getLogger(name)

    if not logger.handlers:
        logger.setLevel(logging.INFO)

        formatter = logging.Formatter(
            fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        )

        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(formatter)

        logger.addHandler(console_handler)

        logger.propagate = False

    return logger

upload(method, error_context, caso_id, files, should_zip=True, save_zip=False, use_full_path=True, *args, **kwargs)

Realiza o upload de arquivos ou diretórios para um caso.

Parameters:

Name	Type	Description	Default
method	Callable[..., UploadResponse]	Função de upload a ser chamada.	required
error_context	str	Contexto do erro para mensagens de log.	required
caso_id	Union[UUID, str]	Identificador único do caso.	required
files	Union[List[Path], List[str], Path, str]	Arquivo(s) ou diretório(s) a serem enviados.	required
should_zip	bool	Indica se os arquivos devem ser compactados em um zip antes do upload.	True
save_zip	bool	Indica se o arquivo zip deve ser salvo em disco.	False
use_full_path	bool	Indica se o caminho completo deve ser usado no upload.	True

Returns:

Type	Description
Union[FileUploadResponse, None]	Resposta do upload.

Source code in pldpro_sdk/core/helpers.py

def upload(
    method: Callable[..., UploadResponse],
    error_context: str,
    caso_id: Union[UUID, str],
    files: Union[List[Path], List[str], Path, str],
    should_zip: bool = True,
    save_zip: bool = False,
    use_full_path: bool = True,
    *args,
    **kwargs,
) -> UploadResponse | None:
    """
    Realiza o upload de arquivos ou diretórios para um caso.

    :param method: Função de upload a ser chamada.
    :type method: Callable
    :param error_context: Contexto do erro para mensagens de log.
    :type error_context: str
    :param caso_id: Identificador único do caso.
    :type caso_id: Union[UUID, str]
    :param files: Arquivo(s) ou diretório(s) a serem enviados.
    :type files: Union[List[Path], List[str], Path, str]
    :param should_zip: Indica se os arquivos devem ser compactados em um zip antes do upload.
    :type should_zip: bool
    :param save_zip: Indica se o arquivo zip deve ser salvo em disco.
    :type save_zip: bool
    :param use_full_path: Indica se o caminho completo deve ser usado no upload.
    :type use_full_path: bool
    :return: Resposta do upload.
    :rtype: Union[FileUploadResponse, None]
    """

    def zip_directory_to_memory(dir_path: Path) -> bytes:
        """Compacta um diretório em um arquivo zip na memória e retorna os bytes."""
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
            for root, dirs, files in os.walk(dir_path):
                for file in files:
                    file_path = os.path.join(root, file)
                    rel_path = os.path.relpath(file_path, start=dir_path)
                    zip_file.write(file_path, rel_path)
        zip_buffer.seek(0)
        return zip_buffer.read()

    def zip_list_of_files_to_memory(file_paths: List[Path]) -> bytes:
        """Compacta uma lista de arquivos em um arquivo zip na memória e retorna os bytes."""
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
            for file_path in file_paths:
                zip_file.write(file_path, file_path.name)
        zip_buffer.seek(0)
        return zip_buffer.read()

    def save_zip_to_disk(zip_bytes: bytes, zip_path: Path) -> None:
        """Salva os bytes do arquivo zip em disco."""
        with open(zip_path, "wb") as f:
            f.write(zip_bytes)

    def list_of_files_to_upload(file_paths: List[Path]) -> List[tuple]:
        """Prepara uma lista de tuplas (nome_arquivo, bytes_arquivo) para upload."""
        upload_files = []
        for file_path in file_paths:
            with open(file_path, "rb") as f:
                file_data = f.read()
            upload_files.append((file_path.name, file_data))
        return upload_files

    try:
        if isinstance(caso_id, str):
            caso_id = UUID(caso_id)

        # Se for uma lista de arquivos
        if isinstance(files, list):
            if all(isinstance(f, str) for f in files):
                file_paths = [Path(f) for f in files]
            elif all(isinstance(f, Path) for f in files):
                file_paths: List[Path] = files  # type: ignore
            else:
                raise ValueError(
                    "A lista de arquivos deve conter apenas strings ou objetos Path."
                )

            if should_zip:
                zip_data = zip_list_of_files_to_memory(file_paths)
                # Salvando o arquivo zip em disco
                if save_zip:
                    first_file_path = file_paths[0]
                    zip_disk_path = first_file_path.parent / "upload_files.zip"
                    save_zip_to_disk(zip_data, zip_disk_path)
                ret = method(caso_id, [("upload_files.zip", zip_data)], *args, **kwargs)
            else:
                # NOTE: Alguns métodos aceitam múltiplos arquivos sem zip
                upload_files = list_of_files_to_upload(file_paths)
                logger.debug("Uploading files: %s", upload_files)
                ret = method(caso_id, upload_files, *args, **kwargs)
            return ret

        # Se for um único Path ou string
        elif isinstance(files, (Path, str)):
            file_path = Path(files)

            if file_path.is_dir():
                file_data = zip_directory_to_memory(file_path)
                filename = f"{file_path.name}.zip"

                # FIXME: Verificar se é necessário validar diretório!
                # if not validate_filename_for_method(filename, method.__name__):
                #     raise ValueError(f"O diretório {file_path} não é válido para o método {method.__name__}.")

                # Salvando o arquivo zip em disco
                if save_zip:
                    save_zip_to_disk(file_data, file_path.with_suffix(".zip"))
                ret = method(caso_id, [(filename, file_data)], *args, **kwargs)
                return ret

            elif file_path.is_file():
                # Validar nome do arquivo
                if not validate_filename_for_method(file_path.name, method.__name__):
                    raise ValueError(
                        f"O arquivo {file_path.name} não é válido para o método {method.__name__}."
                    )

                # Para os endpoints que aceitam upload direto sem zip
                if not should_zip:
                    with open(file_path, "rb") as f:
                        file_data = f.read()

                # Se já for um .zip, envia direto
                elif file_path.suffix.lower() == ".zip":
                    with open(file_path, "rb") as f:
                        file_data = f.read()

                else:
                    # Qualquer outro arquivo, compacta antes de enviar
                    file_data = zip_list_of_files_to_memory([file_path])

                    # salva zip localmente se for necessário
                    if save_zip:
                        zip_disk_path = file_path.with_suffix(".zip")
                        save_zip_to_disk(file_data, zip_disk_path)

                # Executa o método de upload
                upload_target = str(file_path) if use_full_path else file_path.name
                ret = method(caso_id, (upload_target, file_data), *args, **kwargs)
                return ret

            else:
                raise ValueError(
                    f"{file_path} não é um arquivo nem um diretório válido."
                )

    except (ApiException, ValueError, OSError) as e:
        error_msg = get_error_message(e) if isinstance(e, ApiException) else str(e)
        logger.exception(
            f"Ocorreu o seguinte erro durante o upload {error_context}:\n\t {error_msg}",
            exc_info=True,
        )
    return None

upload_license(method, error_context, files, should_zip=True, save_zip=False, use_full_path=True, *args, **kwargs)

Realiza o upload de arquivos ou diretórios para endpoints sem caso_id.

Parameters:

Name	Type	Description	Default
method	Callable[..., UploadResponse]	Função de upload a ser chamada.	required
error_context	str	Contexto do erro para mensagens de log.	required
files	Union[List[Path], List[str], Path, str]	Arquivo(s) ou diretório(s) a serem enviados.	required
should_zip	bool	Indica se os arquivos devem ser compactados em um zip antes do upload.	True
save_zip	bool	Indica se o arquivo zip deve ser salvo em disco.	False
use_full_path	bool	Indica se o caminho completo deve ser usado no upload.	True

Returns:

Type	Description
Union[FileUploadResponse, None]	Resposta do upload.

Source code in pldpro_sdk/core/helpers.py

def upload_license(
    method: Callable[..., UploadResponse],
    error_context: str,
    files: Union[List[Path], List[str], Path, str],
    should_zip: bool = True,
    save_zip: bool = False,
    use_full_path: bool = True,
    *args,
    **kwargs,
) -> UploadResponse | None:
    """
    Realiza o upload de arquivos ou diretórios para endpoints sem `caso_id`.

    :param method: Função de upload a ser chamada.
    :type method: Callable
    :param error_context: Contexto do erro para mensagens de log.
    :type error_context: str
    :param files: Arquivo(s) ou diretório(s) a serem enviados.
    :type files: Union[List[Path], List[str], Path, str]
    :param should_zip: Indica se os arquivos devem ser compactados em um zip antes do upload.
    :type should_zip: bool
    :param save_zip: Indica se o arquivo zip deve ser salvo em disco.
    :type save_zip: bool
    :param use_full_path: Indica se o caminho completo deve ser usado no upload.
    :type use_full_path: bool
    :return: Resposta do upload.
    :rtype: Union[FileUploadResponse, None]
    """
    return _execute_upload(
        method=method,
        error_context=error_context,
        files=files,
        should_zip=should_zip,
        save_zip=save_zip,
        use_full_path=use_full_path,
        upload_callback=lambda payload, *callback_args, **callback_kwargs: method(
            payload, *callback_args, **callback_kwargs
        ),
        *args,
        **kwargs,
    )

validate_filename_for_method(filename, method_name)

Valida se o nome do arquivo é compatível com o método de upload.

Parameters:

Name	Type	Description	Default
filename	str	Nome do arquivo a ser validado.	required
method_name	str	Nome do método de upload.	required

Returns:

Type	Description
bool	True se o nome do arquivo for válido para o método, False caso contrário.

Source code in pldpro_sdk/core/helpers.py

def validate_filename_for_method(filename: str, method_name: str) -> bool:
    """
    Valida se o nome do arquivo é compatível com o método de upload.

    :param filename: Nome do arquivo a ser validado.
    :type filename: str
    :param method_name: Nome do método de upload.
    :type method_name: str
    :return: True se o nome do arquivo for válido para o método, False caso contrário.
    :rtype: bool
    """
    method_config = methods_config.get(method_name)

    if not method_config:
        return True  # Sem configuração específica, aceita qualquer arquivo

    valid_filenames = method_config.get("file_names", [])
    if filename in valid_filenames:
        return True

    valid_file_patterns = method_config.get("file_patterns", [])
    for pattern in valid_file_patterns:
        if re.fullmatch(pattern, filename):
            return True
    return False