Pular para conteúdo

data

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

Data

Operações para consultar dados e estratégias disponíveis.

Source code in pldpro_sdk/core/proxies/data.py 
 10
 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
145
146
147
148
149
150
151
152
153
154
class Data:
    """
    Operações para consultar dados e estratégias disponíveis.
    """

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

    def get_available_data_types(self) -> List[str]:
        """
        Lista os tipos de dados disponíveis.

        :return: Nomes dos tipos de dados disponíveis.
        :rtype: List[str]
        """
        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 consultar dados:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def get_available_data_types_volume(self, uuid_auth_node: UUID) -> List[str]:
        """
        Lista os tipos de dados de volume disponíveis para um AuthNode.

        :param uuid_auth_node: Identificador único do AuthNode consultado.
        :type uuid_auth_node: UUID
        :return: Nomes dos tipos de dados de volume disponíveis.
        :rtype: List[str]
        """
        try:
            api_response = self._data_api.get_available_types_by_volume(
                auth_node_uuid=uuid_auth_node,
            )
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(
                f"Ocorreu o seguinte erro ao consultar dados:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def get_available_data_items(self, type: str) -> List[str]:
        """
        Lista os itens disponíveis para um tipo de dado.

        :param type: Tipo de dado consultado.
        :type type: str
        :return: Itens disponíveis para o tipo informado.
        :rtype: List[str]
        """
        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 consultar dados:\n\t{error_msg}",
                exc_info=True,
            )
            return []

    def get_available_data_items_volume( self, uuid_auth_node: UUID, type: str) -> List[str]:
        """
        Lista os itens de volume disponíveis para um AuthNode e tipo de dado.

        :param uuid_auth_node: Identificador único do AuthNode consultado.
        :type uuid_auth_node: UUID
        :param type: Tipo de dado de volume consultado.
        :type type: str
        :return: Itens de volume disponíveis para os filtros informados.
        :rtype: List[str]
        """
        try:
            api_response = self._data_api.get_available_data_items_by_volume(data_type=type, auth_node_uuid=uuid_auth_node)
            return api_response
        except ApiException as e:
            error_msg = get_error_message(e)
            logger.error(f"Ocorreu o seguinte erro ao consultar dados:\n\t{error_msg}", exc_info=True,)
            return []

    def list_data_strategies(self) -> List[DataStrategyInfo]:
        """
        Lista as estratégias disponíveis para dados.

        :return: Estratégias de dados disponíveis.
        :rtype: List[DataStrategyInfo]
        """
        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 consultar dados:\n\t{error_msg}",exc_info=True,)
            return []

    def list_data_strategies_volume(self) -> List[DataStrategyInfo]:
        """
        Lista as estratégias disponíveis para dados de volume.

        :return: Estratégias de dados de volume disponíveis.
        :rtype: List[DataStrategyInfo]
        """
        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 consultar dados:\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 uma estratégia de dados para um caso.

        :param data_type: Tipo de dado que será processado.
        :type data_type: str
        :param item_name: Nome do item de dado selecionado.
        :type item_name: str
        :param caso_uuid: Identificador único do caso.
        :type caso_uuid: UUID ou str
        :param strategy_name: Nome da estratégia a ser executada.
        :type strategy_name: str
        :return: None.
        :rtype: None
        """
        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 consultar dados:\n\t{error_msg}",exc_info=True,)
            raise

execute_data_strategy(data_type, item_name, caso_uuid, strategy_name)

Executa uma estratégia de dados para um caso.

Parameters:

Name Type Description Default
data_type str

Tipo de dado que será processado.

 required
item_name str

Nome do item de dado selecionado.

 required
caso_uuid UUID | str

Identificador único do caso.

 required
strategy_name str

Nome da estratégia a ser executada.

 required

Returns:

Type Description
None

None.
Source code in pldpro_sdk/core/proxies/data.py 
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
def execute_data_strategy(
    self, data_type: str, item_name: str, caso_uuid: UUID | str, strategy_name: str
) -> None:
    """
    Executa uma estratégia de dados para um caso.

    :param data_type: Tipo de dado que será processado.
    :type data_type: str
    :param item_name: Nome do item de dado selecionado.
    :type item_name: str
    :param caso_uuid: Identificador único do caso.
    :type caso_uuid: UUID ou str
    :param strategy_name: Nome da estratégia a ser executada.
    :type strategy_name: str
    :return: None.
    :rtype: None
    """
    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 consultar dados:\n\t{error_msg}",exc_info=True,)
        raise

get_available_data_items(type)

Lista os itens disponíveis para um tipo de dado.

Parameters:

Name Type Description Default
type str

Tipo de dado consultado.

 required

Returns:

Type Description
List[str]

Itens disponíveis para o tipo informado.
Source code in pldpro_sdk/core/proxies/data.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_available_data_items(self, type: str) -> List[str]:
    """
    Lista os itens disponíveis para um tipo de dado.

    :param type: Tipo de dado consultado.
    :type type: str
    :return: Itens disponíveis para o tipo informado.
    :rtype: List[str]
    """
    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 consultar dados:\n\t{error_msg}",
            exc_info=True,
        )
        return []

get_available_data_items_volume(uuid_auth_node, type)

Lista os itens de volume disponíveis para um AuthNode e tipo de dado.

Parameters:

Name Type Description Default
uuid_auth_node UUID

Identificador único do AuthNode consultado.

 required
type str

Tipo de dado de volume consultado.

 required

Returns:

Type Description
List[str]

Itens de volume disponíveis para os filtros informados.
Source code in pldpro_sdk/core/proxies/data.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def get_available_data_items_volume( self, uuid_auth_node: UUID, type: str) -> List[str]:
    """
    Lista os itens de volume disponíveis para um AuthNode e tipo de dado.

    :param uuid_auth_node: Identificador único do AuthNode consultado.
    :type uuid_auth_node: UUID
    :param type: Tipo de dado de volume consultado.
    :type type: str
    :return: Itens de volume disponíveis para os filtros informados.
    :rtype: List[str]
    """
    try:
        api_response = self._data_api.get_available_data_items_by_volume(data_type=type, auth_node_uuid=uuid_auth_node)
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(f"Ocorreu o seguinte erro ao consultar dados:\n\t{error_msg}", exc_info=True,)
        return []

get_available_data_types()

Lista os tipos de dados disponíveis.

Returns:

Type Description
List[str]

Nomes dos tipos de dados disponíveis.
Source code in pldpro_sdk/core/proxies/data.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def get_available_data_types(self) -> List[str]:
    """
    Lista os tipos de dados disponíveis.

    :return: Nomes dos tipos de dados disponíveis.
    :rtype: List[str]
    """
    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 consultar dados:\n\t{error_msg}",
            exc_info=True,
        )
        return []

get_available_data_types_volume(uuid_auth_node)

Lista os tipos de dados de volume disponíveis para um AuthNode.

Parameters:

Name Type Description Default
uuid_auth_node UUID

Identificador único do AuthNode consultado.

 required

Returns:

Type Description
List[str]

Nomes dos tipos de dados de volume disponíveis.
Source code in pldpro_sdk/core/proxies/data.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def get_available_data_types_volume(self, uuid_auth_node: UUID) -> List[str]:
    """
    Lista os tipos de dados de volume disponíveis para um AuthNode.

    :param uuid_auth_node: Identificador único do AuthNode consultado.
    :type uuid_auth_node: UUID
    :return: Nomes dos tipos de dados de volume disponíveis.
    :rtype: List[str]
    """
    try:
        api_response = self._data_api.get_available_types_by_volume(
            auth_node_uuid=uuid_auth_node,
        )
        return api_response
    except ApiException as e:
        error_msg = get_error_message(e)
        logger.error(
            f"Ocorreu o seguinte erro ao consultar dados:\n\t{error_msg}",
            exc_info=True,
        )
        return []

list_data_strategies()

Lista as estratégias disponíveis para dados.

Returns:

Type Description
List[DataStrategyInfo]

Estratégias de dados disponíveis.
Source code in pldpro_sdk/core/proxies/data.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def list_data_strategies(self) -> List[DataStrategyInfo]:
    """
    Lista as estratégias disponíveis para dados.

    :return: Estratégias de dados disponíveis.
    :rtype: List[DataStrategyInfo]
    """
    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 consultar dados:\n\t{error_msg}",exc_info=True,)
        return []

list_data_strategies_volume()

Lista as estratégias disponíveis para dados de volume.

Returns:

Type Description
List[DataStrategyInfo]

Estratégias de dados de volume disponíveis.
Source code in pldpro_sdk/core/proxies/data.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def list_data_strategies_volume(self) -> List[DataStrategyInfo]:
    """
    Lista as estratégias disponíveis para dados de volume.

    :return: Estratégias de dados de volume disponíveis.
    :rtype: List[DataStrategyInfo]
    """
    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 consultar dados:\n\t{error_msg}",exc_info=True, )
        return []