feat: dataservices

datagouv/__init__.py

@@ -3,3 +3,4 @@
 from datagouv.api.organization import Organization  # noqa
 from datagouv.api.resource import Resource  # noqa
 from datagouv.api.topic import Topic  # noqa
+from datagouv.api.api import API  # noqa

datagouv/api/api.py

@@ -0,0 +1,91 @@
+import logging
+import re
+from typing import Iterator, Optional
+
+import niquests
+
+from datagouv.api.client import Client
+from datagouv.api.dataset import Dataset
+from datagouv.utils.base_object import BaseObject, Creator, assert_auth
+from datagouv.utils.retry import simple_connection_retry
+
+
+class API(BaseObject):
+    _attributes = [
+        "access_audiences",
+        "access_type",
+        "badges",
+        "base_api_url",
+        "business_documentation_url",
+        "created_at",
+        "deleted_at",
+        "description",
+        "extras",
+        "last_modified",
+        "machine_documentation_url",
+        "metrics",
+        "organization",
+        "rate_limiting",
+        "tags",
+        "title",
+        "url",
+    ]
+
+    def __init__(
+        self,
+        id: str,
+        fetch: bool = True,
+        _client: Client = Client(),
+        _from_response: dict | None = None,
+    ):
+        BaseObject.__init__(self, id, _client)
+        self.uri = f"{_client.base_url}/api/1/dataservices/{id}/"
+        self.front_url = self.uri.replace("/api/1", "")
+        if fetch or _from_response:
+            self.refresh(_from_response=_from_response)
+
+    def __call__(self, *args, **kwargs):
+        return API(*args, **kwargs)
+
+    # TODO: to avoid code duplication, _update_method could be a class-level attribute
+    def update(self, payload: dict) -> niquests.Response:
+        assert_auth(self._client)
+        if not isinstance(payload, dict):
+            raise TypeError(f"payload should be a dictionary and not {type(payload)}")
+
+        if self._client.verbose:
+            logging.info(f"🔁 Putting {self.uri} with {payload}")
+        r = self._client.session.patch(self.uri, json=payload)
+        r.raise_for_status()
+        self.refresh(_from_response=r.json())
+        return r
+
+    @property
+    def organization_id(self) -> Optional[str]:
+        if self.organization:  # type: ignore
+            return self.organization["id"]  # type: ignore
+
+    @property
+    def associated_datasets(self) -> Iterator[Dataset]:
+        if not re.fullmatch(r"[0-9a-f]{24}", self.id):
+            raise Exception(
+                f"Current API's ID is a slug : {self.id}. Please recreate the object with its ID."
+            )
+        url = f"api/1/datasets/?dataservice={self.id}"
+        response = self._client.get_all_from_api_query(base_query=url, cast_as=Dataset)
+        return response  # type: ignore - we cast as Dataset in the function
+
+
+class APICreator(Creator):
+    @simple_connection_retry
+    def create(self, payload: dict) -> API:
+        assert_auth(self._client)
+        if self._client.verbose:
+            logging.info(f"Creating third-party API '{payload['title']}'")
+        r = self._client.session.post(f"{self._client.base_url}/api/1/dataservices/", json=payload)
+        try:
+            r.raise_for_status()
+        except Exception as e:
+            raise Exception(r.text) from e
+        metadata = r.json()
+        return API(metadata["id"], _client=self._client, _from_response=metadata)

datagouv/api/client.py

@@ -4,7 +4,7 @@
 import niquests
 
 if TYPE_CHECKING:
-    from datagouv import Dataset, Organization, Resource, Topic
+    from datagouv import API, Dataset, Organization, Resource, Topic
 
 PYTHON_USER_AGENT = {"User-Agent": f"datagouv-python/{version('datagouv_client')}"}
 
@@ -70,11 +70,21 @@ def dataset(self, id: str, **kwargs) -> "Dataset":
 
         return Dataset(id, _client=self, **kwargs)
 
+    def api(self, id: str, **kwargs) -> "API":
+        from datagouv.api.api import API
+
+        return API(id, _client=self, **kwargs)
+
     def create_dataset(self, payload: dict) -> "Dataset":
         from datagouv.api.dataset import DatasetCreator
 
         return DatasetCreator(_client=self).create(payload=payload)
 
+    def create_API(self, payload: dict) -> "API":
+        from datagouv.api.api import APICreator
+
+        return APICreator(_client=self).create(payload=payload)
+
     def topic(self, id: str, **kwargs) -> "Topic":
         from datagouv.api.topic import Topic
 
@@ -101,8 +111,8 @@ def get_all_from_api_query(
         next_page: str = "next_page",
         mask: str | None = None,
         _ignore_base_url: bool = False,
-        cast_as: "Dataset|Organization|Resource|Topic|None" = None,
-    ) -> Iterator["Dataset|Organization|Resource|Topic|dict"]:
+        cast_as: type["Dataset | Organization | Resource | Topic"] | None = None,
+    ) -> Iterator["Dataset | Organization | Resource | Topic | dict"]:
         """⚠️ only for paginated endpoints"""
 
         def get_link_next_page(elem: dict, separated_keys: str) -> str | None:
