Merge pull request #9 from luizcdc/lista-resultados

Melhora a usabilidade da paginação de resultados

README.md

@@ -58,12 +58,12 @@ for processo in processos:
 from escavador import CriterioOrdenacao, Ordem
 from escavador.v2 import Processo
 
-busca = Processo.por_oab(numero=12345,
-                         estado="SP",
-                         ordena_por=CriterioOrdenacao.INICIO,
-                         ordem=Ordem.DESC)
+advogado, processos = Processo.por_oab(numero=12345,
+                                       estado="SP",
+                                       ordena_por=CriterioOrdenacao.INICIO,
+                                       ordem=Ordem.DESC)
 
-processo = busca.pop()
+processo = processos.pop()
 
 print(f"{processo.numero_cnj}: {processo.titulo_polo_ativo} X {processo.titulo_polo_passivo}")
 ```
@@ -82,7 +82,7 @@ while resultado:
         print(f"{movimentacao.data} - {movimentacao.tipo}:")
         print(f"{movimentacao.conteudo}")
         print()
-    resultado = resultado[0].continuar_busca() # Solicita mais movimentações.
+    resultado = resultado.continuar_busca() # Solicita a próxima página de movimentações
 ```
 
 ### Consultando a última movimentação dos processos mais recentes de uma pessoa pelo nome usando a API V2

escavador/resources/__init__.py

@@ -7,3 +7,4 @@
 from .helpers.enums_v2 import (SiglaTribunal,
                                CriterioOrdenacao,
                                Ordem)
+from .helpers.lista_resultados import ListaResultados

escavador/resources/helpers/consume_cursor.py

@@ -3,6 +3,7 @@
 from typing import Dict, Callable, List
 
 from escavador.method import Method
+from .lista_resultados import ListaResultados
 
 _methods = Method(api_version=2)
 
@@ -17,7 +18,9 @@ def consumir_cursor(cursor: str) -> Dict:
     return _methods.get(endpoint_cursor)
 
 
-def json_to_class(resposta: Dict, constructor: Callable, add_cursor=False) -> List:
+def json_to_class(
+    resposta: Dict, constructor: Callable, add_cursor=False
+) -> ListaResultados:
     """Instancia os itens de uma resposta a partir de um construtor
 
     :param resposta: a resposta da primeira requisição, onde 'items' é uma lista de dicts (jsons)
@@ -27,7 +30,7 @@ def json_to_class(resposta: Dict, constructor: Callable, add_cursor=False) -> Li
     """
     items = resposta["resposta"]["items"]
     cursor_url = resposta["resposta"].get("links", {}).get("next", "")
