Add automatic datetime field schema mismatch detection

CRITICAL PRODUCTION SAFETY FEATURE:

Detects when users deploy new datetime indexing code without running
the required migration, preventing runtime query failures.

Features:
- Automatic detection during query execution with helpful warnings
- Manual schema checking via 'om migrate-data check-schema' command
- Programmatic API for application startup validation
- Detailed mismatch reporting with specific models and fields
- Clear guidance on resolution steps

Detection scenarios:
- Code expects NUMERIC datetime indexing (new format)
- Redis has TAG datetime indexing (old format)
- Prevents cryptic syntax errors during queries

Usage:
  om migrate-data check-schema  # Check for mismatches
  om migrate-data datetime      # Fix detected mismatches

This addresses the critical deployment safety issue where users
could deploy new code without running migrations, causing production
query failures. Essential for safe 1.0 rollout.

Author: abrookins

aredis_om/model/cli/migrate_data.py

@@ -575,5 +575,64 @@ def clear_progress(migrations_dir: str, module: str, yes: bool):
     click.echo("✅ Saved migration progress cleared.")
 
 
+@migrate_data.command()
+@click.option(
+    "--migrations-dir",
+    default="",
+    help="Directory containing migration files (default: <root>/data-migrations)",
+)
+@click.option("--module", help="Python module containing migrations")
+@handle_redis_errors
+def check_schema(migrations_dir: str, module: str):
+    """Check for datetime field schema mismatches between code and Redis."""
+    import os
+
+    from ...settings import get_root_migrations_dir
+    from ..migrations.datetime_migration import DatetimeFieldDetector
+
+    resolved_dir = migrations_dir or os.path.join(
+        get_root_migrations_dir(), "data-migrations"
+    )
+    migrator = DataMigrator(
+        migrations_dir=resolved_dir,
+        module_name=module,
+    )
+
+    async def check_schema_async():
+        click.echo("🔍 Checking for datetime field schema mismatches...")
+
+        models = migrator.get_models()
+        detector = DatetimeFieldDetector(migrator.redis)
+        result = await detector.check_for_schema_mismatches(models)
+
+        if not result['has_mismatches']:
+            click.echo("✅ No schema mismatches detected - all datetime fields are properly indexed")
+            return
+
+        click.echo(f"⚠️  Found {len(result['mismatches'])} datetime field schema mismatch(es):")
+        click.echo()
+
+        for mismatch in result['mismatches']:
+            click.echo(f"  Model: {mismatch['model']}")
+            click.echo(f"  Field: {mismatch['field']}")
+            click.echo(f"  Current Redis type: {mismatch['current_type']}")
+            click.echo(f"  Expected type: {mismatch['expected_type']}")
+            click.echo(f"  Index: {mismatch['index_name']}")
+            click.echo()
+
+        click.echo("🚨 CRITICAL ISSUE DETECTED:")
+        click.echo(result['recommendation'])
+        click.echo()
+        click.echo("To fix this issue, run:")
+        click.echo("  om migrate-data datetime")
+        click.echo()
+        click.echo("This will convert your datetime fields from TAG to NUMERIC indexing,")
+        click.echo("enabling proper range queries and sorting.")
+
+        raise click.ClickException("Schema mismatches detected")
+
+    run_async(check_schema_async())
+
+
 if __name__ == "__main__":
     migrate_data()

aredis_om/model/migrations/datetime_migration.py

@@ -20,6 +20,137 @@
 log = logging.getLogger(__name__)
 
 
