feat(tables_db): add client-side row validation against table schema closes #134

appwrite/services/tables_db.py

@@ -3,6 +3,7 @@
 from typing import Any, Dict, List, Optional, Union, Type, TypeVar
 from ..exception import AppwriteException
 from appwrite.utils.deprecated import deprecated
+from appwrite.utils.validator import validate_row_data
 from ..models.database_list import DatabaseList
 from ..models.database import Database
 from ..models.dedicated_database_specification_list import DedicatedDatabaseSpecificationList
@@ -4413,6 +4414,7 @@ def create_row(
         permissions: Optional[List[str]] = None,
         transaction_id: Optional[str] = None,
         model_type: Type[T] = dict,
+        schema: Optional[Any] = None,
     ) -> Row[T]:
         """
         Create a new Row. Before using this route, you should create a new table resource using either a [server integration](https://appwrite.io/docs/references/cloud/server-dart/tablesDB#createTable) API or directly from your database console.
@@ -4433,6 +4435,8 @@ def create_row(
             Transaction ID for staging the operation.
         model_type : Type[T], optional
             Pydantic model class for the user-defined data. Defaults to dict for backward compatibility.
+        schema : Optional[Any], optional
+            Optional table schema (ColumnList or list of column definitions) to validate row data client-side before submission.
 
         Returns
         -------
@@ -4442,7 +4446,7 @@ def create_row(
         Raises
         ------
         AppwriteException
-            If API request fails
+            If API request fails or client-side schema validation fails
         """
 
         api_path = '/tablesdb/{databaseId}/tables/{tableId}/rows'
