Add Tinybird query runner

README.md

@@ -90,6 +90,7 @@ Redash supports more than 35 SQL and NoSQL [data sources](https://redash.io/help
 - SPARQL
 - SQLite
 - TiDB
+- Tinybird
 - TreasureData
 - Trino
 - Uptycs

redash/query_runner/clickhouse.py

@@ -53,10 +53,6 @@ def configuration_schema(cls):
             "secret": ["password"],
         }
 
-    @classmethod
-    def type(cls):
-        return "clickhouse"
-
     @property
     def _url(self):
         return urlparse(self.configuration["url"])
@@ -162,7 +158,7 @@ def _define_column_type(column):
             return TYPE_STRING
 
     def _clickhouse_query(self, query, session_id=None, session_check=None):
-        logger.debug("Clickhouse is about to execute query: %s", query)
+        logger.debug(f"{self.name()} is about to execute query: %s", query)
 
         query += "\nFORMAT JSON"
 

redash/query_runner/tinybird.py

@@ -0,0 +1,113 @@
+import logging
+
+import requests
+
+from redash.query_runner import register
+from redash.query_runner.clickhouse import ClickHouse
+
+logger = logging.getLogger(__name__)
+
+
+class Tinybird(ClickHouse):
+    noop_query = "SELECT count() FROM tinybird.pipe_stats LIMIT 1"
+
+    DEFAULT_URL = "https://api.tinybird.co"
+
+    SQL_ENDPOINT = "/v0/sql"
+    DATASOURCES_ENDPOINT = "/v0/datasources"
+    PIPES_ENDPOINT = "/v0/pipes"
+
+    @classmethod
+    def configuration_schema(cls):
+        return {
+            "type": "object",
+            "properties": {
+                "url": {"type": "string", "default": cls.DEFAULT_URL},
+                "token": {"type": "string", "title": "Auth Token"},
+                "timeout": {
+                    "type": "number",
+                    "title": "Request Timeout",
+                    "default": 30,
+                },
+                "verify": {
+                    "type": "boolean",
+                    "title": "Verify SSL certificate",
+                    "default": True,
+                },
+            },
+            "order": ["url", "token"],
+            "required": ["token"],
+            "extra_options": ["timeout", "verify"],
+            "secret": ["token"],
+        }
+
+    def _get_tables(self, schema):
+        self._collect_tinybird_schema(
+            schema,
+            self.DATASOURCES_ENDPOINT,
+            "datasources",
+        )
+
+        self._collect_tinybird_schema(
+            schema,
+            self.PIPES_ENDPOINT,
+            "pipes",
+        )
+
+        return list(schema.values())
+
+    def _send_query(self, data, session_id=None, session_check=None):
+        return self._get_from_tinybird(
+            self.SQL_ENDPOINT,
+            params={"q": data.encode("utf-8", "ignore")},
+        )
+
+    def _collect_tinybird_schema(self, schema, endpoint, resource_type):
+        response = self._get_from_tinybird(endpoint)
+        resources = response.get(resource_type, [])
+
+        for r in resources:
+            if r["name"] not in schema:
+                schema[r["name"]] = {"name": r["name"], "columns": []}
+
+            if resource_type == "pipes" and not r.get("endpoint"):
+                continue
+
+            query = f"SELECT * FROM {r['name']} LIMIT 1 FORMAT JSON"
+            try:
+                query_result = self._send_query(query)
+            except Exception:
+                logger.exception(f"error in schema {r['name']}")
+                continue
+
+            columns = [meta["name"] for meta in query_result["meta"]]
+            schema[r["name"]]["columns"].extend(columns)
+
+        return schema
+
+    def _get_from_tinybird(self, endpoint, params=None):
+        url = f"{self.configuration.get('url', self.DEFAULT_URL)}{endpoint}"
+        authorization = f"Bearer {self.configuration.get('token')}"
+
+        try:
+            response = requests.get(
+                url,
+                timeout=self.configuration.get("timeout", 30),
+                params=params,
+                headers={"Authorization": authorization},
+                verify=self.configuration.get("verify", True),
+            )
+        except requests.RequestException as e:
+            if e.response:
+                details = f"({e.__class__.__name__}, Status Code: {e.response.status_code})"
+            else:
+                details = f"({e.__class__.__name__})"
+            raise Exception(f"Connection error to: {url} {details}.")
+
+        if response.status_code >= 400:
+            raise Exception(response.text)
+
+        return response.json()
+
+
+register(Tinybird)

redash/settings/__init__.py

@@ -296,6 +296,7 @@ def email_server_is_configured():
     "redash.query_runner.impala_ds",
     "redash.query_runner.vertica",
     "redash.query_runner.clickhouse",
+    "redash.query_runner.tinybird",
     "redash.query_runner.yandex_metrica",
     "redash.query_runner.rockset",
     "redash.query_runner.treasuredata",

