Monitor.list()

aflanagan · aflanagan · commit ef6826d81150 · 2025-10-07T18:44:52.000-07:00
diff --git a/README.md b/README.md
@@ -233,6 +233,55 @@ monitors = cronitor.Monitor.put([
 ])
 ```
 
+### Listing and Inspecting Monitors
+
+You can fetch multiple monitors using `Monitor.list()` with optional filtering and pagination:
+
+```python
+import cronitor
+
+# Fetch specific monitors by key
+monitors = cronitor.Monitor.list(['backup-job', 'health-check', 'send-invoices'])
+
+# Fetch all job monitors (first page, 100 results by default)
+monitors = cronitor.Monitor.list(type='job')
+
+# Fetch monitors with filters
+monitors = cronitor.Monitor.list(type='check', group='production', state='failing')
+
+# Fetch a specific page
+monitors = cronitor.Monitor.list(type='job', page=2, pageSize=50)
+
+# Fetch all pages automatically
+monitors = cronitor.Monitor.list(type='job', auto_paginate=True)
+
+# Search monitors
+monitors = cronitor.Monitor.list(search='backup')
+```
+
+After fetching a monitor, access its data using the `.data` property. Nested data is automatically accessible via attributes:
+
+```python
+import cronitor
+
+# Fetch an existing monitor
+monitor = cronitor.Monitor('send-invoices')
+
+# Access monitor attributes
+print(monitor.data.name)
+print(monitor.data.type)
+print(monitor.data.schedule)
+
+# Access nested data
+print(monitor.data.attributes.code)
+if monitor.data.latest_event:
+    print(monitor.data.latest_event.stamp)
+    print(monitor.data.latest_event.event)
+
+# Pretty print the entire monitor as JSON
+print(monitor.data)
+```
+
 ### Pausing, Reseting, and Deleting
 
 ```python
@@ -297,11 +346,17 @@ Fork, then clone the repo:
 Set up your machine:
 
     pip install -r requirements
+    pip install pytest
 
 Make sure the tests pass:
 
     pytest
 
+**Optional:** Run integration tests against the real Cronitor API:
+
+    export CRONITOR_TEST_API_KEY=your_api_key_here
+    pytest cronitor/tests/test_integration.py -v
+
 Make your change. Add tests for your change. Make the tests pass:
 
     pytest
diff --git a/cronitor/monitor.py b/cronitor/monitor.py
@@ -124,6 +124,16 @@ def __init__(self, key, api_key=None, api_version=None, env=None):
 
     @property
     def data(self):
+        """
+        Monitor data with attribute access. Nested dicts are automatically
+        converted to Structs.
+
+        Example:
+            >>> monitor = Monitor('my-monitor')
+            >>> print(monitor.data.name)
+            >>> print(monitor.data.request.url)
+            >>> print(monitor.data)  # Pretty JSON output
+        """
         if self._data and type(self._data) is not Struct:
             self._data = Struct(**self._data)
         elif not self._data:
@@ -199,6 +209,86 @@ def _clean_params(self, params):
     def _ping_api_url(self):
         return "https://cronitor.link/p/{}/{}".format(self.api_key, self.key)
 
+    @classmethod
+    def list(cls, keys=None, page=1, pageSize=100, auto_paginate=False, **filters):
+        """
+        Fetch monitors with optional filtering and pagination.
+
+        Args:
+            keys: Optional list of monitor keys to fetch specifically
+            page: Page number (default: 1)
+            pageSize: Results per page (default: 100)
+            auto_paginate: If True, automatically fetch all pages (default: False)
+            **filters: type, group, tag, state, env, search, sort
+
+        Returns:
+            List of Monitor instances
+
+        Examples:
+            # Fetch specific monitors
+            monitors = Monitor.list(['key1', 'key2'])
+
+            # Fetch first page of job monitors
+            monitors = Monitor.list(type='job')
+
+            # Fetch specific page
+            monitors = Monitor.list(type='job', page=2, pageSize=50)
+
+            # Fetch all pages automatically
+            monitors = Monitor.list(type='job', auto_paginate=True)
+        """
+        if keys:
+            # Fetch specific monitors individually
+            monitors = [cls(key) for key in keys]
+            # Populate data immediately
+            for m in monitors:
+                _ = m.data  # Triggers fetch
+            return monitors
+
+        # Fetch from API with filters
+        monitors = []
+        current_page = page
+
+        while True:
+            result = cls._fetch_page(current_page, pageSize, **filters)
+            monitors.extend(result)
+
+            if not auto_paginate or len(result) < pageSize:
+                # Either not auto-paginating or no more results
+                break
+
+            current_page += 1
+
+        return monitors
+
+    @classmethod
+    def _fetch_page(cls, page, pageSize, **filters):
+        """Fetch a single page of monitors from the API"""
+        api_key = filters.pop('api_key', None) or cronitor.api_key
+        api_version = filters.pop('api_version', None) or cronitor.api_version
+        timeout = cronitor.timeout or 10
+
+        params = dict(filters, page=page, pageSize=pageSize)
+
+        resp = cls._req.get(
+            cls._monitor_api_url(),
+            auth=(api_key, ''),
+            params=params,
+            headers=dict(cls._headers, **{'Cronitor-Version': api_version}),
+            timeout=timeout
+        )
+
+        if resp.status_code == 200:
+            data = resp.json()
+            monitors = []
+            for monitor_data in data.get('monitors', []):
+                m = cls(monitor_data['key'])
+                m.data = monitor_data
+                monitors.append(m)
+            return monitors
+        else:
+            raise cronitor.APIError("Unexpected error %s" % resp.text)
+
     @classmethod
     def _monitor_api_url(cls, key=None):
         if not key: return "https://cronitor.io/api/monitors"
