Pular para conteúdo

data

Proxy para operações relacionadas a dados disponíveis na API PLDPro.

Data

Proxy de alto nível para operações relacionadas a:

  • Dados disponíveis na API PLDPro

Parameters:

Name Type Description Default
api_client ApiClient

Cliente de API para comunicação com o backend.

 required
Source code in pldpro_sdk/core/proxies/data.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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
class Data:
    """
    Proxy de alto nível para operações relacionadas a:

    - Dados disponíveis na API PLDPro

    :param api_client: Cliente de API para comunicação com o backend.
    """

    def __init__(self, api_client: ApiClient) -> None:
        self._data_api = pldpro_sdk.api_client.DataApi(api_client)

    def get_available_data_types(self) -> List[str]:
        """
        Recupera a lista do tipo de dados disponíveis na API.

        :return: Lista com os nomes dos tipos de dados, por exemplo [cortes].
        """
        try:
            api_response = self._data_api.get_available_data_types()
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def get_available_data_types_volume(self, uuid_auth_node: UUID) -> List[str]:
        """
        Recupera a lista do tipo de dados disponíveis na API.

        :return: Lista com os nomes dos tipos de dados, por exemplo [cortes].
        """
        try:
            api_response = self._data_api.get_available_types_by_volume(uuid_auth_node)
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def get_available_data_items(self, type: str) -> List[str]:
        """
        Recupera a lista de dados disponíves para um tipo específico.

        :return: Lista de itens disponíveis ou lista vazia em erro.
        """
        try:
            api_response = self._data_api.get_available_data_items(type)
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def get_available_data_items_volume(self, uuid_auth_node: UUID, type: str) -> List[str]:
        """
        Recupera a lista de dados disponíves para um tipo específico.

        :return: Lista de itens disponíveis ou lista vazia em erro.
        """
        try:
            api_response = self._data_api.get_available_data_items_by_volume(uuid_auth_node,type)
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def list_data_strategies(self) -> List[DataStrategyInfo]:
        """
        Recupera a lista do tipo de dados disponíveis na API.

        :return: Lista de DataStrategyInfo ou lista vazia em erro.
        """
        try:
            api_response = self._data_api.list_data_strategies()
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def list_data_strategies_volume(self) -> List[DataStrategyInfo]:
        """
        Recupera a lista do tipo de dados disponíveis na API.

        :return: Lista de DataStrategyInfo ou lista vazia em erro.
        """
        try:
            api_response = self._data_api.list_volume_data_strategies()
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def execute_data_strategy(
        self, data_type: str, item_name: str, caso_uuid: UUID | str, strategy_name: str
    ) -> None:
        """
        Executa o strategy
        """
        if isinstance(caso_uuid, str):
            caso_uuid = UUID(caso_uuid)

        try:
            self._data_api.execute_data_strategy(
                data_type, item_name, caso_uuid, strategy_name
            )
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
                exc_info=True,
            )
            raise

execute_data_strategy(data_type, item_name, caso_uuid, strategy_name)

Executa o strategy

Source code in pldpro_sdk/core/proxies/data.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def execute_data_strategy(
    self, data_type: str, item_name: str, caso_uuid: UUID | str, strategy_name: str
) -> None:
    """
    Executa o strategy
    """
    if isinstance(caso_uuid, str):
        caso_uuid = UUID(caso_uuid)

    try:
        self._data_api.execute_data_strategy(
            data_type, item_name, caso_uuid, strategy_name
        )
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        raise

get_available_data_items(type)

Recupera a lista de dados disponíves para um tipo específico.

Returns:

Type Description
List[str]

Lista de itens disponíveis ou lista vazia em erro.
Source code in pldpro_sdk/core/proxies/data.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_available_data_items(self, type: str) -> List[str]:
    """
    Recupera a lista de dados disponíves para um tipo específico.

    :return: Lista de itens disponíveis ou lista vazia em erro.
    """
    try:
        api_response = self._data_api.get_available_data_items(type)
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        return []

get_available_data_items_volume(uuid_auth_node, type)

Recupera a lista de dados disponíves para um tipo específico.

Returns:

Type Description
List[str]

Lista de itens disponíveis ou lista vazia em erro.
Source code in pldpro_sdk/core/proxies/data.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def get_available_data_items_volume(self, uuid_auth_node: UUID, type: str) -> List[str]:
    """
    Recupera a lista de dados disponíves para um tipo específico.

    :return: Lista de itens disponíveis ou lista vazia em erro.
    """
    try:
        api_response = self._data_api.get_available_data_items_by_volume(uuid_auth_node,type)
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        return []

get_available_data_types()

Recupera a lista do tipo de dados disponíveis na API.

Returns:

Type Description
List[str]

Lista com os nomes dos tipos de dados, por exemplo [cortes].
Source code in pldpro_sdk/core/proxies/data.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def get_available_data_types(self) -> List[str]:
    """
    Recupera a lista do tipo de dados disponíveis na API.

    :return: Lista com os nomes dos tipos de dados, por exemplo [cortes].
    """
    try:
        api_response = self._data_api.get_available_data_types()
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        return []

get_available_data_types_volume(uuid_auth_node)

Recupera a lista do tipo de dados disponíveis na API.

Returns:

Type Description
List[str]

Lista com os nomes dos tipos de dados, por exemplo [cortes].
Source code in pldpro_sdk/core/proxies/data.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def get_available_data_types_volume(self, uuid_auth_node: UUID) -> List[str]:
    """
    Recupera a lista do tipo de dados disponíveis na API.

    :return: Lista com os nomes dos tipos de dados, por exemplo [cortes].
    """
    try:
        api_response = self._data_api.get_available_types_by_volume(uuid_auth_node)
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        return []

list_data_strategies()

Recupera a lista do tipo de dados disponíveis na API.

Returns:

Type Description
List[DataStrategyInfo]

Lista de DataStrategyInfo ou lista vazia em erro.
Source code in pldpro_sdk/core/proxies/data.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def list_data_strategies(self) -> List[DataStrategyInfo]:
    """
    Recupera a lista do tipo de dados disponíveis na API.

    :return: Lista de DataStrategyInfo ou lista vazia em erro.
    """
    try:
        api_response = self._data_api.list_data_strategies()
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        return []

list_data_strategies_volume()

Recupera a lista do tipo de dados disponíveis na API.

Returns:

Type Description
List[DataStrategyInfo]

Lista de DataStrategyInfo ou lista vazia em erro.
Source code in pldpro_sdk/core/proxies/data.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def list_data_strategies_volume(self) -> List[DataStrategyInfo]:
    """
    Recupera a lista do tipo de dados disponíveis na API.

    :return: Lista de DataStrategyInfo ou lista vazia em erro.
    """
    try:
        api_response = self._data_api.list_volume_data_strategies()
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao listar os nodes:\n\t{error_msg}",
            exc_info=True,
        )
        return []