tests/query_runner/test_tinybird.py

@@ -0,0 +1,123 @@
+import json
+from unittest import TestCase
+from unittest.mock import Mock, patch
+
+from redash.query_runner import TYPE_DATETIME, TYPE_INTEGER, TYPE_STRING
+from redash.query_runner.tinybird import Tinybird
+
+DATASOURCES_RESPONSE = {
+    "datasources": [
+        {
+            "id": "t_datasource_id",
+            "name": "test_datasource",
+            "columns": [
+                {
+                    "name": "string_attribute",
+                    "type": "String",
+                    "nullable": False,
+                    "normalized_name": "string_attribute",
+                },
+                {
+                    "name": "number_attribute",
+                    "type": "Int32",
+                    "nullable": False,
+                    "normalized_name": "number_attribute",
+                },
+                {"name": "date_attribute", "type": "DateTime", "nullable": False, "normalized_name": "date_attribute"},
+            ],
+        }
+    ]
+}
+
+PIPES_RESPONSE = {"pipes": [{"id": "t_pipe_id", "name": "test_pipe", "endpoint": "t_endpoint_id", "type": "endpoint"}]}
+
+SCHEMA_RESPONSE = {
+    "meta": [
+        {"name": "string_attribute", "type": "String"},
+        {"name": "number_attribute", "type": "Int32"},
+        {"name": "date_attribute", "type": "DateTime"},
+    ]
+}
+
+QUERY_RESPONSE = {
+    **SCHEMA_RESPONSE,
+    "data": [
+        {"string_attribute": "hello world", "number_attribute": 123, "date_attribute": "2023-01-01 00:00:03.001000"},
+    ],
+    "rows": 1,
+    "statistics": {"elapsed": 0.011556914, "rows_read": 87919, "bytes_read": 17397219},
+}
+
+
+class TestTinybird(TestCase):
+    @patch("requests.get")
+    def test_get_schema_scans_pipes_and_datasources(self, get_request):
+        query_runner = self._build_query_runner()
+
+        get_request.side_effect = self._mock_tinybird_schema_requests
+
+        schema = query_runner.get_schema()
+
+        self.assertEqual(
+            schema,
+            [
+                {"name": "test_datasource", "columns": ["string_attribute", "number_attribute", "date_attribute"]},
+                {"name": "test_pipe", "columns": ["string_attribute", "number_attribute", "date_attribute"]},
+            ],
+        )
+
+        (url,), kwargs = get_request.call_args
+
+        self.assertEqual(kwargs["timeout"], 60)
+        self.assertEqual(kwargs["headers"], {"Authorization": "Bearer p.test.token"})
+
+    @patch("requests.get")
+    def test_run_query(self, get_request):
+        query_runner = self._build_query_runner()
+
+        get_request.return_value = Mock(
+            status_code=200, text=json.dumps(QUERY_RESPONSE), json=Mock(return_value=QUERY_RESPONSE)
+        )
+
+        data, error = query_runner.run_query("SELECT * FROM test_datasource LIMIT 1", None)
+
+        self.assertIsNone(error)
+        self.assertEqual(
+            json.loads(data),
+            {
+                "columns": [
+                    {"name": "string_attribute", "friendly_name": "string_attribute", "type": TYPE_STRING},
+                    {"name": "number_attribute", "friendly_name": "number_attribute", "type": TYPE_INTEGER},
+                    {"name": "date_attribute", "friendly_name": "date_attribute", "type": TYPE_DATETIME},
+                ],
+                "rows": [
+                    {
+                        "string_attribute": "hello world",
+                        "number_attribute": 123,
+                        "date_attribute": "2023-01-01 00:00:03.001000",
+                    }
+                ],
+            },
+        )
+
+        (url,), kwargs = get_request.call_args
+
+        self.assertEqual(url, "https://api.tinybird.co/v0/sql")
+        self.assertEqual(kwargs["timeout"], 60)
+        self.assertEqual(kwargs["headers"], {"Authorization": "Bearer p.test.token"})
+        self.assertEqual(kwargs["params"], {"q": b"SELECT * FROM test_datasource LIMIT 1\nFORMAT JSON"})
+
+    def _mock_tinybird_schema_requests(self, endpoint, **kwargs):
+        response = {}
+
+        if endpoint.endswith(Tinybird.PIPES_ENDPOINT):
+            response = PIPES_RESPONSE
+        if endpoint.endswith(Tinybird.DATASOURCES_ENDPOINT):
+            response = DATASOURCES_RESPONSE
+        if endpoint.endswith(Tinybird.SQL_ENDPOINT):
+            response = SCHEMA_RESPONSE
+
+        return Mock(status_code=200, text=json.dumps(response), json=Mock(return_value=response))
+
+    def _build_query_runner(self):
+        return Tinybird({"url": "https://api.tinybird.co", "token": "p.test.token", "timeout": 60})