@@ -116,8 +126,8 @@ def get_link_next_page(elem: dict, separated_keys: str) -> str | None:
         def cast_elem(
             elem: dict,
             client: Client,
-            cast_as: "Dataset|Organization|Resource|Topic|None",
-        ) -> "Dataset|Organization|Resource|Topic|dict":
+            cast_as: type["Dataset | Organization | Resource | Topic"] | None,
+        ) -> "Dataset | Organization | Resource | Topic | dict":
             return (
                 elem
                 if cast_as is None

datagouv/api/organization.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Iterator
 
+from datagouv.api.api import API, APICreator
 from datagouv.api.client import Client
 from datagouv.api.dataset import Dataset, DatasetCreator
 from datagouv.utils.base_object import BaseObject, Creator, assert_auth
@@ -67,6 +68,16 @@ def create_dataset(self, payload: dict) -> Dataset:
             payload=payload | {"organization": self.id}
         )
 
+    def create_API(self, payload: dict) -> API:
+        # we don't simply heritate from DatasetCreator to have a different method name
+        for key in ["organization", "owner"]:
+            if payload.get(key):
+                raise ValueError(
+                    f"It is not possible to specify the {key} when creating an API "
+                    "from an organization, it will be attached to it."
+                )
+        return APICreator(_client=self._client).create(payload=payload | {"organization": self.id})
+
 
 class OrganizationCreator(Creator):
     @simple_connection_retry

datagouv/utils/base_object.py

@@ -52,6 +52,8 @@ def refresh(self, _from_response: dict | None = None) -> dict:
     @simple_connection_retry
     def update(self, payload: dict) -> niquests.Response:
         assert_auth(self._client)
+        if type(payload) is not dict:
+            raise TypeError(f"payload should be a dictionary and not {type(payload)}")
         if self._client.verbose:
             logging.info(f"🔁 Putting {self.uri} with {payload}")
         r = self._client.session.put(self.uri, json=payload)

tests/api_metadata.json

@@ -0,0 +1,20 @@
+{
+  "id": "deadbeef1234567890abcdef",
+  "access_audiences": ["api_client"],
+  "access_type": "open",
+  "badges": [],
+  "base_api_url": "https://example.com/api/",
+  "business_documentation_url": "https://example.com/docs",
+  "created_at": "2024-01-15T10:00:00+00:00",
+  "deleted_at": null,
+  "description": "A test dataservice API",
+  "extras": {},
+  "last_modified": "2024-06-01T12:00:00+00:00",
+  "machine_documentation_url": "https://example.com/openapi.json",
+  "metrics": {"views": 42},
+  "organization": {"id": "646b7187b50b2a93b1ae3d45"},
+  "rate_limiting": null,
+  "tags": ["test", "api"],
+  "title": "Test API",
+  "url": "https://example.com/"
+}

tests/conftest.py

@@ -13,6 +13,7 @@
 OWNER_ID = "637b5c6eef50bb3f5a97b24f"
 DATAGOUV_URL = "https://www.data.gouv.fr/"
 TOPIC_ID = "68b6e6dbdac745f47d4ff6e0"
+API_ID = "deadbeef1234567890abcdef"
 
 with open("tests/dataset_metadata.json", "r") as f:
     dataset_metadata = json.load(f)
@@ -32,6 +33,9 @@
 with open("tests/topic_metadata.json", "r") as f:
     topic_metadata = json.load(f)
 
+with open("tests/api_metadata.json", "r") as f:
+    api_metadata = json.load(f)
+
 with open("tests/elements_metadata.json", "r") as f:
     elements_metadata = json.load(f)
 
@@ -108,6 +112,12 @@ def remote_resource_api2_call(niquests_mock):
     yield niquests_mock
 
 