@@ -217,4 +307,27 @@ def _prepare_payload(monitors, rollback=False, request_format=JSON):
 
 class Struct(object):
     def __init__(self, **kwargs):
-        self.__dict__.update(kwargs)
+        for key, value in kwargs.items():
+            if isinstance(value, dict):
+                value = Struct(**value)
+            elif isinstance(value, list):
+                value = [Struct(**item) if isinstance(item, dict) else item for item in value]
+            setattr(self, key, value)
+
+    def __repr__(self):
+        items = ', '.join(f'{k}={v!r}' for k, v in sorted(self.__dict__.items()))
+        return f"Struct({items})"
+
+    def __str__(self):
+        return json.dumps(self._to_dict(), indent=2, sort_keys=True, default=str)
+
+    def _to_dict(self):
+        result = {}
+        for key, value in self.__dict__.items():
+            if isinstance(value, Struct):
+                result[key] = value._to_dict()
+            elif isinstance(value, list):
+                result[key] = [item._to_dict() if isinstance(item, Struct) else item for item in value]
+            else:
+                result[key] = value
+        return result
diff --git a/cronitor/tests/test_integration.py b/cronitor/tests/test_integration.py
@@ -0,0 +1,156 @@
+"""
+Integration tests that make real API calls to Cronitor.
+
+These tests are skipped by default unless CRONITOR_TEST_API_KEY environment
+variable is set. They test against the real Cronitor API to verify behavior.
+
+Usage:
+    export CRONITOR_TEST_API_KEY=your_api_key_here
+    python -m pytest cronitor/tests/test_integration.py -v
+    # or
+    python -m unittest cronitor.tests.test_integration -v
+"""
+
+import os
+import unittest
+import cronitor
+
+
+# Check if integration tests should run
+INTEGRATION_API_KEY = os.getenv('CRONITOR_TEST_API_KEY')
+SKIP_INTEGRATION = not INTEGRATION_API_KEY
+SKIP_REASON = "Set CRONITOR_TEST_API_KEY environment variable to run integration tests"
+
+
+@unittest.skipIf(SKIP_INTEGRATION, SKIP_REASON)
+class MonitorListIntegrationTests(unittest.TestCase):
+    """Integration tests for Monitor.list() against real API"""
+
+    @classmethod
+    def setUpClass(cls):
+        """Set up API key for all tests"""
+        cronitor.api_key = INTEGRATION_API_KEY
+
+    def test_list_all_monitors(self):
+        """Test listing all monitors (first page)"""
+        monitors = cronitor.Monitor.list()
+
+        # Should return a list (may be empty for new accounts)
+        self.assertIsInstance(monitors, list)
+
+        # If there are monitors, verify structure
+        if len(monitors) > 0:
+            monitor = monitors[0]
+            self.assertIsInstance(monitor, cronitor.Monitor)
+            self.assertIsNotNone(monitor.data.key)
+            self.assertIsNotNone(monitor.data.name)
+            print(f"\n✓ Found {len(monitors)} monitors on first page (default pageSize=100)")
+            print(f"  First monitor: {monitor.data.name} ({monitor.data.key})")
+
+    def test_list_all_monitors_auto_paginate(self):
+        """Test listing all monitors with auto_paginate"""
+        monitors_all = cronitor.Monitor.list(auto_paginate=True)
+
+        self.assertIsInstance(monitors_all, list)
+
+        # Check if there were multiple pages
+        monitors_page1 = cronitor.Monitor.list(pageSize=100)
+        if len(monitors_all) > 100:
+            print(f"\n✓ Auto-paginate fetched all {len(monitors_all)} monitors across multiple pages")
+            print(f"  First page had {len(monitors_page1)}, total is {len(monitors_all)}")
+        else:
+            print(f"\n✓ Auto-paginate fetched {len(monitors_all)} monitors (all fit in one page)")
+
+    def test_list_with_pagination(self):
+        """Test listing monitors with specific page size"""
+        monitors = cronitor.Monitor.list(pageSize=5)
+
+        self.assertIsInstance(monitors, list)
+        # Should return at most 5 monitors
+        self.assertLessEqual(len(monitors), 5)
+        print(f"\n✓ Pagination works, got {len(monitors)} monitors (max 5)")
+
+    def test_list_with_filter(self):
+        """Test listing monitors with type filter"""
+        monitors = cronitor.Monitor.list(type='job')
+
+        self.assertIsInstance(monitors, list)
+
+        # Verify all returned monitors are jobs
+        for monitor in monitors:
+            self.assertEqual(monitor.data.type, 'job')
+
+        print(f"\n✓ Filter works, got {len(monitors)} job monitors")
+
+    def test_list_with_search(self):
+        """Test listing monitors with search parameter"""
+        monitors = cronitor.Monitor.list(search='test job')
+
+        self.assertIsInstance(monitors, list)
+
+        # Should return monitors matching search term
+        if len(monitors) > 0:
+            print(f"\n✓ Search works, found {len(monitors)} monitors matching 'test job'")
+            for monitor in monitors[:3]:  # Show first 3
+                print(f"  - {monitor.data.name} ({monitor.data.key})")
+        else:
+            print(f"\n✓ Search works, found 0 monitors matching 'test job'")
+
+    def test_list_specific_keys(self):
+        """Test listing specific monitors by key"""
+        # First get some monitors to test with
+        all_monitors = cronitor.Monitor.list(pageSize=2)
+
+        if len(all_monitors) == 0:
+            self.skipTest("No monitors found in account")
+
+        # Get keys to fetch
+        keys_to_fetch = [m.data.key for m in all_monitors[:min(2, len(all_monitors))]]
+
+        # Fetch them specifically
+        monitors = cronitor.Monitor.list(keys_to_fetch)
+
+        self.assertEqual(len(monitors), len(keys_to_fetch))
+        returned_keys = [m.data.key for m in monitors]
+        self.assertEqual(set(returned_keys), set(keys_to_fetch))
+
+        print(f"\n✓ Fetched specific monitors: {', '.join(keys_to_fetch)}")
+
+    def test_monitor_data_structure(self):
+        """Test that monitor data structure is correct"""
+        monitors = cronitor.Monitor.list(pageSize=1)
+
+        if len(monitors) == 0:
+            self.skipTest("No monitors found in account")
+
+        monitor = monitors[0]
+
+        # Test basic fields exist
+        self.assertIsNotNone(monitor.data.key)
+        self.assertIsNotNone(monitor.data.name)
+        self.assertIsNotNone(monitor.data.type)
+
+        # Test nested attribute access works
+        self.assertIsNotNone(monitor.data.attributes)
+        self.assertIsNotNone(monitor.data.attributes.code)
+
+        # Test pretty printing works
+        json_str = str(monitor.data)
+        self.assertIn(monitor.data.key, json_str)
+        self.assertIn('\n', json_str)  # Pretty formatted
+
+        print(f"\n✓ Monitor data structure correct")
+        print(f"  Key: {monitor.data.key}")
+        print(f"  Name: {monitor.data.name}")
+        print(f"  Type: {monitor.data.type}")
+
+
+if __name__ == '__main__':
+    if SKIP_INTEGRATION:
+        print(f"\n⚠️  {SKIP_REASON}\n")
+        print("Example:")
+        print("  export CRONITOR_TEST_API_KEY=your_api_key_here")
+        print("  python -m unittest cronitor.tests.test_integration -v\n")
+    else:
+        print(f"\n🚀 Running integration tests against Cronitor API...\n")
+        unittest.main()
diff --git a/cronitor/tests/test_monitor.py b/cronitor/tests/test_monitor.py
