feat(registry): add MetabaseRegistry

src/metabase_manager/registry.py

@@ -0,0 +1,58 @@
+from dataclasses import dataclass, field
+from typing import List
+
+from metabase import Database, Field, Metabase, Metric, Segment, Table, User
+
+
+@dataclass
+class MetabaseRegistry:
+    client: Metabase
+
+    databases: List[Database] = field(default_factory=list)
+    tables: List[Table] = field(default_factory=list)
+    users: List[User] = field(default_factory=list)
+    fields: List[Field] = field(default_factory=list)
+    metrics: List[Metric] = field(default_factory=list)
+    segments: List[Segment] = field(default_factory=list)
+
+    def cache_databases(self):
+        """Find all Databases in Metabase and cache in the instance."""
+        self.databases = Database.list(using=self.client)
+
+    def cache_tables(self):
+        """Find all Tables in Metabase and cache in the instance."""
+        self.tables = Table.list(using=self.client)
+
+    def cache_users(self):
+        """Find all Users in Metabase and cache in the instance."""
+        self.users = User.list(using=self.client)
+
+    def cache_fields(self):
+        """Find all Fields in Metabase and cache in the instance."""
+        for table in self.tables:
+            for field in table.fields():
+                self.fields.append(field)
+
+    def get_database(self, id: int) -> Database:
+        """Get a Database by ID."""
+        return next(filter(lambda db: db.id == id, self.databases))
+
+    def get_table(self, id: int) -> Table:
+        """Get a Table by ID."""
+        return next(filter(lambda table: table.id == id, self.tables))
+
+    def get_field(self, id: int) -> Field:
+        """Get a Field by ID."""
+        return next(filter(lambda field: field.id == id, self.fields))
+
+    def get_user(self, id: int) -> User:
+        """Get a User by ID."""
+        return next(filter(lambda user: user.id == id, self.users))
+
+    def get_metric(self, id: int) -> Metric:
+        """Get a Metric by ID."""
+        return next(filter(lambda metric: metric.id == id, self.metrics))
+
+    def get_segment(self, id: int) -> Segment:
+        """Get a Segment by ID."""
+        return next(filter(lambda segment: segment.id == id, self.segments))

tests/test_registry.py

@@ -0,0 +1,94 @@
+from metabase import Database, Field, Metric, Segment, Table, User
+
+from tests.helpers import IntegrationTestCase
+
+from metabase_manager.registry import MetabaseRegistry
+
+
+class MetabaseTests(IntegrationTestCase):
+    def test_cache_databases(self):
+        """Ensure MetabaseRegistry.cache_databases() saves all databases to the instance."""
+        registry = MetabaseRegistry(client=self.metabase)
+
+        registry.cache_databases()
+        self.assertIsInstance(registry.databases, list)
+        self.assertIsInstance(next(iter(registry.databases)), Database)
+
+    def test_cache_tables(self):
+        """Ensure MetabaseRegistry.cache_tables() saves all tables to the instance."""
+        registry = MetabaseRegistry(client=self.metabase)
+
+        registry.cache_tables()
+        self.assertIsInstance(registry.tables, list)
+        self.assertIsInstance(next(iter(registry.tables)), Table)
+
+    def test_cache_users(self):
+        """Ensure MetabaseRegistry.cache_users() saves all users to the instance."""
+        registry = MetabaseRegistry(client=self.metabase)
+
+        registry.cache_users()
+        self.assertIsInstance(registry.users, list)
+        self.assertIsInstance(next(iter(registry.users)), User)
+
+    def test_cache_fields(self):
+        """Ensure MetabaseRegistry.cache_fields() saves all fields to the instance."""
+        registry = MetabaseRegistry(client=self.metabase)
+
+        registry.cache_tables()
+        registry.cache_fields()
+        self.assertIsInstance(registry.fields, list)
+        self.assertIsInstance(next(iter(registry.fields)), Field)
+
+    def test_get_database(self):
+        """Ensure MetabaseRegistry.get_database() returns a Database by ID."""
+        db1 = Database(id=1, _using=self.metabase)
+        db2 = Database(id=2, _using=self.metabase)
+        registry = MetabaseRegistry(databases=[db1, db2], client=self.metabase)
+
+        self.assertEqual(db1, registry.get_database(1))
+        self.assertEqual(db2, registry.get_database(2))
+
+    def test_get_table(self):
+        """Ensure MetabaseRegistry.get_table() returns a Table by ID."""
+        table1 = Table(id=1, _using=self.metabase)
+        table2 = Table(id=2, _using=self.metabase)
+        registry = MetabaseRegistry(tables=[table1, table2], client=self.metabase)
+
+        self.assertEqual(table1, registry.get_table(1))
+        self.assertEqual(table2, registry.get_table(2))
+
+    def test_get_field(self):
+        """Ensure MetabaseRegistry.get_field() returns a Field by ID."""
+        field1 = Field(id=1, _using=self.metabase)
+        field2 = Field(id=2, _using=self.metabase)
+        registry = MetabaseRegistry(fields=[field1, field2], client=self.metabase)
+
+        self.assertEqual(field1, registry.get_field(1))
+        self.assertEqual(field2, registry.get_field(2))
+
+    def test_get_user(self):
+        """Ensure MetabaseRegistry.get_user() returns a User by ID."""
+        user1 = User(id=1, _using=self.metabase)
+        user2 = User(id=2, _using=self.metabase)
+        registry = MetabaseRegistry(users=[user1, user2], client=self.metabase)
+
+        self.assertEqual(user1, registry.get_user(1))
+        self.assertEqual(user2, registry.get_user(2))
+
+    def test_get_metric(self):
+        """Ensure MetabaseRegistry.get_metric() returns a Metric by ID."""
+        metric1 = Metric(id=1, _using=self.metabase)
+        metric2 = Metric(id=2, _using=self.metabase)
+        registry = MetabaseRegistry(metrics=[metric1, metric2], client=self.metabase)
+
+        self.assertEqual(metric1, registry.get_metric(1))
+        self.assertEqual(metric2, registry.get_metric(2))
+
+    def test_get_segment(self):
+        """Ensure MetabaseRegistry.get_segment() returns a Segment by ID."""
+        segment1 = Segment(id=1, _using=self.metabase)
+        segment2 = Segment(id=2, _using=self.metabase)
+        registry = MetabaseRegistry(segments=[segment1, segment2], client=self.metabase)
+
+        self.assertEqual(segment1, registry.get_segment(1))
+        self.assertEqual(segment2, registry.get_segment(2))