feat(utils): Add shared manifest inspection utilities

- describe_manifest(): Extract structured schema from manifests
- format_arrow_type(): Format Arrow types into readable strings
- print_schema(): Pretty-print schema in human-readable format

Author: fordN

src/amp/utils/__init__.py

@@ -0,0 +1,5 @@
+"""Utility modules for amp Python client."""
+
+from .manifest_inspector import describe_manifest, format_arrow_type, print_schema
+
+__all__ = ['describe_manifest', 'format_arrow_type', 'print_schema']

src/amp/utils/manifest_inspector.py

@@ -0,0 +1,98 @@
+"""Shared utilities for inspecting dataset manifests.
+
+This module provides functions to parse and display dataset schemas
+from manifest files in a human-readable format.
+"""
+
+from typing import Any, Dict
+
+
+def describe_manifest(manifest: dict) -> Dict[str, list[Dict[str, str | bool]]]:
+    """Extract structured schema information from a manifest.
+
+    Args:
+        manifest: Dataset manifest dictionary
+
+    Returns:
+        dict: Mapping of table names to column information. Each column is a dict with:
+            - name: Column name
+            - type: Arrow type (simplified string representation)
+            - nullable: Whether the column allows NULL values
+    """
+    tables = manifest.get('tables', {})
+
+    result = {}
+    for table_name, table_def in tables.items():
+        schema = table_def.get('schema', {}).get('arrow', {})
+        fields = schema.get('fields', [])
+
+        columns = []
+        for field in fields:
+            col_type = format_arrow_type(field.get('type'))
+            columns.append(
+                {
+                    'name': field.get('name', ''),
+                    'type': col_type,
+                    'nullable': field.get('nullable', True),
+                }
+            )
+
+        result[table_name] = columns
+
+    return result
+
+
+def format_arrow_type(type_def: Any) -> str:
+    """Format Arrow type definition into a readable string.
+
+    Args:
+        type_def: Arrow type definition (str or dict)
+
+    Returns:
+        str: Human-readable type string
+    """
+    if isinstance(type_def, str):
+        return type_def
+    elif isinstance(type_def, dict):
+        # Handle complex types like Timestamp, FixedSizeBinary, Decimal128
+        if 'Timestamp' in type_def:
+            unit = type_def['Timestamp'][0] if type_def['Timestamp'] else 'Unknown'
+            return f'Timestamp({unit})'
+        elif 'FixedSizeBinary' in type_def:
+            size = type_def['FixedSizeBinary']
+            return f'FixedSizeBinary({size})'
+        elif 'Decimal128' in type_def:
+            precision, scale = type_def['Decimal128']
+            return f'Decimal128({precision},{scale})'
+        else:
+            # Fallback for unknown complex types
+            return str(type_def)
+    else:
+        return str(type_def)
+
+
+def print_schema(schema: Dict[str, list[Dict[str, Any]]], header: str = None) -> None:
+    """Pretty-print a schema dictionary.
+
+    Args:
+        schema: Schema dictionary from describe_manifest()
+        header: Optional header text to print before the schema
+    """
+    if header:
+        print(f'\n{header}')
+
+    if not schema:
+        print('\n  (No tables found in manifest)')
+        return
+
+    # Print each table
+    for table_name, columns in schema.items():
+        print(f'\n{table_name} ({len(columns)} columns)')
+        for col in columns:
+            nullable_str = 'NULL    ' if col['nullable'] else 'NOT NULL'
+            # Pad column name for alignment
+            col_name = col['name'].ljust(20)
+            col_type = col['type'].ljust(20)
+            print(f'  {col_name} {col_type} {nullable_str}')
+
+    print()  # Empty line at end

tests/unit/test_registry_inspect.py