+class SchemaMismatchError(Exception):
+    """Raised when deployed code expects different field types than what's in Redis."""
+    pass
+
+
+class DatetimeFieldDetector:
+    """Detects datetime field schema mismatches between code and Redis."""
+
+    def __init__(self, redis):
+        self.redis = redis
+
+    async def check_for_schema_mismatches(self, models: List[Any]) -> Dict[str, Any]:
+        """
+        Check if any models have datetime fields that are indexed as TAG instead of NUMERIC.
+
+        This detects the scenario where:
+        1. User had old code with datetime fields indexed as TAG
+        2. User deployed new code that expects NUMERIC indexing
+        3. User hasn't run the migration yet
+
+        Returns:
+            Dict with mismatch information and recommended actions
+        """
+        mismatches = []
+
+        for model in models:
+            try:
+                # Get the current index schema from Redis
+                index_name = f"{model._meta.global_key_prefix}:{model._meta.model_key_prefix}"
+
+                try:
+                    # Try to get index info
+                    index_info = await self.redis.execute_command("FT.INFO", index_name)
+                    current_schema = self._parse_index_schema(index_info)
+                except Exception:
+                    # Index doesn't exist or other error - skip this model
+                    continue
+
+                # Check datetime fields in the model
+                datetime_fields = self._get_datetime_fields(model)
+
+                for field_name, field_info in datetime_fields.items():
+                    redis_field_type = current_schema.get(field_name, {}).get('type')
+
+                    if redis_field_type == 'TAG' and field_info.get('expected_type') == 'NUMERIC':
+                        mismatches.append({
+                            'model': model.__name__,
+                            'field': field_name,
+                            'current_type': 'TAG',
+                            'expected_type': 'NUMERIC',
+                            'index_name': index_name
+                        })
+
+            except Exception as e:
+                log.warning(f"Could not check schema for model {model.__name__}: {e}")
+                continue
+
+        return {
+            'has_mismatches': len(mismatches) > 0,
+            'mismatches': mismatches,
+            'total_affected_models': len(set(m['model'] for m in mismatches)),
+            'recommendation': self._get_recommendation(mismatches)
+        }
+
+    def _parse_index_schema(self, index_info: List) -> Dict[str, Dict[str, Any]]:
+        """Parse FT.INFO output to extract field schema information."""
+        schema = {}
+
+        # FT.INFO returns a list of key-value pairs
+        info_dict = {}
+        for i in range(0, len(index_info), 2):
+            if i + 1 < len(index_info):
+                key = index_info[i].decode() if isinstance(index_info[i], bytes) else str(index_info[i])
+                value = index_info[i + 1]
+                info_dict[key] = value
+
+        # Extract attributes (field definitions)
+        attributes = info_dict.get('attributes', [])
+
+        for attr in attributes:
+            if isinstance(attr, list) and len(attr) >= 4:
+                field_name = attr[0].decode() if isinstance(attr[0], bytes) else str(attr[0])
+                field_type = attr[2].decode() if isinstance(attr[2], bytes) else str(attr[2])
+
+                schema[field_name] = {
+                    'type': field_type,
+                    'raw_attr': attr
+                }
+
+        return schema
+
+    def _get_datetime_fields(self, model) -> Dict[str, Dict[str, Any]]:
+        """Get datetime fields from a model and their expected types."""
+        datetime_fields = {}
+
+        try:
+            # Get model fields in a compatible way
+            if hasattr(model, '_get_model_fields'):
+                model_fields = model._get_model_fields()
+            elif hasattr(model, 'model_fields'):
+                model_fields = model.model_fields
+            else:
+                model_fields = getattr(model, '__fields__', {})
+
+            for field_name, field_info in model_fields.items():
+                # Check if this is a datetime field
+                field_type = getattr(field_info, 'annotation', None)
+                if field_type in (datetime.datetime, datetime.date):
+                    datetime_fields[field_name] = {
+                        'expected_type': 'NUMERIC',  # New code expects NUMERIC
+                        'field_info': field_info
+                    }
+
+        except Exception as e:
+            log.warning(f"Could not analyze fields for model {model.__name__}: {e}")
+
+        return datetime_fields
+
+    def _get_recommendation(self, mismatches: List[Dict]) -> str:
+        """Get recommendation based on detected mismatches."""
+        if not mismatches:
+            return "No schema mismatches detected."
+
+        return (
+            f"CRITICAL: Found {len(mismatches)} datetime field(s) with schema mismatches. "
+            f"Your deployed code expects NUMERIC indexing but Redis has TAG indexing. "
+            f"Run 'om migrate-data datetime' to fix this before queries fail. "
+            f"Affected models: {', '.join(set(m['model'] for m in mismatches))}"
+        )
+
+
 class ConversionFailureMode(Enum):
     """How to handle datetime conversion failures."""
 

aredis_om/model/model.py

@@ -1603,7 +1603,23 @@ async def execute(
 
         # If the offset is greater than 0, we're paginating through a result set,
         # so append the new results to results already in the cache.
-        raw_result = await self.model.db().execute_command(*args)
+        try:
+            raw_result = await self.model.db().execute_command(*args)
+        except Exception as e:
+            error_msg = str(e).lower()
+
+            # Check if this might be a datetime field schema mismatch
+            if "syntax error" in error_msg and self._has_datetime_fields():
+                log.warning(
+                    f"Query failed with syntax error on model with datetime fields. "
+                    f"This might indicate a schema mismatch where datetime fields are "
+                    f"indexed as TAG but code expects NUMERIC. "
+                    f"Run 'om migrate-data check-schema' to verify and "
+                    f"'om migrate-data datetime' to fix."
+                )
+
+            # Re-raise the original exception
+            raise
         if return_raw_result:
             return raw_result
         count = raw_result[0]
@@ -1806,6 +1822,21 @@ async def get_item(self, item: int):
         result = await query.execute()
         return result[0]
 
+    def _has_datetime_fields(self) -> bool:
+        """Check if the model has any datetime fields."""
+        try:
+            import datetime
+            model_fields = self.model._get_model_fields()
+
+            for field_name, field_info in model_fields.items():
+                field_type = getattr(field_info, "annotation", None)
+                if field_type in (datetime.datetime, datetime.date):
+                    return True
+
+            return False
+        except Exception:
+            return False
+
 
 class PrimaryKeyCreator(Protocol):
     def create_pk(self, *args, **kwargs) -> str:
@@ -2226,6 +2257,39 @@ def _get_model_fields(cls):
         else:
             return cls.__fields__
 
+    @classmethod
+    async def check_datetime_schema_compatibility(cls) -> Dict[str, Any]:
+        """
+        Check if this model's datetime fields have compatible schema in Redis.
+
+        This detects if the model was deployed with new datetime indexing code
+        but the migration hasn't been run yet.
+
+        Returns:
+            Dict with compatibility information and warnings
+        """
+        try:
+            from .migrations.datetime_migration import DatetimeFieldDetector
+
+            detector = DatetimeFieldDetector(cls.db())
+            result = await detector.check_for_schema_mismatches([cls])
+
+            if result['has_mismatches']:
+                log.warning(
+                    f"Schema mismatch detected for {cls.__name__}: "
+                    f"{result['recommendation']}"
+                )
+
+            return result
+
+        except Exception as e:
+            log.debug(f"Could not check datetime schema compatibility for {cls.__name__}: {e}")
+            return {
+                'has_mismatches': False,
+                'error': str(e),
+                'recommendation': 'Could not check schema compatibility'
+            }
+
     def __init__(__pydantic_self__, **data: Any) -> None:
         if PYDANTIC_V2:
             is_indexed = __pydantic_self__.model_config.get("index") is True