feat: add Segment

Author: Charles Larivier

metabase/resources/segment.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import List, Optional
+
+from metabase.resource import ListResource, CreateResource, GetResource, UpdateResource
+from metabase.missing import MISSING
+
+
+class Segment(ListResource, CreateResource, GetResource, UpdateResource):
+    ENDPOINT = "/api/segment"
+
+    id: int
+    table_id: int
+    creator_id: int
+
+    name: str
+    description: str
+    definition: dict
+    points_of_interest: str
+    caveats: str
+
+    show_in_getting_started: bool
+    archived: bool
+
+    updated_at: str
+    created_at: str
+
+    @classmethod
+    def list(cls) -> List[Segment]:
+        return super(Segment, cls).list()
+
+    @classmethod
+    def get(cls, id: int) -> Segment:
+        return super(Segment, cls).get(id)
+
+    @classmethod
+    def create(
+            cls,
+            name: str,
+            table_id: int,
+            definition: dict,
+            description: str = "Created by metabase-python.",
+            **kwargs
+    ) -> Segment:
+        return super(Segment, cls).create(
+            name=name, table_id=table_id, definition=definition, description=description, **kwargs
+        )
+
+    def update(
+            self,
+            revision_message: str = "Updated by metabase-python.",
+            name: str = MISSING,
+            description: str = MISSING,
+            definition: dict = MISSING,
+            show_in_getting_started: str = MISSING,
+            points_of_interest: str = MISSING,
+            caveats: str = MISSING,
+            archived: bool = MISSING,
+            **kwargs
+    ) -> None:
+        return super(Segment, self).update(
+            revision_message=revision_message,
+            name=name,
+            description=description,
+            definition=definition,
+            points_of_interest=points_of_interest,
+            caveats=caveats,
+            archived=archived,
+            show_in_getting_started=show_in_getting_started,
+            **kwargs
+        )
+
+    def archive(self):
+        return self.update(archived=True, revision_message="Archived by metabase-python.")

tests/resources/test_segment.py

@@ -0,0 +1,119 @@
+from exceptions import NotFoundError
+from metabase.resources.segment import Segment
+
+from tests.helpers import IntegrationTestCase
+
+
+class SegmentTests(IntegrationTestCase):
+    def tearDown(self) -> None:
+        segments = Segment.list()
+        for segment in segments:
+            segment.archive()
+
+    def test_list(self):
+        """Ensure Segment.list returns a list of Segment instances."""
+        # fixture
+        _ = Segment.create(
+            name="My Segment",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+        _ = Segment.create(
+            name="My Segment",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+
+        segments = Segment.list()
+
+        self.assertIsInstance(segments, list)
+        self.assertEqual(2, len(segments))
+        self.assertTrue(all([isinstance(m, Segment) for m in segments]))
+
+    def test_get(self):
+        """
+        Ensure Segment.get returns a Segment instance for a given ID, or
+        raises a NotFoundError when it does not exist.
+        """
+        # fixture
+        segment = Segment.create(
+            name="My Segment",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+        self.assertIsInstance(segment, Segment)
+
+        m = Segment.get(segment.id)
+        self.assertIsInstance(m, Segment)
+        self.assertEqual(segment.id, m.id)
+
+        with self.assertRaises(NotFoundError):
+            _ = Segment.get(12345)
+
+    def test_create(self):
+        """Ensure Segment.create creates a Segment in Metabase and returns a Segment instance."""
+        segment = Segment.create(
+            name="My Segment",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+
+        self.assertIsInstance(segment, Segment)
+        self.assertEqual("My Segment", segment.name)
+        self.assertEqual(1, segment.table_id)
+        self.assertEqual({"filter": ["=", ["field", 1, None], 0]}, segment.definition)
+
+    def test_update(self):
+        """Ensure Segment.update updates an existing Segment in Metabase."""
+        # fixture
+        segment = Segment.create(
+            name="My Segment",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+
+        self.assertIsInstance(segment, Segment)
+        self.assertEqual("My Segment", segment.name)
+
+        segment.update(
+            name="New Name"
+        )
+        # assert local instance is mutated
+        self.assertEqual("New Name", segment.name)
+
+        # assert metabase object is mutated
+        m = Segment.get(segment.id)
+        self.assertEqual("New Name", m.name)
+
+    def test_archive(self):
+        """Ensure Segment.archive updates archived=True."""
+        # fixture
+        segment = Segment.create(
+            name="My Segment",
+            table_id=1,
+            definition={
+                "filter": ["=", ["field", 1, None], 0],
+            }
+        )
+
+        self.assertIsInstance(segment, Segment)
+        self.assertEqual(False, segment.archived)
+
+        segment.archive()
+        # assert local instance is mutated
+        self.assertEqual(True, segment.archived)
+
+        # assert metabase object is mutated
+        m = Segment.get(segment.id)
+        self.assertEqual(True, m.archived)
+