feat: add Insights Snowflake connectivity validation

analytics_data_api/management/commands/check_insights_snowflake.py

@@ -0,0 +1,49 @@
+"""
+Validate the Insights Snowflake connection from the deployed Analytics API service.
+
+Temporary Phase 1 smoke-test command. Remove it after Snowflake connectivity
+validation is complete and real endpoint migration work owns the connection path.
+"""
+
+from django.core.management.base import BaseCommand, CommandError
+
+from analytics_data_api.snowflake_client import (
+    DEFAULT_VALIDATION_TABLE,
+    SnowflakeConfigurationError,
+    get_insights_snowflake_config,
+    get_table_row_count,
+)
+
+
+class Command(BaseCommand):
+    """Run a manual read-only Snowflake connectivity check."""
+
+    help = 'Validate Insights Snowflake connectivity with a read-only query.'
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--table',
+            default=DEFAULT_VALIDATION_TABLE,
+            help='Snowflake table to count for validation. Defaults to COURSE_ACTIVITY_WEEKLY.',
+        )
+
+    def handle(self, *args, **options):
+        table = options['table']
+
+        try:
+            config = get_insights_snowflake_config()
+            row_count = get_table_row_count(table)
+        except SnowflakeConfigurationError as exc:
+            raise CommandError(str(exc))
+        except Exception as exc:  # pylint: disable=broad-except
+            raise CommandError('Insights Snowflake connectivity check failed: {}'.format(exc))
+
+        self.stdout.write('Insights Snowflake connectivity check succeeded.')
+        self.stdout.write('Account: {}'.format(config['ACCOUNT']))
+        self.stdout.write('User: {}'.format(config['USER']))
+        self.stdout.write('Role: {}'.format(config['ROLE']))
+        self.stdout.write('Warehouse: {}'.format(config['WAREHOUSE']))
+        self.stdout.write('Database: {}'.format(config['DATABASE']))
+        self.stdout.write('Schema: {}'.format(config['SCHEMA']))
+        self.stdout.write('Table: {}'.format(table))
+        self.stdout.write('Row count: {}'.format(row_count))

analytics_data_api/management/commands/tests/test_check_insights_snowflake.py

@@ -0,0 +1,66 @@
+"""Tests for the temporary Insights Snowflake smoke-test command."""
+
+from io import StringIO
+from unittest.mock import patch
+
+from django.core.management import CommandError, call_command
+from django.test import SimpleTestCase
+
+from analytics_data_api.snowflake_client import SnowflakeConfigurationError
+
+VALID_CONFIG = {
+    'ACCOUNT': 'edx.us-east-1',
+    'USER': 'INSIGHTS_API_SERVICE_USER',
+    'ROLE': 'INSIGHTS_API_SERVICE_ROLE',
+    'WAREHOUSE': 'INSIGHTS_API_SERVICE',
+    'DATABASE': 'PROD',
+    'SCHEMA': 'INSIGHTS',
+    'PRIVATE_KEY': 'secret-private-key',
+}
+
+
+class CheckInsightsSnowflakeCommandTests(SimpleTestCase):
+    """
+    Cover the Phase 1 smoke-test command.
+
+    Remove these tests with the command after Snowflake connectivity validation
+    is complete and endpoint migration owns the connection path.
+    """
+
+    @patch('analytics_data_api.management.commands.check_insights_snowflake.get_table_row_count')
+    @patch('analytics_data_api.management.commands.check_insights_snowflake.get_insights_snowflake_config')
+    def test_command_prints_non_secret_connection_context(self, mock_get_config, mock_get_table_row_count):
+        mock_get_config.return_value = VALID_CONFIG
+        mock_get_table_row_count.return_value = 42
+        stdout = StringIO()
+
+        call_command('check_insights_snowflake', '--table', 'COURSE_ACTIVITY_WEEKLY', stdout=stdout)
+
+        output = stdout.getvalue()
+        self.assertIn('Insights Snowflake connectivity check succeeded.', output)
+        self.assertIn('Account: edx.us-east-1', output)
+        self.assertIn('User: INSIGHTS_API_SERVICE_USER', output)
+        self.assertIn('Role: INSIGHTS_API_SERVICE_ROLE', output)
+        self.assertIn('Warehouse: INSIGHTS_API_SERVICE', output)
+        self.assertIn('Database: PROD', output)
+        self.assertIn('Schema: INSIGHTS', output)
+        self.assertIn('Table: COURSE_ACTIVITY_WEEKLY', output)
+        self.assertIn('Row count: 42', output)
+        self.assertNotIn('secret-private-key', output)
+        mock_get_table_row_count.assert_called_once_with('COURSE_ACTIVITY_WEEKLY')
+
+    @patch('analytics_data_api.management.commands.check_insights_snowflake.get_insights_snowflake_config')
+    def test_command_raises_config_errors(self, mock_get_config):
+        mock_get_config.side_effect = SnowflakeConfigurationError('missing config')
+
+        with self.assertRaisesRegex(CommandError, 'missing config'):
+            call_command('check_insights_snowflake')
+
+    @patch('analytics_data_api.management.commands.check_insights_snowflake.get_table_row_count')
+    @patch('analytics_data_api.management.commands.check_insights_snowflake.get_insights_snowflake_config')
+    def test_command_wraps_unexpected_errors(self, mock_get_config, mock_get_table_row_count):
+        mock_get_config.return_value = VALID_CONFIG
+        mock_get_table_row_count.side_effect = RuntimeError('snowflake unavailable')
+
+        with self.assertRaisesRegex(CommandError, 'Insights Snowflake connectivity check failed'):
+            call_command('check_insights_snowflake')