+@pytest.fixture
+def api_api_call(niquests_mock):
+    niquests_mock.get(f"{DATAGOUV_URL}api/1/dataservices/{API_ID}/").respond(json=api_metadata)
+    yield niquests_mock
+
+
 @pytest.fixture
 def organization_api_call(niquests_mock):
     niquests_mock.get(f"{DATAGOUV_URL}api/1/organizations/{ORGANIZATION_ID}/").respond(

tests/test_api.py

@@ -0,0 +1,149 @@
+from unittest.mock import patch
+
+import pytest
+from conftest import (
+    API_ID,
+    DATAGOUV_URL,
+    ORGANIZATION_ID,
+    OWNER_ID,
+    api_metadata,
+    dataset_metadata,
+    organization_metadata,
+)
+
+from datagouv.api.api import API
+from datagouv.api.client import Client
+from datagouv.api.dataset import Dataset
+from datagouv.api.organization import Organization
+from datagouv.utils.base_object import BaseObject
+
+
+def test_api_instance(api_api_call):
+    assert isinstance(Client().api(API_ID), API)
+
+
+def test_api_attributes_and_methods(api_api_call):
+    client = Client()
+    a = client.api(API_ID)
+    with patch("niquests.Session.get") as mock_func:
+        a_from_response = API(api_metadata["id"], _from_response=api_metadata)
+        mock_func.assert_not_called()
+    for attribute in (
+        ["id", "uri", "front_url", "organization_id", "associated_datasets"]
+        + API._attributes
+        + [method for method in dir(BaseObject) if not method.startswith("__")]
+    ):
+        assert attribute in dir(a)
+        assert attribute in dir(a_from_response)
+
+
+def test_api_no_fetch():
+    with patch("niquests.Session.get") as mock_func:
+        a = API(API_ID, fetch=False)
+        mock_func.assert_not_called()
+    assert all(getattr(a, attr, None) is None for attr in API._attributes)
+    assert a.uri
+
+
+def test_authentication_assertion():
+    client = Client()
+    with pytest.raises(PermissionError):
+        client.create_API({"title": "Test API"})
+    a = API(API_ID, _from_response=api_metadata)
+    with pytest.raises(PermissionError):
+        a.delete()
+    with pytest.raises(PermissionError):
+        a.update({})
+    with pytest.raises(PermissionError):
+        a.update_extras({})
+    with pytest.raises(PermissionError):
+        a.delete_extras([])
+
+
+def test_api_update(api_api_call, niquests_mock):
+    updated_metadata = api_metadata.copy()
+    payload = {"title": "Updated API Title", "description": "Updated description"}
+    niquests_mock.patch(f"{DATAGOUV_URL}api/1/dataservices/{API_ID}/").respond(
+        json=updated_metadata | payload,
+        status_code=200,
+    )
+    client = Client(api_key="test-api-key")
+    a = client.api(API_ID)
+    response = a.update(payload)
+    assert response.status_code == 200
+    for attr in payload:
+        assert getattr(a, attr) == payload[attr]
+
+
+def test_api_update_invalid_payload():
+    client = Client(api_key="test-api-key")
+    a = API(API_ID, _client=client, _from_response=api_metadata)
+    with pytest.raises(TypeError):
+        a.update("not a dict")
+
+
+def test_api_create(niquests_mock):
+    niquests_mock.post(f"{DATAGOUV_URL}api/1/dataservices/").respond(
+        json=api_metadata,
+        status_code=201,
+    )
+    client = Client(api_key="test-api-key")
+    created = client.create_API({"title": "New API", "organization": ORGANIZATION_ID})
+    assert isinstance(created, API)
+    for attr in API._attributes:
+        assert getattr(created, attr) == api_metadata[attr]
+
+
+def test_api_delete(api_api_call, niquests_mock):
+    niquests_mock.delete(f"{DATAGOUV_URL}api/1/dataservices/{API_ID}/").respond(status_code=204)
+    client = Client(api_key="test-api-key")
+    a = client.api(API_ID)
+    response = a.delete()
+    assert response.status_code == 204
+
+
+def test_organization_id():
+    a = API(API_ID, _from_response=api_metadata)
+    assert a.organization_id == ORGANIZATION_ID
+
+
+def test_associated_datasets(niquests_mock):
+    niquests_mock.get(f"{DATAGOUV_URL}api/1/datasets/?dataservice={API_ID}").respond(
+        json={"data": [dataset_metadata], "next_page": None}
+    )
+    a = API(API_ID, _from_response=api_metadata)
+    datasets = list(a.associated_datasets)
+    assert len(datasets) == 1
+    assert isinstance(datasets[0], Dataset)
+
+
+def test_associated_datasets_slug():
+    slug = "my-api-slug"
+    a = API(slug, _from_response=api_metadata)
+    with pytest.raises(Exception, match="slug"):
+        list(a.associated_datasets)
+
+
+def test_organization_create_api(niquests_mock):
+    niquests_mock.post(f"{DATAGOUV_URL}api/1/dataservices/").respond(
+        json=api_metadata,
+        status_code=201,
+    )
+    client = Client(api_key="test-api-key")
+    org = Organization(ORGANIZATION_ID, _client=client, _from_response=organization_metadata)
+    created = org.create_API({"title": "New API"})
+    assert isinstance(created, API)
+
+
+def test_organization_create_api_org_override():
+    org = Organization(ORGANIZATION_ID, _from_response=organization_metadata)
+    with pytest.raises(ValueError):
+        org.create_API({"title": "New API", "organization": ORGANIZATION_ID})
+    with pytest.raises(ValueError):
+        org.create_API({"title": "New API", "owner": OWNER_ID})
+
+
+def test_client_api_methods():
+    client = Client()
+    assert hasattr(client, "api")
+    assert hasattr(client, "create_API")