add tinybird api clients (#6)

Author: thrau

README.md

@@ -9,8 +9,7 @@ Verdin
   <a href="https://github.com/psf/black"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a>
 </p>
 
-Verdin is a [tiny bird](https://en.wikipedia.org/wiki/Verdin), and also a [Tinybird](https://tinybird.co) SDK for Python
-.
+Verdin is a [tiny bird](https://en.wikipedia.org/wiki/Verdin), and also a [Tinybird](https://tinybird.co) SDK for Python.
 
 Install
 -------
@@ -58,7 +57,7 @@ response: tinybird.PipeJsonResponse = pipe.query({"key": "val"})
 print(response.data)
 ```
 
-### Append to a DataSource
+### Append to a data source
 
 ```python
 from verdin import tinybird
@@ -75,6 +74,19 @@ datasource.append([
 ])
 ```
 
+### Append to a data source using high-frequency ingest
+
+The `DataSource` object also gives you access to `/v0/events`, which is the high-frequency ingest, to append data.
+Use the `send_events` method and pass JSON serializable documents to it.
+
+```python
+datasource.send_events(records=[
+    {"key": "val1"},
+    {"key": "val2"},
+    ...
+])
+```
+
 ### Queue and batch records into a DataSource
 
 Verdin provides a way to queue and batch data continuously:
@@ -100,6 +112,61 @@ records.put(("col1-row1", "col2-row1"))
 records.put(("col1-row2", "col2-row2"))
 ```
 
+### API access
+
+The DataSource and Pipes objects presented so far are high-level abstractions that provide a convenience Python API
+to deal with the most common use cases. Verdin also provides more low-level access to APIs via `client.api`.
+The following APIs are available:
+
+* `/v0/datasources`: `client.api.datasources`
+* `/v0/events`: `client.api.events`
+* `/v0/pipes`: `client.api.pipes`
+* `/v0/tokens`: `client.api.tokens`
+* `/v0/variables`: `client.api.variables`
+
+Note that for some (datasources, pipes, tokens), manipulation operations are not implemented as they are typically done
+through tb deployments and not through the API.
+
+Also note that API clients do not take care of retries or rate limiting. The caller is expected to handle fault
+tolerance.
+
+#### Example (Querying a pipe)
+
+You can query a pipe through the pipes API as follows:
+
+```python
+from verdin import tinybird
+
+client = tinybird.Client(...)
+
+response = client.api.pipes.query(
+    "my_pipe",
+    parameters={"my_param": "..."},
+    query="SELECT * FROM _ LIMIT 10",
+)
+
+for record in response.data:
+    # each record is a dictionary
+    ...
+```
+
+#### Example (High-frequency ingest)
+
+You can use the HFI endpoint `/v0/events` through the `events` api. As records, you can pass a list of JSON serializable
+documents.
+
+```python
+from verdin import tinybird
+
+client = tinybird.Client(...)
+
+response = client.api.events.send("my_datasource", records=[
+    {"id": "...", "value": "..."},
+    ...
+])
+assert response.quarantined_rows == 0
+```
+
 Develop
 -------
 

tests/integration/conftest.py

@@ -3,10 +3,21 @@
 
 import pytest
 
+from verdin.api import ApiError
 from verdin.client import Client
 from verdin.test.cli import TinybirdCli
 from verdin.test.container import TinybirdLocalContainer
 
+# os.environ["SKIP_TINYBIRD_LOCAL_START"] = "1"
+
+
+def _is_skip_tinybird_local_start() -> bool:
+    """
+    Set SKIP_TINYBIRD_LOCAL_START=1 if you have a tb local container running already with the project deployed. This
+    allows faster iterations.
+    """
+    return os.environ.get("SKIP_TINYBIRD_LOCAL_START") in ["1", "true", "True", True]
+
 
 @pytest.fixture(scope="session")
 def client(tinybird_local_container) -> Client:
@@ -33,17 +44,44 @@ def tinybird_local_container():
 
     container = TinybirdLocalContainer(cwd=project_dir)
 
-    container.start()
+    if not _is_skip_tinybird_local_start():
+        container.start()
+
     container.wait_is_up()
 
     yield container
 
     # cleanup
-    container.stop()
+    if not _is_skip_tinybird_local_start():
+        container.stop()
 
 
 @pytest.fixture(scope="session", autouse=True)
 def deployed_project(cli):
+    if _is_skip_tinybird_local_start():
+        yield
+        return
+
     time.sleep(5)
     cli.deploy(wait=True, auto=True)
     yield
+
+
+@pytest.fixture(autouse=True)
+def _truncate_datasource(client):
+    # make sure to truncate "simple" datasource and its quarantine table before and after each test
+
+    client.api.datasources.truncate("simple")
+    try:
+        # also truncate the quarantine table if it exists
+        client.api.datasources.truncate("simple_quarantine")
+    except ApiError:
+        pass
+
+    yield
+    client.api.datasources.truncate("simple")
+
+    try:
+        client.api.datasources.truncate("simple_quarantine")
+    except ApiError:
+        pass

tests/integration/project/endpoints/simple_pipe.pipe

@@ -0,0 +1,14 @@
+VERSION 0
+
+DESCRIPTION >
+    Endpoint to select specific keys from the table
+
+NODE endpoint
+SQL >
+    %
+    SELECT *
+    FROM simple
+    WHERE 1=1
+        {% if defined(key) %} AND key == {{ String(key) }} {% end %}
+
+TYPE ENDPOINT

tests/integration/test_datasources.py

@@ -0,0 +1,166 @@
+import json
+
+import pytest
+
+from verdin.api import ApiError
+from verdin.api.datasources import DataSourceNotFoundError
+from tests.utils import retry
+
+
+class TestDataSourcesApi:
+    def test_list(self, client):
+        response = client.api.datasources.list()
+
+        assert len(response.datasources) >= 1
+
+        # find "simple" datasource in the list of data sources
+        ds = None
+        for datasource in response.datasources:
+            if datasource["name"] == "simple":
+                ds = datasource
+                break
+
+        assert ds
+
+        # smoke tests some attributes
+        assert ds["engine"]["engine"] == "MergeTree"
+        assert "simple_kv" in [x["name"] for x in ds["used_by"]]
+
+    def test_get_information(self, client):
+        response = client.api.datasources.get_information("simple")
+
+        # smoke tests some attributes
+        assert response.info["name"] == "simple"
+        assert response.info["engine"]["engine"] == "MergeTree"
+        assert "simple_kv" in [x["name"] for x in response.info["used_by"]]
+
+    def test_get_information_on_non_existing_datasource(self, client):
+        with pytest.raises(DataSourceNotFoundError) as e:
+            client.api.datasources.get_information("non_existing_datasource")
+
+        e.match('Data Source "non_existing_datasource" does not exist')
+        assert e.value.status_code == 404
+
+    def test_truncate(self, client):
+        ds = client.datasource("simple")
+        ds.append_ndjson(
+            [
+                {
+                    "Id": "e7f2af3e-99d1-4d4f-8a8c-d6aee4ab89b0",
+                    "Timestamp": "2024-01-23T10:30:00.123456",
+                    "Key": "foo",
+                    "Value": "bar",
+                },
+                {
+                    "Id": "d7792957-21d8-46e6-a4e0-188eb36e2758",
+                    "Timestamp": "2024-02-23T11:45:00.234567",
+                    "Key": "baz",
+                    "Value": "ed",
+                },
+            ]
+        )
+
+        def _wait_for_count(cnt: int):
+            query = client.sql("SELECT count(*) as cnt FROM simple")
+            assert query.json().data == [{"cnt": cnt}]
+
+        retry(_wait_for_count, args=(2,))
+
+        client.api.datasources.truncate("simple")
+
+        retry(_wait_for_count, args=(0,))
+
+    def test_append_to_non_existing_data_source(self, client):
+        with pytest.raises(ApiError) as e:
+            client.api.datasources.append("non_existing_datasource", "foo,bar\n")
+
+        # this is odd behavior, but currently, this raises a 403, with the error
+        # "Adding or modifying data sources to this workspace can only be done via deployments"
+        # due to the way tinybird behaves (apparently it doesn't check mode=append)
+
+        assert e.value.status_code == 403
+
+    def test_append_csv(self, client):
+        ds = client.api.datasources
+
+        data = "5b6859d2-e060-40a4-949a-7e7fab8e3207,2024-01-23T10:30:00.123456,foo,bar\n"
+        data += "af49ffce-559c-426e-9787-ddb08628b547,2024-02-23T11:45:00.234567,baz,ed"
+
+        response = ds.append("simple", data)
+        assert not response.error
+        assert response.quarantine_rows == 0
+        assert response.invalid_lines == 0
+        assert response.datasource["name"] == "simple"
+
+        assert client.sql("SELECT * FROM simple").json().data == [
+            {
+                "id": "5b6859d2-e060-40a4-949a-7e7fab8e3207",
+                "timestamp": "2024-01-23 10:30:00.123456",
+                "key": "foo",
+                "value": "bar",
+            },
+            {
+                "id": "af49ffce-559c-426e-9787-ddb08628b547",
+                "timestamp": "2024-02-23 11:45:00.234567",
+                "key": "baz",
+                "value": "ed",
+            },
+        ]
+
+    def test_append_csv_with_invalid_data(self, client):
+        ds = client.api.datasources
+
+        data = "5b6859d2-e060-40a4-949a-7e7fab8e3207,2024-01-23T10:30:00.123456,foo,bar\n"
+        data += "af49ffce-559c-426e-9787-ddb08628b5472024-02-23T11:45:00.234567,baz,ed"  # error in this line
+
+        response = ds.append("simple", data)
+        assert (
+            response.error
+            == "There was an error with file contents: 1 row in quarantine and 1 invalid line."
+        )
+        assert response.invalid_lines == 1
+        assert response.quarantine_rows == 1
+
+    def test_append_ndjson(self, client):
+        ds = client.api.datasources
+        ds.truncate("simple")
+
+        records = [
+            {
+                "Id": "e7f2af3e-99d1-4d4f-8a8c-d6aee4ab89b0",
+                "Timestamp": "2024-01-23T10:30:00.123456",
+                "Key": "foo",
+                "Value": "bar",
+            },
+            {
+                "Id": "d7792957-21d8-46e6-a4e0-188eb36e2758",
+                "Timestamp": "2024-02-23T11:45:00.234567",
+                "Key": "baz",
+                "Value": "ed",
+            },
+        ]
+
+        def _data():
+            for r in records:
+                yield json.dumps(r) + "\n"
+
+        response = ds.append("simple", _data(), format="ndjson")
+        assert not response.error
+        assert response.quarantine_rows == 0
+        assert response.invalid_lines == 0
+        assert response.datasource["name"] == "simple"
+
+        assert client.sql("SELECT * FROM simple").json().data == [
+            {
+                "id": "e7f2af3e-99d1-4d4f-8a8c-d6aee4ab89b0",
+                "timestamp": "2024-01-23 10:30:00.123456",
+                "key": "foo",
+                "value": "bar",
+            },
+            {
+                "id": "d7792957-21d8-46e6-a4e0-188eb36e2758",
+                "timestamp": "2024-02-23 11:45:00.234567",
+                "key": "baz",
+                "value": "ed",
+            },
+        ]