Use Pydantic in Search

pygleif/api/__init__.py

@@ -1,4 +1,6 @@
 """Pydantic Representation of models."""
+from typing import List
+
 from pydantic import BaseModel, Field
 
 from .data import Data
@@ -10,3 +12,10 @@ class GLEIFResponse(BaseModel):
 
     meta: Meta = Field(alias="meta")
     data: Data = Field(alias="data")
+
+
+class SearchResponse(BaseModel):
+    """Represent search result response."""
+
+    meta: Meta = Field(alias="meta")
+    data: List[Data] = Field(alias="data")

pygleif/api/meta.py

@@ -1,5 +1,6 @@
 """Meta data."""
 from datetime import datetime
+from typing import Optional
 
 from pydantic import BaseModel, Field
 
@@ -10,7 +11,20 @@ class GoldenCopy(BaseModel):
     publish_date: datetime = Field(alias="publishDate")
 
 
+class Pagination(BaseModel):
+    """Represent response pagination."""
+
+    current_page: int = Field(alias="currentPage")
+    per_page: int = Field(alias="perPage")
+    _from: int = Field(alias="from")
+    to: int = Field(alias="to")
+    total: int = Field(alias="total")
+    last_page: int = Field(alias="lastPage")
+
+
 class Meta(BaseModel):
     """Represent meta information."""
 
     golden_copy: GoldenCopy = Field(alias="goldenCopy")
+    # Pagination is part of the search response
+    pagination: Optional[Pagination] = Field(alias="pagination")

pygleif/const.py

@@ -1,16 +1,8 @@
 """Constants."""
 URL_API = "https://api.gleif.org/api/v1/lei-records/"
 URL_SEARCH = "https://api.gleif.org/api/v1/lei-records?filter%5Bfulltext%5D="
-URL_DIRECT_CHILD = (
-    "https://api.gleif.org/api/v1/lei-records/{}/direct-child-relationships"
-)
 
 
-ATTR_REGISTRATION = "registration"
-
-# Allowed attributes for a record
-ALLOW_ATTR_REGISTRATION_STATUS = ["ISSUED", "LAPSED"]
-
 # The ELF codes are unique per country, not globally.
 # Hence the nested dict.
 LEGAL_FORMS = {

pygleif/gleif.py

@@ -1,5 +1,6 @@
 """GLEIF API."""
 from .api import GLEIFResponse
+from .const import URL_API
 from .utils import load_json
 
 
@@ -8,5 +9,5 @@ class PyGleif:
 
     def __init__(self, lei_code: str) -> None:
         """Init class."""
-        json_data = load_json(lei_code)
+        json_data = load_json(search_url=URL_API, search_string=lei_code)
         self.response = GLEIFResponse(**json_data)

pygleif/search.py

@@ -1,45 +1,16 @@
 """Search."""
-import json
-from urllib import request as url
 
-from pygleif.const import ALLOW_ATTR_REGISTRATION_STATUS, URL_SEARCH
+
+from pygleif.api import SearchResponse
+from pygleif.const import URL_SEARCH
+
+from .utils import load_json
 
 
 class Search:
     """Class to use the search form of the GLEIF web site."""
 
-    def __init__(self, orgnr=None):
+    def __init__(self, orgnr: str) -> None:
         """Init class."""
-        # Allow searching using organisation number
-        self.orgnr = orgnr
-
-    @property
-    def json_data(self):
-        """Get raw data from the service."""
-        return url.urlopen(URL_SEARCH + url.quote(self.orgnr))
-
-    @property
-    def raw(self):
-        """Return parsed json."""
-        return json.loads(self.json_data.read().decode("UTF-8"))
-
-    @property
-    def valid_record(self):
-        """Loop through data to find a valid record. Return first valid."""
-        for d in self.raw["data"]:
-
-            # We're not very greedy here, but it seems some records have
-            # lapsed even through the issuer is active
-            if (
-                d["attributes"]["registration"]["status"]
-                in ALLOW_ATTR_REGISTRATION_STATUS
-            ):
-                return d["attributes"]
-
-    @property
-    def lei(self):
-        """Return the LEI code."""
-        try:
-            return self.valid_record["lei"]
-        except (IndexError, TypeError):
-            return None
+        json_data = load_json(search_url=URL_SEARCH, search_string=orgnr)
+        self.response = SearchResponse(**json_data)

pygleif/utils.py

@@ -3,12 +3,10 @@
 
 import json
 from typing import Any, Dict, List, Union, cast
-import urllib.request as url
+from urllib import request
 
-from .const import URL_API
 
-
-def load_json(lei_code: str) -> list[Any] | dict[Any, Any]:
+def load_json(search_url: str, search_string: str) -> list[Any] | dict[Any, Any]:
     """Download data as JSON."""
-    with url.urlopen(f"{URL_API}{lei_code}") as fdesc:
+    with request.urlopen(f"{search_url}{request.quote(search_string)}") as fdesc:
         return cast(Union[Dict[Any, Any], List[Any]], json.loads(fdesc.read()))

tests/test_gleif.py

@@ -1,7 +1,7 @@
 """Tests."""
 import pytest
 
-from pygleif import PyGleif
+from pygleif import PyGleif, Search
 
 
 @pytest.fixture(scope="module")
@@ -10,6 +10,12 @@ def gleif_fixture_1() -> PyGleif:
     return PyGleif("549300MLUDYVRQOOXS22")
 
 
+@pytest.fixture(scope="module")
+def gleif_search_fixture() -> Search:
+    """Fixture."""
+    return Search("917685991")
+
+
 def test_lei(gleif_fixture_1: PyGleif):
     """Test LEI attribute."""
     assert gleif_fixture_1.response.data.attributes.lei, "549300MLUDYVRQOOXS22"
@@ -18,3 +24,13 @@ def test_lei(gleif_fixture_1: PyGleif):
 def test_id(gleif_fixture_1: PyGleif):
     """Test ID attribute."""
     assert gleif_fixture_1.response.data.id, "549300MLUDYVRQOOXS22"
+
+
+def test_search_lei(gleif_search_fixture: Search):
+    """Test LEI attribute."""
+    assert gleif_search_fixture.response.data[0].attributes.lei, "549300MLUDYVRQOOXS22"
+
+
+def test_search_id(gleif_search_fixture: Search):
+    """Test ID attribute."""
+    assert gleif_search_fixture.response.data[0].id, "549300MLUDYVRQOOXS22"