-    return (
+    return ListaResultados(
         [constructor(item, ultimo_cursor=cursor_url) for item in items]
         if add_cursor
         else [constructor(item) for item in items]

escavador/resources/helpers/lista_resultados.py

@@ -0,0 +1,42 @@
+class CustomListTypeHint(type):
+    def __getitem__(self, item):
+        if isinstance(item, type):
+            item = item.__name__
+        return f"{self.__name__}[{item}]"
+
+
+class ListaResultados(list, metaclass=CustomListTypeHint):
+    def continuar_busca(self) -> "ListaResultados":
+        """Retorna a próxima página de resultados, caso exista."""
+        return self[-1].continuar_busca() if len(self) else ListaResultados()
+
+    def mais_paginas(self, num_paginas: int = 1) -> int:
+        """Extende a lista de resultados com mais resultados, caso existam.
+
+        :param num_paginas: informa quantas páginas adicionais devem ser solicitadas.
+        Se omitido, busca-se apenas a próxima página.
+        :return: Número de páginas adicionais recebidas com sucesso.
+        """
+        return sum(int(self._mais_uma_pagina()) for _ in range(num_paginas))
+
+    def _mais_uma_pagina(self) -> bool:
+        """Extende a lista de resultados com a próxima página, caso exista.
+
+        :return: True se recebeu uma nova página com sucesso, False caso contrário.
+        """
+        if not len(self):
+            return False
+
+        novos_resultados = self[-1].continuar_busca()
+
+        if not novos_resultados:
+            return False
+
+        if len(novos_resultados) > 1 and isinstance(
+            novos_resultados[1], ListaResultados
+        ):
+            novos_resultados = novos_resultados[1]
+
+        self.extend(novos_resultados)
+
+        return True

escavador/v2/resources/envolvido.py

@@ -2,6 +2,7 @@
 
 from typing import Optional, List, Dict, Tuple, Union, TYPE_CHECKING, Type
 
+from escavador.resources import ListaResultados
 from escavador.exceptions import FailedRequest
 from escavador.resources.helpers.enums_v2 import CriterioOrdenacao, Ordem, SiglaTribunal
 from escavador.resources.helpers.endpoint import DataEndpoint
@@ -50,22 +51,31 @@ class EnvolvidoEncontrado:
     tipo_pessoa: str
     quantidade_processos: int
     last_valid_cursor: str = field(default="None", hash=False, compare=False)
-    _classe_buscada: Type["DataEndpoint"] = field(default=None, hash=False, compare=False)
+    _classe_buscada: Type["DataEndpoint"] = field(
+        default=None, hash=False, compare=False
+    )
 
     @classmethod
-    def from_json(cls, json_dict: Optional[Dict], last_cursor: str = "", classe_buscada: Type["DataEndpoint"] = None) -> Optional["EnvolvidoEncontrado"]:
+    def from_json(
+        cls,
+        json_dict: Optional[Dict],
+        last_cursor: str = "",
+        classe_buscada: Type["DataEndpoint"] = None,
+    ) -> Optional["EnvolvidoEncontrado"]:
         if json_dict is None:
             return None
 
         return cls(
             nome=json_dict["nome"],
-            tipo_pessoa=json_dict["tipo_pessoa"],
+            tipo_pessoa=json_dict.get(
+                "tipo_pessoa", "FISICA"
+            ),  # Se não houver tipo_pessoa, assume-se que é advogado.
             quantidade_processos=json_dict["quantidade_processos"],
             last_valid_cursor=last_cursor,
             _classe_buscada=classe_buscada,
         )
 
-    def continuar_busca(self) -> Union[List["DataEndpoint"], FailedRequest]:
+    def continuar_busca(self) -> Union[ListaResultados["DataEndpoint"], FailedRequest]:
         """Retorna mais resultados para a busca que gerou o objeto atual.
 
         :return: lista contendo a próxima página de resultados, ou FailedRequest em caso de erro
@@ -77,8 +87,12 @@ def continuar_busca(self) -> Union[List["DataEndpoint"], FailedRequest]:
                 conteudo = resposta.get("resposta", {})
                 return FailedRequest(status=resposta["http_status"], **conteudo)
 
-            self.last_valid_cursor = resposta["resposta"].get("links", {}).get("next", "")
-            return json_to_class(resposta, self._classe_buscada.from_json, add_cursor=True)
+            self.last_valid_cursor = (
+                resposta["resposta"].get("links", {}).get("next", "")
+            )
+            return json_to_class(
+                resposta, self._classe_buscada.from_json, add_cursor=True
+            )
 
         return []
 

escavador/v2/resources/movimentacao.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass, field
 from typing import Optional, Dict, Union, List, TYPE_CHECKING
 
+from escavador.resources import ListaResultados
 from escavador.exceptions import FailedRequest
 from escavador.v2.resources.tribunal import Tribunal
 from escavador.resources.helpers.consume_cursor import consumir_cursor, json_to_class
@@ -88,7 +89,7 @@ def from_json(
     @staticmethod
     def movimentacoes(
         processo: Union["Processo", str], **kwargs
-    ) -> Union[List["Movimentacao"], FailedRequest]:
+    ) -> Union[ListaResultados["Movimentacao"], FailedRequest]:
         """Busca as movimentações de um processo.
 
         Alias do método `Processo.movimentacoes`.
@@ -106,7 +107,7 @@ def movimentacoes(
             else Processo.movimentacoes(processo, **kwargs)
         )
 
-    def continuar_busca(self) -> Union[List["Movimentacao"], FailedRequest]:
+    def continuar_busca(self) -> Union[ListaResultados["Movimentacao"], FailedRequest]:
         """Retorna mais resultados para a busca que gerou a movimentação atual.
 
         :return: lista de movimentações ou FailedRequest
@@ -122,4 +123,4 @@ def continuar_busca(self) -> Union[List["Movimentacao"], FailedRequest]:
 
             return json_to_class(resposta, self.from_json, add_cursor=True)
 
-        return []
+        return ListaResultados()

escavador/v2/resources/processo.py

@@ -2,6 +2,7 @@
 from dataclasses import dataclass, field
 from typing import Optional, Dict, List, Union, Tuple
 
+from escavador.resources import ListaResultados
 from escavador.exceptions import FailedRequest
 from escavador.resources.helpers.endpoint import DataEndpoint
 from escavador.resources.helpers.enums_v2 import Ordem, CriterioOrdenacao, SiglaTribunal
@@ -102,7 +103,7 @@ def por_numero(numero_cnj: str, **kwargs) -> Union["Processo", FailedRequest]:
     @staticmethod
     def movimentacoes(
         numero_cnj: str, **kwargs
-    ) -> Union[List[Movimentacao], FailedRequest]:
+    ) -> Union[ListaResultados[Movimentacao], FailedRequest]:
         """
         Busca as movimentações de um processo pelo seu número único do CNJ.
 
@@ -132,15 +133,18 @@ def por_nome(
         ordem: Optional[Ordem] = None,
         tribunais: Optional[List[SiglaTribunal]] = None,
         **kwargs,
-    ) -> Union[List["Processo"], FailedRequest]:
+    ) -> Union[
+        Tuple[Optional[EnvolvidoEncontrado], ListaResultados["Processo"]], FailedRequest
+    ]:
         """
         Busca os processos envolvendo uma pessoa ou empresa a partir do seu nome.
 
         :param nome: o nome da pessoa ou empresa
         :param ordena_por: critério de ordenação
         :param ordem: determina ordenação ascendente ou descendente
         :param tribunais: lista de siglas de tribunais para filtrar a busca
-        :return: uma lista de processos, ou FailedRequest caso ocorra algum erro
+        :return: tupla com os dados do envolvido encontrado e uma lista de processos,
+        ou FailedRequest caso ocorra algum erro
 
         >>> Processo.por_nome("Escavador Engenharia e Construcoes Ltda") # doctest: +SKIP
 
@@ -164,15 +168,20 @@ def por_cpf(
         ordem: Optional[Ordem] = None,
         tribunais: Optional[List[SiglaTribunal]] = None,
         **kwargs,
-    ) -> Union[List["Processo"], FailedRequest]:
+    ) -> Union[
+        Tuple[Optional[EnvolvidoEncontrado], ListaResultados["Processo"]], FailedRequest
+    ]:
         """
         Busca os processos envolvendo uma pessoa a partir de seu CPF.
 
+        É possível filtrar por tribunal ou ordenar os resultados.
+
         :param cpf: o CPF da pessoa
         :param ordena_por: critério de ordenação
         :param ordem: determina ordenação ascendente ou descendente
         :param tribunais: lista de siglas de tribunais para filtrar a busca
-        :return: uma lista de processos, ou FailedRequest caso ocorra algum erro
+        :return: tupla com os dados do envolvido encontrado e uma lista de processos,
+        ou FailedRequest caso ocorra algum erro
 
         >>> Processo.por_cpf("123.456.789-99") # doctest: +SKIP
 
@@ -196,15 +205,20 @@ def por_cnpj(
         ordem: Optional[Ordem] = None,
         tribunais: Optional[List[SiglaTribunal]] = None,
         **kwargs,
-    ) -> Union[List["Processo"], FailedRequest]:
+    ) -> Union[
+        Tuple[Optional[EnvolvidoEncontrado], ListaResultados["Processo"]], FailedRequest
+    ]:
         """
         Busca os processos envolvendo uma instituição a partir de seu CNPJ.
 
+        É possível filtrar por tribunal e ordenar os resultados.
+
         :param cnpj: o CNPJ da instituição
         :param ordena_por: critério de ordenação
         :param ordem: determina ordenação ascendente ou descendente
         :param tribunais: lista de siglas de tribunais para filtrar a busca
-        :return: uma lista de processos, ou FailedRequest caso ocorra algum erro
+        :return: tupla com os dados do envolvido encontrado e uma lista de processos,
+        ou FailedRequest caso ocorra algum erro
 
         >>> Processo.por_cnpj("07838351002160") # doctest: +SKIP
 
@@ -229,11 +243,13 @@ def por_envolvido(
         ordem: Optional[Ordem] = None,
         tribunais: Optional[List[SiglaTribunal]] = None,
         **kwargs,
-    ) -> Union[Tuple[Optional[EnvolvidoEncontrado], List["Processo"]], FailedRequest]:
+    ) -> Union[
+        Tuple[Optional[EnvolvidoEncontrado], ListaResultados["Processo"]], FailedRequest
+    ]:
         """
         Busca os processos envolvendo uma pessoa ou instituição a partir de seu nome e/ou CPF/CNPJ.
 
