feat: add Table

Author: Charles Larivier

metabase/resources/table.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Dict, List
+
+from metabase.resource import ListResource, GetResource, Resource, UpdateResource
+from metabase.resources.field import Field
+from metabase.missing import MISSING
+from metabase.resources.metric import Metric
+from resources.segment import Segment
+
+
+class Dimension(Resource):
+    ENDPOINT = None
+
+    id: str
+    name: str
+    mbql: List[Any]
+    type: str
+
+
+class Table(ListResource, GetResource, UpdateResource):
+    ENDPOINT = "/api/table"
+
+    id: int
+    name: str
+    schema: str
+    db_id: int
+    db: dict
+
+    display_name: str
+    description: str
+
+    entity_name: str
+    entity_type: str
+
+    pk_field: int
+    visibility_type: Table.VisibilityType
+    field_order: Table.FieldOrder
+    points_of_interest: str
+
+    active: bool
+    show_in_getting_started: bool
+
+    created_at: str
+    updated_at: str
+
+    _metrics = None
+    _fields = None
+
+    class VisibilityType(str, Enum):
+        cruft = "cruft"
+        hidden = "hidden"
+        technical = "technical"
+
+    class FieldOrder(str, Enum):
+        alphabetical = "alphabetical"
+        custom = "custom"
+        database = "database"
+        smart = "smart"
+
+    @classmethod
+    def list(cls) -> List[Table]:
+        return super(Table, cls).list()
+
+    @classmethod
+    def get(cls, id: int) -> Table:
+        return super(Table, cls).get(id)
+
+    def update(
+            self,
+            display_name: str = MISSING,
+            description: str = MISSING,
+            field_order: Table.FieldOrder = MISSING,
+            visibility_type: Table.VisibilityType = MISSING,
+            entity_type: str = MISSING,
+            points_of_interest: str = MISSING,
+            caveats: str = MISSING,
+            show_in_getting_started: bool = MISSING,
+            **kwargs
+    ) -> None:
+        return super(Table, self).update(
+            display_name=display_name,
+            description=description,
+            field_order=field_order,
+            visibility_type=visibility_type,
+            entity_type=entity_type,
+            points_of_interest=points_of_interest,
+            caveats=caveats,
+            show_in_getting_started=show_in_getting_started
+        )
+
+    def foreign_keys(self) -> List[dict]:
+        return self.connection().get(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/fks").json()
+
+    def query_metadata(self) -> Dict[str, Any]:
+        return self.connection().get(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/query_metadata").json()
+
+    def related(self) -> Dict[str, Any]:
+        return self.connection().get(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/related").json()
+
+    def discard_values(self):
+        self.connection().post(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/discard_values")
+
+    def rescan_values(self):
+        self.connection().post(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/rescan_values")
+
+    def fields(self) -> List[Field]:
+        return [Field(**field) for field in self.query_metadata().get("fields")]
+
+    def dimensions(self) -> List[Dimension]:
+        return [Dimension(id=id, **dimension) for id, dimension in self.query_metadata().get("dimension_options", {}).items()]
+
+    def metrics(self) -> List[Metric]:
+        return [Metric(**metric) for metric in self.related().get("metrics")]
+
+    def segments(self) -> List[Segment]:
+        return [Segment(**segment) for segment in self.related().get("segments")]
+
+    def get_field(self, id: int) -> Field:
+        return next(filter(lambda field: field.id == id, self.fields()))
+
+    def get_dimension(self, id: str) -> Dimension:
+        return next(filter(lambda dimension: dimension.id == id, self.dimensions()))
+
+    def get_metric(self, id: int) -> Metric:
+        return next(filter(lambda metric: metric.id == id, self.metrics()))
+
+    def get_segment(self, id: int) -> Segment:
+        return next(filter(lambda segment: segment.id == id, self.segments()))

tests/resources/test_table.py

@@ -0,0 +1,141 @@
+from metabase.resources.field import Field
+from metabase.resources.table import Dimension, Table
+from metabase.resources.metric import Metric
+from resources.segment import Segment
+from tests.helpers import IntegrationTestCase
+
+
+class TableTests(IntegrationTestCase):
+    def setUp(self) -> None:
+        super(TableTests, self).setUp()
+
+    def test_list(self):
+        """Ensure Table.list() returns a list of Table instances."""
+        tables = Table.list()
+
+        self.assertIsInstance(tables, list)
+        self.assertTrue(len(tables) > 0)
+        self.assertTrue(all(isinstance(t, Table) for t in tables))
+
+    def test_get(self):
+        """Ensure Table.get() returns a Table instance for a given ID."""
+        table = Table.get(1)
+
+        self.assertIsInstance(table, Table)
+        self.assertEqual(1, table.id)
+
+    def test_update(self):
+        """Ensure Table.update() updates an existing Table in Metabase."""
+        table = Table.get(1)
+
+        display_name = table.display_name
+        table.update(display_name="New Name")
+
+        # assert local instance is mutated
+        self.assertEqual("New Name", table.display_name)
+
+        # assert metabase object is mutated
+        t = Table.get(table.id)
+        self.assertEqual("New Name", t.display_name)
+
+        # teardown
+        t.update(display_name=display_name)
+
+    def test_foreign_keys(self):
+        """Ensure Table.foreign_keys() returns a list of foreign keys as dict."""
+        table = Table.get(1)
+        fks = table.foreign_keys()
+
+        self.assertIsInstance(fks, list)
+        self.assertTrue(len(fks) > 0)
+        self.assertTrue(all(isinstance(fk, dict) for fk in fks))
+
+    def test_query_metadata(self):
+        """Ensure Table.query_metadata() returns a dict."""
+        table = Table.get(1)
+        query_metadata = table.query_metadata()
+
+        self.assertIsInstance(query_metadata, dict)
+
+    def test_related(self):
+        """Ensure Table.related() returns a dict."""
+        table = Table.get(1)
+        related = table.related()
+
+        self.assertIsInstance(related, dict)
+
+    def test_discard_values(self):
+        # TODO
+        pass
+
+    def test_rescan_values(self):
+        # TODO
+        pass
+
+    def test_fields(self):
+        """Ensure Table.fields() returns a list of Field instances."""
+        table = Table.get(1)
+        fields = table.fields()
+
+        self.assertIsInstance(fields, list)
+        self.assertTrue(len(fields) > 0)
+        self.assertTrue(all(isinstance(field, Field) for field in fields))
+
+    def test_dimensions(self):
+        """Ensure Table.dimensions() returns a list of Dimension instances."""
+        table = Table.get(1)
+        dimensions = table.dimensions()
+
+        self.assertIsInstance(dimensions, list)
+        self.assertTrue(len(dimensions) > 0)
+        self.assertTrue(all(isinstance(field, Dimension) for field in dimensions))
+
+    def test_metrics(self):
+        """Ensure Table.metrics() returns a list of Metric instances."""
+        table = Table.get(1)
+        metrics = table.metrics()
+
+        self.assertIsInstance(metrics, list)
+        self.assertEqual(0, len(metrics))
+
+        # fixture
+        metric = Metric.create(
+            name="Products",
+            table_id=1,
+            definition={
+                "aggregation": [["count"]],
+            }
+        )
+
+        metrics = table.metrics()
+        self.assertIsInstance(metrics, list)
+        self.assertEqual(1, len(metrics))
+        self.assertEqual(metric.id, metrics[0].id)
+
+        # teardown
+        metric.archive()
+
+    def test_segments(self):
+        """Ensure Table.segments() returns a list of Segment instances."""
+        table = Table.get(1)
+        segments = table.segments()
+
+        self.assertIsInstance(segments, list)
+        self.assertEqual(0, len(segments))
+
+        # fixture
+        segment = Segment.create(
+            name="Free",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+
+        segments = table.segments()
+        self.assertIsInstance(segments, list)
+        self.assertEqual(1, len(segments))
+        self.assertEqual(segment.id, segments[0].id)
+
+        # teardown
+        segment.archive()