From 3b26ca9737da3de74fcb4286a2b78eac773816a6 Mon Sep 17 00:00:00 2001 From: Gaurav Patil Date: Tue, 13 Oct 2026 00:07:09 +0530 Subject: [PATCH] feat(tables_db): add client-side row validation against table schema Signed-off-by: Gaurav Patil --- appwrite/services/tables_db.py | 9 +++- appwrite/utils/__init__.py | 1 + appwrite/utils/validator.py | 83 ++++++++++++++++++++++++++++++++++ test/test_validator.py | 69 ++++++++++++++++++++++++++++ 4 files changed, 161 insertions(+), 1 deletion(-) create mode 100644 appwrite/utils/validator.py create mode 100644 test/test_validator.py diff --git a/appwrite/services/tables_db.py b/appwrite/services/tables_db.py index f7e8ed2e..32a858f8 100644 --- a/appwrite/services/tables_db.py +++ b/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) diff --git a/appwrite/utils/__init__.py b/appwrite/utils/__init__.py index 5bc7dd0e..31d761ec 100644 --- a/appwrite/utils/__init__.py +++ b/appwrite/utils/__init__.py @@ -1 +1,2 @@ # This file makes the 'utils' directory a Python package. +from .validator import validate_row_data diff --git a/appwrite/utils/validator.py b/appwrite/utils/validator.py new file mode 100644 index 00000000..a7854f44 --- /dev/null +++ b/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') diff --git a/test/test_validator.py b/test/test_validator.py new file mode 100644 index 00000000..5f048f7f --- /dev/null +++ b/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()