feat: add Database

Signed-off-by: Charles Larivier <charles@dribbble.com>

Author: Charles Larivier

metabase/__init__.py

@@ -1,9 +1 @@
 from metabase.metabase import Metabase
-from metabase.resources.dataset import Dataset
-from metabase.resources.field import Field
-from metabase.resources.metric import Metric
-from metabase.resources.permission_group import PermissionGroup
-from metabase.resources.permission_membership import PermissionMembership
-from metabase.resources.segment import Segment
-from metabase.resources.table import Table
-from metabase.resources.user import User

metabase/resources/database.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from metabase import Field, Table
+from metabase.missing import MISSING
+from metabase.resource import (
+    CreateResource,
+    DeleteResource,
+    GetResource,
+    ListResource,
+    UpdateResource,
+)
+
+
+class Database(
+    ListResource, CreateResource, GetResource, UpdateResource, DeleteResource
+):
+    ENDPOINT = "/api/database"
+
+    id: int
+    name: str
+    description: str
+    engine: str
+
+    features: List[Any]
+    details: Dict[str, str]
+    options: str
+    native_permissions: str
+
+    timezone: str
+    metadata_sync_schedule: str
+    cache_field_values_schedule: str
+    cache_ttl: int
+
+    caveats: str
+    refingerprint: str
+    points_of_interest: str
+
+    auto_run_queries: bool
+    is_full_sync: bool
+    is_on_demand: bool
+    is_sample: bool
+
+    updated_at: str
+    created_at: str
+
+    @classmethod
+    def list(cls) -> List[Database]:
+        response = cls.connection().get(cls.ENDPOINT)
+        records = [cls(**db) for db in response.json().get("data", [])]
+        return records
+
+    @classmethod
+    def get(cls, id: int) -> Database:
+        return super(Database, cls).get(id)
+
+    @classmethod
+    def create(
+        cls,
+        name: str,
+        engine: str,
+        details: dict,
+        is_full_sync: bool = None,
+        is_on_demand: bool = None,
+        schedules: dict = None,
+        auto_run_queries: bool = None,
+        cache_ttl: int = None,
+        **kwargs,
+    ) -> Database:
+        """
+        Add a new Database.
+
+        You must be a superuser to do this.
+        """
+        return super(Database, cls).create(
+            name=name,
+            engine=engine,
+            details=details,
+            is_full_sync=is_full_sync,
+            is_on_demand=is_on_demand,
+            schedules=schedules,
+            auto_run_queries=auto_run_queries,
+            cache_ttl=cache_ttl,
+            **kwargs,
+        )
+
+    def update(
+        self,
+        name: str = MISSING,
+        description: str = MISSING,
+        engine: str = MISSING,
+        schedules: dict = MISSING,
+        refingerprint: bool = MISSING,
+        points_of_interest: str = MISSING,
+        auto_run_queries: bool = MISSING,
+        caveats: str = MISSING,
+        is_full_sync: bool = MISSING,
+        cache_ttl: int = MISSING,
+        details: dict = MISSING,
+        is_on_demand: bool = MISSING,
+        **kwargs,
+    ) -> None:
+        """
+        Update a Database.
+
+        You must be a superuser to do this.
+        """
+        return super(Database, self).update(
+            name=name,
+            description=description,
+            engine=engine,
+            schedules=schedules,
+            refingerprint=refingerprint,
+            points_of_interest=points_of_interest,
+            auto_run_queries=auto_run_queries,
+            caveats=caveats,
+            is_full_sync=is_full_sync,
+            cache_ttl=cache_ttl,
+            details=details,
+            is_on_demand=is_on_demand,
+        )
+
+    def delete(self) -> None:
+        """Delete a Database."""
+        return super(Database, self).delete()
+
+    def fields(self) -> List[Field]:
+        """Get a list of all Fields in Database."""
+        fields = (
+            self.connection()
+            .get(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/fields")
+            .json()
+        )
+        return [Field(**payload) for payload in fields]
+
+    def idfields(self) -> List[Field]:
+        """Get a list of all primary key Fields for Database."""
+        fields = (
+            self.connection()
+            .get(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/idfields")
+            .json()
+        )
+        return [Field(**payload) for payload in fields]
+
+    def schemas(self) -> List[str]:
+        """Returns a list of all the schemas found for the database id."""
+        return (
+            self.connection()
+            .get(self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/schemas")
+            .json()
+        )
+
+    def tables(self, schema: str) -> List[Table]:
+        """Returns a list of Tables for the given Database id and schema."""
+        tables = (
+            self.connection()
+            .get(
+                self.ENDPOINT
+                + f"/{getattr(self, self.PRIMARY_KEY)}"
+                + "/schema"
+                + f"/{schema}"
+            )
+            .json()
+        )
+        return [Table(**payload) for payload in tables]
+
+    def discard_values(self):
+        """
+        Discards all saved field values for this Database.
+
+        You must be a superuser to do this.
+        """
+        return self.connection().post(
+            self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/discard_values"
+        )
+
+    def rescan_values(self):
+        """
+        Trigger a manual scan of the field values for this Database.
+
+        You must be a superuser to do this.
+        """
+        return self.connection().post(
+            self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/rescan_values"
+        )
+
+    def sync(self):
+        """Update the metadata for this Database. This happens asynchronously."""
+        return self.connection().post(
+            self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/sync"
+        )
+
+    def sync_schema(self):
+        """
+        Trigger a manual update of the schema metadata for this Database.
+
+        You must be a superuser to do this.
+        """
+        return self.connection().post(
+            self.ENDPOINT + f"/{getattr(self, self.PRIMARY_KEY)}" + "/sync_schema"
+        )

tests/resources/test_database.py

@@ -0,0 +1,152 @@
+from exceptions import NotFoundError
+
+from metabase import Database
+from metabase.resources.field import Field
+from metabase.resources.table import Table
+from tests.helpers import IntegrationTestCase
+
+
+class DatabaseTests(IntegrationTestCase):
+    def setUp(self) -> None:
+        super(DatabaseTests, self).setUp()
+
+    def test_import(self):
+        """Ensure Database can be imported from Metabase."""
+        from metabase import Database
+
+        self.assertIsNotNone(Database())
+
+    def test_list(self):
+        """Ensure Database.list() returns a list of Database instances."""
+        databases = Database.list()
+
+        self.assertIsInstance(databases, list)
+        self.assertTrue(len(databases) > 0)
+        self.assertTrue(all(isinstance(t, Database) for t in databases))
+
+    def test_get(self):
+        """Ensure Database.get() returns a Database instance for a given ID."""
+        database = Database.get(1)
+
+        self.assertIsInstance(database, Database)
+        self.assertEqual(1, database.id)
+
+    def test_create(self):
+        """Ensure Database.create() creates a Database in Metabase and returns a Database instance."""
+        database = Database.create(
+            name="Test",
+            engine="h2",
+            details={
+                "db": "zip:/app/metabase.jar!/sample-dataset.db;USER=GUEST;PASSWORD=guest"
+            },
+        )
+
+        self.assertIsInstance(database, Database)
+        self.assertEqual("Test", database.name)
+        self.assertEqual("h2", database.engine)
+        self.assertIsInstance(
+            Database.get(database.id), Database
+        )  # instance exists in Metabase
+
+        # teardown
+        database.delete()
+
+    def test_update(self):
+        """Ensure Database.update() updates an existing Database in Metabase."""
+        database = Database.get(1)
+
+        name = database.name
+        database.update(name="New Name")
+
+        # assert local instance is mutated
+        self.assertEqual("New Name", database.name)
+
+        # assert metabase object is mutated
+        t = Database.get(database.id)
+        self.assertEqual("New Name", t.name)
+
+        # teardown
+        t.update(name=name)
+
+    def test_delete(self):
+        """Ensure Database.delete() deletes a Database in Metabase."""
+        # fixture
+        database = Database.create(
+            name="Test",
+            engine="h2",
+            details={
+                "db": "zip:/app/metabase.jar!/sample-dataset.db;USER=GUEST;PASSWORD=guest"
+            },
+        )
+        self.assertIsInstance(database, Database)
+
+        database.delete()
+
+        # assert metabase object is mutated
+        with self.assertRaises(NotFoundError):
+            _ = Database.get(database.id)
+
+    def test_fields(self):
+        """Ensure Database.fields() returns a list of Field instances."""
+        database = Database.get(1)
+        fields = database.fields()
+
+        self.assertIsInstance(fields, list)
+        self.assertTrue(len(fields) > 0)
+        self.assertTrue(all(isinstance(t, Field) for t in fields))
+
+    def test_idfields(self):
+        """Ensure Database.idfields() returns a list of Field instances."""
+        database = Database.get(1)
+        fields = database.idfields()
+
+        self.assertIsInstance(fields, list)
+        self.assertTrue(len(fields) > 0)
+        self.assertTrue(all(isinstance(t, Field) for t in fields))
+
+    def test_schemas(self):
+        """Ensure Database.schemas() returns a list of strings."""
+        database = Database.get(1)
+        schemas = database.schemas()
+
+        self.assertIsInstance(schemas, list)
+        self.assertTrue(len(schemas) > 0)
+        self.assertTrue(all(isinstance(t, str) for t in schemas))
+
+    def test_tables(self):
+        """Ensure Database.tables() returns a list of Table instances."""
+        database = Database.get(1)
+        schema = database.schemas()[0]
+        tables = database.tables(schema)
+
+        self.assertIsInstance(tables, list)
+        self.assertTrue(len(tables) > 0)
+        self.assertTrue(all(isinstance(t, Table) for t in tables))
+
+    def test_discard_values(self):
+        """Ensure Database.discard_values() does not raise an error."""
+        database = Database.get(1)
+        response = database.discard_values()
+
+        self.assertEqual(200, response.status_code)
+
+    def test_rescan_values(self):
+        """Ensure Database.rescan_values() does not raise an error."""
+        database = Database.get(1)
+        response = database.rescan_values()
+
+        self.assertEqual(200, response.status_code)
+
+    def test_sync(self):
+        """Ensure Database.sync() does not raise an error."""
+        database = Database.get(1)
+        response = database.sync()
+
+        self.assertEqual(200, response.status_code)
+
+    def test_sync_schema(self):
+        """Ensure Database.sync_schema() does not raise an error."""
+        database = Database.get(1)
+        response = database.sync_schema()
+
+        self.assertEqual(200, response.status_code)