@@ -0,0 +1,150 @@
+"""Unit tests for registry dataset inspection methods."""
+
+import pytest
+
+from amp.registry.datasets import RegistryDatasetsClient
+from amp.utils.manifest_inspector import format_arrow_type
+
+
+class MockRegistryClient:
+    """Mock registry client for testing."""
+
+    def __init__(self, manifest):
+        self.manifest = manifest
+
+    def _request(self, method, path, params=None):
+        """Mock HTTP request."""
+
+        class MockResponse:
+            def json(self):
+                return manifest
+
+        return MockResponse()
+
+
+# Sample manifest for testing
+manifest = {
+    'kind': 'manifest',
+    'dependencies': {},
+    'tables': {
+        'blocks': {
+            'schema': {
+                'arrow': {
+                    'fields': [
+                        {'name': 'block_num', 'type': 'UInt64', 'nullable': False},
+                        {'name': 'timestamp', 'type': {'Timestamp': ['Nanosecond', '+00:00']}, 'nullable': False},
+                        {'name': 'hash', 'type': {'FixedSizeBinary': 32}, 'nullable': False},
+                        {'name': 'base_fee_per_gas', 'type': {'Decimal128': [38, 0]}, 'nullable': True},
+                    ]
+                }
+            },
+        },
+        'transactions': {
+            'schema': {
+                'arrow': {
+                    'fields': [
+                        {'name': 'tx_hash', 'type': {'FixedSizeBinary': 32}, 'nullable': False},
+                        {'name': 'from', 'type': {'FixedSizeBinary': 20}, 'nullable': False},
+                        {'name': 'value', 'type': {'Decimal128': [38, 0]}, 'nullable': True},
+                    ]
+                }
+            },
+        },
+    },
+}
+
+
+@pytest.mark.unit
+class TestDatasetInspection:
+    """Test dataset inspection methods."""
+
+    def test_format_arrow_type_primitive(self):
+        """Test formatting primitive Arrow types."""
+        assert format_arrow_type('UInt64') == 'UInt64'
+        assert format_arrow_type('Binary') == 'Binary'
+        assert format_arrow_type('Boolean') == 'Boolean'
+
+    def test_format_arrow_type_timestamp(self):
+        """Test formatting Timestamp types."""
+        result = format_arrow_type({'Timestamp': ['Nanosecond', '+00:00']})
+        assert result == 'Timestamp(Nanosecond)'
+
+        result = format_arrow_type({'Timestamp': ['Microsecond', '+00:00']})
+        assert result == 'Timestamp(Microsecond)'
+
+    def test_format_arrow_type_fixed_binary(self):
+        """Test formatting FixedSizeBinary types."""
+        result = format_arrow_type({'FixedSizeBinary': 32})
+        assert result == 'FixedSizeBinary(32)'
+
+        result = format_arrow_type({'FixedSizeBinary': 20})
+        assert result == 'FixedSizeBinary(20)'
+
+    def test_format_arrow_type_decimal(self):
+        """Test formatting Decimal128 types."""
+        result = format_arrow_type({'Decimal128': [38, 0]})
+        assert result == 'Decimal128(38,0)'
+
+        result = format_arrow_type({'Decimal128': [18, 6]})
+        assert result == 'Decimal128(18,6)'
+
+    def test_describe_returns_correct_structure(self):
+        """Test that describe returns the expected structure."""
+        # Create mock client with test manifest
+        mock_registry = MockRegistryClient(manifest)
+        client = RegistryDatasetsClient(mock_registry)
+
+        # Mock get_manifest to return our test manifest
+        client.get_manifest = lambda ns, name, ver: manifest
+
+        # Call describe
+        schema = client.describe('test', 'dataset', 'latest')
+
+        # Verify structure
+        assert 'blocks' in schema
+        assert 'transactions' in schema
+
+        # Check blocks table
+        blocks = schema['blocks']
+        assert len(blocks) == 4
+        assert blocks[0]['name'] == 'block_num'
+        assert blocks[0]['type'] == 'UInt64'
+        assert blocks[0]['nullable'] is False
+
+        # Check formatted complex types
+        assert blocks[1]['name'] == 'timestamp'
+        assert blocks[1]['type'] == 'Timestamp(Nanosecond)'
+
+        assert blocks[2]['name'] == 'hash'
+        assert blocks[2]['type'] == 'FixedSizeBinary(32)'
+
+        assert blocks[3]['name'] == 'base_fee_per_gas'
+        assert blocks[3]['type'] == 'Decimal128(38,0)'
+        assert blocks[3]['nullable'] is True
+
+    def test_describe_handles_empty_manifest(self):
+        """Test that describe handles manifests with no tables."""
+        empty_manifest = {'kind': 'manifest', 'dependencies': {}, 'tables': {}}
+
+        mock_registry = MockRegistryClient(empty_manifest)
+        client = RegistryDatasetsClient(mock_registry)
+        client.get_manifest = lambda ns, name, ver: empty_manifest
+
+        schema = client.describe('test', 'dataset', 'latest')
+        assert schema == {}
+
+    def test_describe_handles_nullable_field(self):
+        """Test that describe correctly identifies nullable fields."""
+        mock_registry = MockRegistryClient(manifest)
+        client = RegistryDatasetsClient(mock_registry)
+        client.get_manifest = lambda ns, name, ver: manifest
+
+        schema = client.describe('test', 'dataset', 'latest')
+
+        # Check nullable fields
+        transactions = schema['transactions']
+        value_field = next(col for col in transactions if col['name'] == 'value')
+        assert value_field['nullable'] is True
+
+        from_field = next(col for col in transactions if col['name'] == 'from')
+        assert from_field['nullable'] is False