analytics_data_api/snowflake_client.py

@@ -0,0 +1,125 @@
+"""
+Helpers for the Insights Snowflake connection.
+
+This module is intentionally not used by existing API views. Phase 1 only uses
+it from a manual management command to validate deployed Snowflake connectivity.
+"""
+
+import re
+from importlib import import_module
+
+from cryptography.hazmat.primitives import serialization
+from django.conf import settings
+
+QUERY_TAG = 'edx_analytics_api_insights_snowflake_phase1'
+DEFAULT_VALIDATION_TABLE = 'COURSE_ACTIVITY_WEEKLY'
+REQUIRED_CONFIG_KEYS = (
+    'ACCOUNT',
+    'USER',
+    'ROLE',
+    'WAREHOUSE',
+    'DATABASE',
+    'SCHEMA',
+    'PRIVATE_KEY',
+)
+IDENTIFIER_PATTERN = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$')
+
+
+class SnowflakeConfigurationError(Exception):
+    """Raised when Insights Snowflake configuration is missing or unsafe."""
+
+
+def get_insights_snowflake_config():
+    """Return the configured Insights Snowflake settings."""
+    config = getattr(settings, 'INSIGHTS_SNOWFLAKE', None)
+    if not isinstance(config, dict):
+        raise SnowflakeConfigurationError('INSIGHTS_SNOWFLAKE must be configured as a dictionary.')
+
+    missing_keys = [key for key in REQUIRED_CONFIG_KEYS if not config.get(key)]
+    if missing_keys:
+        raise SnowflakeConfigurationError(
+            'INSIGHTS_SNOWFLAKE is missing required settings: {}.'.format(', '.join(missing_keys))
+        )
+
+    return config
+
+
+def get_private_key_der(config):
+    """
+    Convert the configured PEM private key string into DER bytes.
+
+    edx-internal stores the Snowflake private key with literal "\\n" sequences.
+    The connector expects a loaded private key serialized as DER/PKCS8 bytes.
+    """
+    private_key_value = config['PRIVATE_KEY']
+    if not isinstance(private_key_value, str):
+        raise SnowflakeConfigurationError('INSIGHTS_SNOWFLAKE PRIVATE_KEY must be a string.')
+
+    normalized_private_key = private_key_value.replace('\\n', '\n').encode('utf-8')
+    passphrase = config.get('PRIVATE_KEY_PASSPHRASE')
+    passphrase_bytes = passphrase.encode('utf-8') if passphrase else None
+
+    private_key = serialization.load_pem_private_key(
+        normalized_private_key,
+        password=passphrase_bytes,
+    )
+
+    return private_key.private_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+
+
+def connect_to_insights_snowflake():
+    """Open a Snowflake connection using the Insights Snowflake settings."""
+    config = get_insights_snowflake_config()
+    snowflake_connector = import_module('snowflake.connector')
+
+    return snowflake_connector.connect(
+        account=config['ACCOUNT'],
+        user=config['USER'],
+        role=config['ROLE'],
+        warehouse=config['WAREHOUSE'],
+        database=config['DATABASE'],
+        schema=config['SCHEMA'],
+        private_key=get_private_key_der(config),
+        session_parameters={
+            'QUERY_TAG': QUERY_TAG,
+        },
+    )
+
+
+def validate_snowflake_identifier(identifier, label):
+    """Return a safe Snowflake identifier or raise a sanitized error."""
+    if not isinstance(identifier, str) or not IDENTIFIER_PATTERN.match(identifier):
+        raise SnowflakeConfigurationError('Unsafe Snowflake {} identifier.'.format(label))
+    return identifier.upper()
+
+
+def run_read_only_query(sql):
+    """Run a read-only Snowflake query and return all rows."""
+    if not sql.lstrip().upper().startswith('SELECT '):
+        raise SnowflakeConfigurationError('Only SELECT queries are allowed for Insights Snowflake validation.')
+
+    connection = connect_to_insights_snowflake()
+    cursor = None
+    try:
+        cursor = connection.cursor()
+        cursor.execute(sql)
+        return cursor.fetchall()
+    finally:
+        if cursor is not None:
+            cursor.close()
+        connection.close()
+
+
+def get_table_row_count(table_name=DEFAULT_VALIDATION_TABLE):
+    """Return the row count for a validated Snowflake table name."""
+    config = get_insights_snowflake_config()
+    database = validate_snowflake_identifier(config['DATABASE'], 'database')
+    schema = validate_snowflake_identifier(config['SCHEMA'], 'schema')
+    table = validate_snowflake_identifier(table_name, 'table')
+
+    rows = run_read_only_query('SELECT COUNT(*) AS row_count FROM {}.{}.{}'.format(database, schema, table))
+    return rows[0][0] if rows else 0