@@ -4455,6 +4459,9 @@ def create_row(
             raise AppwriteException('Missing required parameter: "row_id"')
         if data is None:
             raise AppwriteException('Missing required parameter: "data"')
+
+        if schema is not None:
+            validate_row_data(data, schema)
         api_path = api_path.replace('{databaseId}', str(self._normalize_value(database_id)))
         api_path = api_path.replace('{tableId}', str(self._normalize_value(table_id)))
         api_params['rowId'] = self._normalize_value(row_id)

appwrite/utils/__init__.py

@@ -1 +1,2 @@
 # This file makes the 'utils' directory a Python package.
+from .validator import validate_row_data

appwrite/utils/validator.py

@@ -0,0 +1,83 @@
+from typing import Any, Dict, List, Optional, Union
+from ..exception import AppwriteException
+
+
+def validate_row_data(
+    data: Dict[str, Any],
+    schema: Union[List[Any], Any]
+) -> None:
+    """
+    Validates row data against table schema columns client-side before sending network requests.
+
+    Parameters
+    ----------
+    data : Dict[str, Any]
+        The row payload being inserted or updated.
+    schema : Union[List[Any], ColumnList]
+        A list of Column models or ColumnList representing table schema definitions.
+
+    Raises
+    ------
+    AppwriteException
+        If a required field is missing or an invalid data type is detected.
+    """
+    if not isinstance(data, dict):
+        raise AppwriteException('Row data must be a dictionary/object.')
+
+    if schema is None:
+        return
+
+    columns = getattr(schema, 'columns', schema)
+    if not isinstance(columns, (list, tuple)):
+        return
+
+    type_mapping = {
+        'string': (str,),
+        'text': (str,),
+        'varchar': (str,),
+        'mediumtext': (str,),
+        'longtext': (str,),
+        'email': (str,),
+        'url': (str,),
+        'ip': (str,),
+        'datetime': (str,),
+        'integer': (int,),
+        'bigint': (int,),
+        'float': (float, int),
+        'boolean': (bool,),
+    }
+
+    for col in columns:
+        key = getattr(col, 'key', None) or (col.get('key') if isinstance(col, dict) else None)
+        if not key:
+            continue
+
+        required = getattr(col, 'required', False) or (col.get('required', False) if isinstance(col, dict) else False)
+        col_type = getattr(col, 'type', None) or (col.get('type') if isinstance(col, dict) else None)
+        is_array = getattr(col, 'array', False) or (col.get('array', False) if isinstance(col, dict) else False)
+
+        if key not in data or data[key] is None:
+            if required:
+                raise AppwriteException(f"Missing required attribute '{key}' based on table schema.", 400, 'attribute_required')
+            continue
+
+        val = data[key]
+
+        if is_array:
+            if not isinstance(val, (list, tuple)):
+                raise AppwriteException(f"Attribute '{key}' must be a list/array.", 400, 'attribute_invalid_type')
+            if col_type and col_type in type_mapping:
+                expected_types = type_mapping[col_type]
+                for item in val:
+                    # bool is a subclass of int in Python, so check bool explicitly
+                    if col_type in ('integer', 'bigint', 'float') and isinstance(item, bool):
+                        raise AppwriteException(f"Element in '{key}' must be of type {col_type}, got boolean.", 400, 'attribute_invalid_type')
+                    if not isinstance(item, expected_types):
+                        raise AppwriteException(f"Element in '{key}' must be of type {col_type}, got {type(item).__name__}.", 400, 'attribute_invalid_type')
+        else:
+            if col_type and col_type in type_mapping:
+                expected_types = type_mapping[col_type]
+                if col_type in ('integer', 'bigint', 'float') and isinstance(val, bool):
+                    raise AppwriteException(f"Attribute '{key}' must be of type {col_type}, got boolean.", 400, 'attribute_invalid_type')
+                if not isinstance(val, expected_types):
+                    raise AppwriteException(f"Attribute '{key}' must be of type {col_type}, got {type(val).__name__}.", 400, 'attribute_invalid_type')

test/test_validator.py

@@ -0,0 +1,69 @@
+import unittest
+from appwrite.exception import AppwriteException
+from appwrite.utils.validator import validate_row_data
+
+
+class TestRowValidation(unittest.TestCase):
+    def setUp(self):
+        self.mock_schema = [
+            {'key': 'title', 'type': 'string', 'required': True, 'array': False},
+            {'key': 'views', 'type': 'integer', 'required': False, 'array': False},
+            {'key': 'rating', 'type': 'float', 'required': False, 'array': False},
+            {'key': 'is_published', 'type': 'boolean', 'required': False, 'array': False},
+            {'key': 'tags', 'type': 'string', 'required': False, 'array': True},
+        ]
+
+    def test_valid_payload_passes(self):
+        payload = {
+            'title': 'Getting Started with Appwrite',
+            'views': 120,
+            'rating': 4.8,
+            'is_published': True,
+            'tags': ['cloud', 'backend', 'python'],
+        }
+        # Should execute cleanly without raising
+        validate_row_data(payload, self.mock_schema)
+
+    def test_missing_required_attribute_raises(self):
+        payload = {
+            'views': 10,
+        }
+        with self.assertRaises(AppwriteException) as ctx:
+            validate_row_data(payload, self.mock_schema)
+        self.assertIn("Missing required attribute 'title'", str(ctx.exception))
+        self.assertEqual(ctx.exception.type, 'attribute_required')
+
+    def test_invalid_type_raises(self):
+        payload = {
+            'title': 12345,  # Expected string
+        }
+        with self.assertRaises(AppwriteException) as ctx:
+            validate_row_data(payload, self.mock_schema)
+        self.assertIn("Attribute 'title' must be of type string", str(ctx.exception))
+        self.assertEqual(ctx.exception.type, 'attribute_invalid_type')
+
+    def test_boolean_in_numeric_field_raises(self):
+        payload = {
+            'title': 'Valid Title',
+            'views': True,  # In python bool is subclass of int, but schema must reject
+        }
+        with self.assertRaises(AppwriteException) as ctx:
+            validate_row_data(payload, self.mock_schema)
+        self.assertIn("Attribute 'views' must be of type integer, got boolean", str(ctx.exception))
+
+    def test_array_attribute_validation(self):
+        payload = {
+            'title': 'Valid Title',
+            'tags': 'not-an-array',
+        }
+        with self.assertRaises(AppwriteException) as ctx:
+            validate_row_data(payload, self.mock_schema)
+        self.assertIn("must be a list/array", str(ctx.exception))
+
+    def test_none_schema_is_noop(self):
+        # Should gracefully pass when schema is None
+        validate_row_data({'title': 'Hello'}, None)
+
+
+if __name__ == '__main__':
+    unittest.main()