-        É possível filtrar
+        É possível filtrar por tribunal e ordenar os resultados.
 
         :param nome: o nome da pessoa ou instituição. Obrigatório se não for informado o CPF/CNPJ
         :param cpf_cnpj: o CPF/CNPJ da pessoa ou instituição. Obrigatório se não for informado o nome
@@ -286,7 +302,9 @@ def por_oab(
         ordena_por: Optional[CriterioOrdenacao] = None,
         ordem: Optional[Ordem] = None,
         **kwargs,
-    ) -> Union[List["Processo"], FailedRequest]:
+    ) -> Union[
+        Tuple[Optional[EnvolvidoEncontrado], ListaResultados["Processo"]], FailedRequest
+    ]:
         """
         Busca os processos de um advogado a partir de sua carteira da OAB.
 
@@ -320,9 +338,17 @@ def por_oab(
             conteudo = first_response.get("resposta", {})
             return FailedRequest(status=first_response["http_status"], **conteudo)
 
-        return json_to_class(first_response, Processo.from_json, add_cursor=True)
+        advogado_encontrado = EnvolvidoEncontrado.from_json(
+            first_response["resposta"].get("advogado_encontrado"),
+            last_cursor=first_response["resposta"].get("links", {}).get("next", ""),
+            classe_buscada=Processo,
+        )
+
+        return advogado_encontrado, json_to_class(
+            first_response, Processo.from_json, add_cursor=True
+        )
 
-    def continuar_busca(self) -> Union[List["Processo"], FailedRequest]:
+    def continuar_busca(self) -> Union[ListaResultados["Processo"], FailedRequest]:
         """Retorna mais resultados para a busca que gerou o processo atual.
 
         :return: lista de processos ou FailedRequest
@@ -336,7 +362,7 @@ def continuar_busca(self) -> Union[List["Processo"], FailedRequest]:
 
             return json_to_class(resposta, self.from_json, add_cursor=True)
 
-        return []
+        return ListaResultados()
 
 
 @dataclass

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "escavador"
-version = "0.2.2"
+version = "0.3.0"
 description = "A library to  interact with Escavador API"
 authors = [
     "Rafael <rafaelcampos@escavador.com>",