analytics_data_api/tests/test_snowflake_client.py

@@ -0,0 +1,126 @@
+"""Tests for the temporary Insights Snowflake connectivity helper."""
+
+from unittest.mock import Mock, patch
+
+from django.test import SimpleTestCase, override_settings
+
+from analytics_data_api.snowflake_client import (
+    QUERY_TAG,
+    SnowflakeConfigurationError,
+    connect_to_insights_snowflake,
+    get_insights_snowflake_config,
+    get_private_key_der,
+    get_table_row_count,
+    run_read_only_query,
+    validate_snowflake_identifier,
+)
+
+VALID_CONFIG = {
+    'ACCOUNT': 'edx.us-east-1',
+    'USER': 'INSIGHTS_API_SERVICE_USER',
+    'ROLE': 'INSIGHTS_API_SERVICE_ROLE',
+    'WAREHOUSE': 'INSIGHTS_API_SERVICE',
+    'DATABASE': 'PROD',
+    'SCHEMA': 'INSIGHTS',
+    'PRIVATE_KEY': '-----BEGIN PRIVATE KEY-----\\nkey\\n-----END PRIVATE KEY-----',
+    'PRIVATE_KEY_PASSPHRASE': None,
+}
+
+
+class SnowflakeClientTests(SimpleTestCase):
+    """Cover the connection helper without making real Snowflake calls."""
+
+    @override_settings(INSIGHTS_SNOWFLAKE=VALID_CONFIG)
+    def test_get_insights_snowflake_config(self):
+        self.assertEqual(get_insights_snowflake_config(), VALID_CONFIG)
+
+    @override_settings(INSIGHTS_SNOWFLAKE=None)
+    def test_get_insights_snowflake_config_requires_dictionary(self):
+        with self.assertRaisesRegex(SnowflakeConfigurationError, 'must be configured as a dictionary'):
+            get_insights_snowflake_config()
+
+    @override_settings(INSIGHTS_SNOWFLAKE={})
+    def test_get_insights_snowflake_config_requires_values(self):
+        with self.assertRaisesRegex(SnowflakeConfigurationError, 'missing required settings'):
+            get_insights_snowflake_config()
+
+    def test_get_private_key_der_requires_string_key(self):
+        config = dict(VALID_CONFIG, PRIVATE_KEY=None)
+
+        with self.assertRaisesRegex(SnowflakeConfigurationError, 'PRIVATE_KEY must be a string'):
+            get_private_key_der(config)
+
+    @patch('analytics_data_api.snowflake_client.serialization.load_pem_private_key')
+    def test_get_private_key_der_converts_key_and_passphrase(self, mock_load_pem_private_key):
+        private_key = Mock()
+        private_key.private_bytes.return_value = b'der-key'
+        mock_load_pem_private_key.return_value = private_key
+        config = dict(VALID_CONFIG, PRIVATE_KEY='line1\\nline2', PRIVATE_KEY_PASSPHRASE='passphrase')
+
+        self.assertEqual(get_private_key_der(config), b'der-key')
+
+        mock_load_pem_private_key.assert_called_once_with(b'line1\nline2', password=b'passphrase')
+
+    @override_settings(INSIGHTS_SNOWFLAKE=VALID_CONFIG)
+    @patch('analytics_data_api.snowflake_client.get_private_key_der')
+    @patch('analytics_data_api.snowflake_client.import_module')
+    def test_connect_to_insights_snowflake_uses_config(self, mock_import_module, mock_get_private_key_der):
+        connector = Mock()
+        mock_import_module.return_value = connector
+        mock_get_private_key_der.return_value = b'der-key'
+
+        connect_to_insights_snowflake()
+
+        connector.connect.assert_called_once_with(
+            account='edx.us-east-1',
+            user='INSIGHTS_API_SERVICE_USER',
+            role='INSIGHTS_API_SERVICE_ROLE',
+            warehouse='INSIGHTS_API_SERVICE',
+            database='PROD',
+            schema='INSIGHTS',
+            private_key=b'der-key',
+            session_parameters={'QUERY_TAG': QUERY_TAG},
+        )
+
+    def test_validate_snowflake_identifier(self):
+        self.assertEqual(validate_snowflake_identifier('course_activity_weekly', 'table'), 'COURSE_ACTIVITY_WEEKLY')
+
+    def test_validate_snowflake_identifier_rejects_unsafe_values(self):
+        with self.assertRaisesRegex(SnowflakeConfigurationError, 'Unsafe Snowflake table identifier'):
+            validate_snowflake_identifier('course_activity_weekly;drop table users', 'table')
+
+    def test_run_read_only_query_rejects_non_select_sql(self):
+        with self.assertRaisesRegex(SnowflakeConfigurationError, 'Only SELECT queries are allowed'):
+            run_read_only_query('DELETE FROM PROD.INSIGHTS.COURSE_ACTIVITY_WEEKLY')
+
+    @patch('analytics_data_api.snowflake_client.connect_to_insights_snowflake')
+    def test_run_read_only_query_closes_resources(self, mock_connect_to_insights_snowflake):
+        cursor = Mock()
+        cursor.fetchall.return_value = [(3,)]
+        connection = Mock()
+        connection.cursor.return_value = cursor
+        mock_connect_to_insights_snowflake.return_value = connection
+
+        self.assertEqual(run_read_only_query('SELECT 1'), [(3,)])
+
+        cursor.execute.assert_called_once_with('SELECT 1')
+        cursor.close.assert_called_once_with()
+        connection.close.assert_called_once_with()
+
+    @override_settings(INSIGHTS_SNOWFLAKE=VALID_CONFIG)
+    @patch('analytics_data_api.snowflake_client.run_read_only_query')
+    def test_get_table_row_count(self, mock_run_read_only_query):
+        mock_run_read_only_query.return_value = [(12,)]
+
+        self.assertEqual(get_table_row_count('course_activity_weekly'), 12)
+
+        mock_run_read_only_query.assert_called_once_with(
+            'SELECT COUNT(*) AS row_count FROM PROD.INSIGHTS.COURSE_ACTIVITY_WEEKLY'
+        )
+
+    @override_settings(INSIGHTS_SNOWFLAKE=VALID_CONFIG)
+    @patch('analytics_data_api.snowflake_client.run_read_only_query')
+    def test_get_table_row_count_handles_empty_result(self, mock_run_read_only_query):
+        mock_run_read_only_query.return_value = []
+
+        self.assertEqual(get_table_row_count(), 0)

analyticsdataserver/settings/base.py

@@ -90,6 +90,19 @@
 }
 ########## END DATABASE CONFIGURATION
 
+########## INSIGHTS SNOWFLAKE CONFIGURATION
+INSIGHTS_SNOWFLAKE = {
+    'ACCOUNT': None,
+    'USER': None,
+    'ROLE': None,
+    'WAREHOUSE': None,
+    'DATABASE': 'PROD',
+    'SCHEMA': 'INSIGHTS',
+    'PRIVATE_KEY': None,
+    'PRIVATE_KEY_PASSPHRASE': None,
+}
+########## END INSIGHTS SNOWFLAKE CONFIGURATION
+
 ########## GENERAL CONFIGURATION
 # See: https://docs.djangoproject.com/en/dev/ref/settings/#time-zone
 TIME_ZONE = 'UTC'