diff --git a/.gitignore b/.gitignore index e3afcfe6..1d7bb844 100644 --- a/.gitignore +++ b/.gitignore @@ -186,3 +186,5 @@ docs/_build/ frontend/playwright/.auth frontend/playwright/.downloads frontend/test-results/ + +backend/api/pandera_checks diff --git a/backend/api/adapter/dynamodb_adapter.py b/backend/api/adapter/dynamodb_adapter.py index 734fe7ba..3c99996c 100644 --- a/backend/api/adapter/dynamodb_adapter.py +++ b/backend/api/adapter/dynamodb_adapter.py @@ -85,7 +85,9 @@ def update_job(self, job: Job) -> None: pass @abstractmethod - def get_latest_successful_upload_job(self, dataset: Type[DatasetMetadata]) -> Optional[Dict]: + def get_latest_successful_upload_job( + self, dataset: Type[DatasetMetadata] + ) -> Optional[Dict]: pass @abstractmethod @@ -177,14 +179,15 @@ def store_schema(self, schema: Schema) -> None: AppLogger.info( f"Storing schema for {schema.metadata.string_representation()}" ) - self.schema_table.put_item( - Item={ - "PK": schema.metadata.dataset_identifier(with_version=False), - "SK": schema.metadata.get_version(), - **schema.metadata.model_dump(), - COLUMNS: [col.model_dump() for col in schema.columns], - } - ) + item = { + "PK": schema.metadata.dataset_identifier(with_version=False), + "SK": schema.metadata.get_version(), + **schema.metadata.model_dump(), + COLUMNS: [col.model_dump() for col in schema.columns], + } + if schema.panderaDataFrameSchema is not None: + item["panderaDataFrameSchema"] = schema.panderaDataFrameSchema.to_json() + self.schema_table.put_item(Item=item) except ClientError as error: self._handle_client_error( f"Error storing schema for {schema.metadata.string_representation()}", @@ -389,7 +392,9 @@ def get_job(self, job_id: str) -> Dict: except ClientError as error: self._handle_client_error("Error fetching job from the database", error) - def get_latest_successful_upload_job(self, dataset: Type[DatasetMetadata]) -> Optional[Dict]: + def get_latest_successful_upload_job( + self, dataset: Type[DatasetMetadata] + ) -> Optional[Dict]: """ Get the most recent successful upload job for a specific dataset. Returns the job details including subject_id (uploader) or None if no successful upload exists. @@ -404,13 +409,15 @@ def get_latest_successful_upload_job(self, dataset: Type[DatasetMetadata]) -> Op & Attr("Layer").eq(dataset.layer) & Attr("Domain").eq(dataset.domain) & Attr("Dataset").eq(dataset.dataset) - ) + ), ) if not jobs: return None - sorted_jobs = sorted(jobs, key=lambda x: x.get("CreatedAt", 0), reverse=True) + sorted_jobs = sorted( + jobs, key=lambda x: x.get("CreatedAt", 0), reverse=True + ) return self._map_job(sorted_jobs[0]) except ClientError as error: AppLogger.warning(f"Error fetching latest upload job for dataset: {error}") diff --git a/backend/api/application/services/dataset_validation.py b/backend/api/application/services/dataset_validation.py index 3a82c2c3..8780ee92 100644 --- a/backend/api/application/services/dataset_validation.py +++ b/backend/api/application/services/dataset_validation.py @@ -1,6 +1,6 @@ import re from typing import Tuple - +import json import pandas as pd from pandas import Timestamp import pandera @@ -18,6 +18,7 @@ ) from api.domain.schema import Schema from api.domain.validation_context import ValidationContext +from api.common.logger import AppLogger def build_validated_dataframe(schema: Schema, dataframe: pd.DataFrame) -> pd.DataFrame: @@ -35,6 +36,7 @@ def transform_and_validate(schema: Schema, data: pd.DataFrame) -> pd.DataFrame: .pipe(dataset_has_correct_data_types, schema) .pipe(dataset_has_no_illegal_characters_in_partition_columns, schema) .pipe(validate_with_pandera, schema) + .pipe(validate_with_pandera_schema, schema) ) if validation_context.has_errors(): @@ -120,7 +122,9 @@ def dataset_has_no_illegal_characters_in_partition_columns( error_list = [] for column in schema.get_partition_columns(): series = data_frame[column.name] - if not column.is_of_data_type(DateType) and pd.api.types.is_string_dtype(series): + if not column.is_of_data_type(DateType) and pd.api.types.is_string_dtype( + series + ): any_illegal_characters = series.str.contains("/", na=False).any() if any_illegal_characters: error_list.append( @@ -183,7 +187,7 @@ def parse_pandera_errors(exc: pandera.errors.SchemaErrors) -> list[str]: # Creating a list of singular (json like) entries from the pandera error string # For example: {'check': 'pandera_check', 'error': 'error message', ...} - failure_object_pattern = r'\{\s*(?:[^{}]*?)\}' + failure_object_pattern = r"\{\s*(?:[^{}]*?)\}" failure_objects = re.findall(failure_object_pattern, error_str) # Extracting and cleaning each error statement @@ -195,15 +199,15 @@ def parse_pandera_errors(exc: pandera.errors.SchemaErrors) -> list[str]: continue error_msg = error_match.group(1) - error_msg = error_msg.replace(r"\'", "'").replace(r'\"', '"') + error_msg = error_msg.replace(r"\'", "'").replace(r"\"", '"') check_match = re.search(r'"check":\s*"([^"]*)"', obj) check_name = check_match.group(1) if check_match else None - if ':' in error_msg and ('Name:' in error_msg or 'dtype:' in error_msg): - error_msg = error_msg.split(':')[0] + if ":" in error_msg and ("Name:" in error_msg or "dtype:" in error_msg): + error_msg = error_msg.split(":")[0] - if check_name and check_name not in ['not_nullable', 'field_uniqueness']: + if check_name and check_name not in ["not_nullable", "field_uniqueness"]: error_msg = f"[{check_name}] {error_msg}" error_messages.append(error_msg) @@ -221,3 +225,37 @@ def validate_with_pandera( except pandera.errors.SchemaErrors as exc: error_list = parse_pandera_errors(exc) return data_frame, error_list + + +def parse_pandera_schema_errors(exc: pandera.errors.SchemaErrors) -> list[str]: + + error_messages = [] + + for error_reason in exc.message["DATA"]: + if error_reason == "CHECK_ERROR": + for error_entry in exc.message["DATA"][error_reason]: + if error_entry["column"] is not None: + error_messages.append( + f"Column '{error_entry['column']}' internal error in check: {error_entry['check']}" + ) + else: + error_messages.append( + f"Internal error in check: {error_entry['check']}" + ) + else: + for error_entry in exc.message["DATA"][error_reason]: + error_messages.append(error_entry["error"]) + + return error_messages + + +def validate_with_pandera_schema( + data_frame: pd.DataFrame, schema: Schema +) -> Tuple[pd.DataFrame, list[str]]: + try: + validated_df = schema.pandera_schema_validate(data_frame, lazy=True) + return validated_df, [] + except pandera.errors.SchemaErrors as exc: + AppLogger.info(f" Pandera schema entire error: {exc.message}") + error_list = parse_pandera_schema_errors(exc) + return data_frame, error_list diff --git a/backend/api/application/services/schema_service.py b/backend/api/application/services/schema_service.py index ad3cc509..3bc5b57b 100644 --- a/backend/api/application/services/schema_service.py +++ b/backend/api/application/services/schema_service.py @@ -56,6 +56,7 @@ def _parse_schema(self, schema: dict, only_metadata: bool = False): return Schema( metadata=metadata, columns=[Column.model_validate(col) for col in schema[COLUMNS]], + panderaDataFrameSchema=schema.get("panderaDataFrameSchema", None), ) def get_schema_metadatas( diff --git a/backend/api/domain/schema.py b/backend/api/domain/schema.py index a98e4e81..9e2598a3 100644 --- a/backend/api/domain/schema.py +++ b/backend/api/domain/schema.py @@ -3,8 +3,11 @@ import awswrangler as wr from pydantic.main import BaseModel +from pydantic import field_serializer, field_validator import pyarrow as pa import pandera +import pandera.pandas as pandera_pandas +import pandera.io.pandas_io as pandera_io from api.domain.schema_metadata import Owner, SchemaMetadata from rapid.items.schema import Column, UpdateBehaviour @@ -16,6 +19,21 @@ class Schema(BaseModel): metadata: SchemaMetadata columns: List[Column] + panderaDataFrameSchema: Optional[pandera_pandas.DataFrameSchema] = None + + @field_serializer("panderaDataFrameSchema", mode="plain") + def pandera_dump(self, value: pandera_pandas.DataFrameSchema) -> str: + if value is not None: + return value.to_json() + else: + return value + + @field_validator("panderaDataFrameSchema", mode="before") + def pandera_load(cls, value: str) -> pandera_pandas.DataFrameSchema: + if type(value) is str: + return pandera_io.from_json(value) + else: + return value def get_layer(self) -> str: return self.metadata.get_layer() @@ -100,9 +118,14 @@ def generate_storage_schema(self) -> pa.schema: ) def pandera_validate(self, df, **kwargs): - pandera_columns = { - col.name: col.to_pandera_column() - for col in self.columns - } - pandera_schema = pandera.DataFrameSchema(metadata=self.metadata, columns=pandera_columns) + pandera_columns = {col.name: col.to_pandera_column() for col in self.columns} + pandera_schema = pandera.DataFrameSchema( + metadata=self.metadata, columns=pandera_columns + ) return pandera_schema.validate(df, **kwargs) + + def pandera_schema_validate(self, df, **kwargs): + if self.panderaDataFrameSchema is not None: + return self.panderaDataFrameSchema.validate(df, **kwargs) + else: + return df diff --git a/backend/api/entry.py b/backend/api/entry.py index 0d01e0a4..9df057f0 100644 --- a/backend/api/entry.py +++ b/backend/api/entry.py @@ -42,6 +42,7 @@ from api.controller.subjects import subjects_router from api.controller.user import user_router from api.exception_handler import add_exception_handlers +from api.pandera_custom_checks_load import pandera_custom_checks_load try: load_dotenv() @@ -58,9 +59,7 @@ permissions_service = PermissionsService() upload_service = DatasetAccessEvaluator() -app = FastAPI( - openapi_url=f"{BASE_API_PATH}/openapi.json", docs_url=None -) +app = FastAPI(openapi_url=f"{BASE_API_PATH}/openapi.json", docs_url=None) app.mount("/static", StaticFiles(directory="static"), name="static") app.openapi = custom_openapi_docs_generator(app) add_exception_handlers(app) @@ -79,6 +78,10 @@ @app.on_event("startup") async def startup_event(): init_logger() + try: + pandera_custom_checks_load() + except Exception as e: + AppLogger.info(f"Failed to load pandera custom checks due to: {str(e)}") @app.middleware("http") @@ -247,9 +250,9 @@ def _set_security_headers(response) -> None: "img-src 'self' data: " "fastapi.tiangolo.com/img/favicon.png;" ) - response.headers[ - "Strict-Transport-Security" - ] = "max-age=31536000 ; includeSubDomains" + response.headers["Strict-Transport-Security"] = ( + "max-age=31536000 ; includeSubDomains" + ) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Permitted-Cross-Domain-Policies"] = "none" response.headers["Referrer-Policy"] = "strict-origin" diff --git a/backend/api/pandera_custom_checks_load.py b/backend/api/pandera_custom_checks_load.py new file mode 100644 index 00000000..f50c7014 --- /dev/null +++ b/backend/api/pandera_custom_checks_load.py @@ -0,0 +1,21 @@ +import os +import json +import importlib + + +def pandera_custom_checks_load(): + pandera_files = os.getenv("PANDERA_FILES", None) + if pandera_files is None: + return + + pandera_files = json.loads(pandera_files) + + for file in pandera_files: + file_path = os.path.join("api", "pandera_checks", file) + + os.makedirs(os.path.dirname(file_path), exist_ok=True) + + with open(file_path, "w") as f: + f.write(pandera_files[file]) + + importlib.import_module(file_path.replace("/", ".").replace("\\", ".")[:-3]) diff --git a/backend/rapid/items/schema.py b/backend/rapid/items/schema.py index a8b77b96..53e592b5 100644 --- a/backend/rapid/items/schema.py +++ b/backend/rapid/items/schema.py @@ -1,8 +1,10 @@ # Note: This class is replicated in the api code, they should be de-duplicated once the external dependencies are removed from the API from strenum import StrEnum from typing import Dict, List, Optional, Union, Any -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, field_serializer, field_validator import pandera +import pandera.pandas as pandera_pandas +import pandera.io.pandas_io as pandera_io class SensitivityLevel(StrEnum): @@ -48,7 +50,7 @@ class Column(BaseModel): def is_of_data_type(self, d_type: StrEnum) -> bool: return self.data_type in list(d_type) - + def to_pandera_column(self) -> pandera.Column: """ Convert Column to Pandera Column for Pandera data validation. @@ -96,8 +98,10 @@ def _dict_to_pandera_check(self, check_dict: Dict[str, Any]) -> pandera.Check: pattern = params.get("pattern") return pandera.Check.str_matches(pattern) else: - raise ValueError(f"Unsupported check type: {check_type}. Valid types are: " - "in_range, isin, str_length, greater_than, less_than, str_matches.") + raise ValueError( + f"Unsupported check type: {check_type}. Valid types are: " + "in_range, isin, str_length, greater_than, less_than, str_matches." + ) class Schema(BaseModel): @@ -143,6 +147,21 @@ class Schema(BaseModel): metadata: SchemaMetadata columns: List[Column] + panderaDataFrameSchema: Optional[pandera_pandas.DataFrameSchema] = None + + @field_serializer("panderaDataFrameSchema", mode="plain") + def pandera_dump(self, value: pandera_pandas.DataFrameSchema) -> str: + if value is not None: + return value.to_json() + else: + return value + + @field_validator("panderaDataFrameSchema", mode="before") + def pandera_load(cls, value: str) -> pandera_pandas.DataFrameSchema: + if type(value) is str: + return pandera_io.from_json(value) + else: + return value def are_columns_the_same( self, new_columns: Union[List[Column], List[dict]] @@ -179,9 +198,8 @@ def pandera_validate(self, df, **kwargs): Raises: pandera.errors.SchemaErrors: If validation fails """ - pandera_columns = { - col.name: col.to_pandera_column() - for col in self.columns - } - pandera_schema = pandera.DataFrameSchema(metadata=self.metadata, columns=pandera_columns) + pandera_columns = {col.name: col.to_pandera_column() for col in self.columns} + pandera_schema = pandera.DataFrameSchema( + metadata=self.metadata, columns=pandera_columns + ) return pandera_schema.validate(df, **kwargs) diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt index 39c45337..a042c59d 100644 --- a/backend/requirements-dev.txt +++ b/backend/requirements-dev.txt @@ -8,5 +8,6 @@ requests-mock setuptools twine pip-audit +watchfiles -r ./requirements.txt diff --git a/backend/requirements.txt b/backend/requirements.txt index 712c84f9..da95f233 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -6,7 +6,7 @@ gunicorn httpx jinja2 pandas -pandera +pandera[io] psutil pyarrow pyjwt @@ -17,4 +17,3 @@ uvicorn requests strenum pytest-order - diff --git a/backend/test/api/adapter/test_dynamodb_adapter.py b/backend/test/api/adapter/test_dynamodb_adapter.py index 97ee885c..ca50a75f 100644 --- a/backend/test/api/adapter/test_dynamodb_adapter.py +++ b/backend/test/api/adapter/test_dynamodb_adapter.py @@ -21,6 +21,8 @@ from rapid.items.schema import Column from api.domain.schema_metadata import SchemaMetadata, Owner +import pandera.pandas as pandera_pandas + class TestDynamoDBAdapterGeneric: def setup_method(self): @@ -1034,6 +1036,40 @@ def setup_method(self): ], ) + self.schema_pandera = Schema( + metadata=SchemaMetadata( + layer="raw", + domain="some", + dataset="other", + version=2, + sensitivity="PUBLIC", + description="This is a test schema", + owners=[Owner(name="owner", email="owner@email.com")], + key_only_tags=["key"], + key_value_tags={"key": "value"}, + ), + columns=[ + Column( + name="colname1", + partition_index=0, + data_type="int", + allow_null=False, + ), + Column( + name="colname2", + partition_index=None, + data_type="string", + allow_null=True, + ), + ], + panderaDataFrameSchema=pandera_pandas.DataFrameSchema( + columns={ + "colname1": pandera_pandas.Column(int), + "colname2": pandera_pandas.Column(str), + }, + ), + ) + def test_store_schema(self): self.dynamo_adapter.store_schema(self.schema) @@ -1075,6 +1111,48 @@ def test_store_schema(self): } ) + def test_store_schema_pandera(self): + self.dynamo_adapter.store_schema(self.schema_pandera) + + self.schema_table.put_item.assert_called_once_with( + Item={ + "PK": "raw/some/other", + "SK": 2, + "layer": "raw", + "domain": "some", + "dataset": "other", + "version": 2, + "sensitivity": "PUBLIC", + "description": "This is a test schema", + "update_behaviour": "APPEND", + "key_value_tags": {"key": "value"}, + "key_only_tags": ["key"], + "owners": [{"name": "owner", "email": "owner@email.com"}], + "is_latest_version": True, + "columns": [ + { + "name": "colname1", + "partition_index": 0, + "data_type": "int", + "allow_null": False, + "format": None, + "unique": False, + "checks": {}, + }, + { + "name": "colname2", + "partition_index": None, + "data_type": "string", + "allow_null": True, + "format": None, + "unique": False, + "checks": {}, + }, + ], + "panderaDataFrameSchema": '{"schema_type": "dataframe", "columns": {"colname1": {"dtype": "int64"}, "colname2": {"dtype": "str"}}}', + } + ) + def test_store_schema_client_error(self): self.schema_table.put_item.side_effect = ClientError( error_response={"Error": {"Code": "TableDoesNotExist"}}, diff --git a/backend/test/api/application/services/test_dataset_validation.py b/backend/test/api/application/services/test_dataset_validation.py index 9e860fa9..c4455bcf 100644 --- a/backend/test/api/application/services/test_dataset_validation.py +++ b/backend/test/api/application/services/test_dataset_validation.py @@ -14,7 +14,7 @@ dataset_has_correct_data_types, dataset_has_no_illegal_characters_in_partition_columns, dataset_has_rows, - validate_with_pandera + validate_with_pandera, ) from api.common.custom_exceptions import ( DatasetValidationError, @@ -25,6 +25,8 @@ from rapid.items.schema import Column, Owner from api.domain.schema_metadata import SchemaMetadata +import pandera.pandas as pandera_pandas + class TestDatasetValidation: def setup_method(self): @@ -60,6 +62,38 @@ def setup_method(self): ], ) + self.valid_schema_pandera = Schema( + metadata=SchemaMetadata( + layer="raw", + domain="somedomain", + dataset="otherDataset", + sensitivity="PUBLIC", + owners=[Owner(name="owner", email="owner@email.com")], + ), + columns=[ + Column( + name="colname1", + partition_index=0, + data_type="int", + allow_null=False, + ), + Column( + name="colname2", + partition_index=None, + data_type="string", + allow_null=False, + ), + ], + panderaDataFrameSchema=pandera_pandas.DataFrameSchema( + columns={ + "colname1": pandera_pandas.Column( + int, checks=[pandera_pandas.Check.less_than(10)] + ), + "colname2": pandera_pandas.Column(str), + }, + ), + ) + def test_fully_valid_dataset(self): full_valid_schema = Schema( metadata=self.schema_metadata, @@ -117,6 +151,38 @@ def test_fully_valid_dataset(self): assert validated_dataframe.to_dict() == expected.to_dict() + def test_fully_valid_dataset_pandera_schema(self): + full_valid_schema = self.valid_schema_pandera + + dataframe = pd.DataFrame( + { + "colname1": [1, 4], + "colname2": ["Carlos", "Ada"], + } + ) + + validated_dataframe = build_validated_dataframe(full_valid_schema, dataframe) + + assert validated_dataframe.to_dict() == dataframe.to_dict() + + def test_invalid_dataset_pandera_schema(self): + full_valid_schema = self.valid_schema_pandera + + dataframe = pd.DataFrame( + { + "colname1": [11, 4], + "colname2": ["Carlos", "Ada"], + } + ) + + try: + build_validated_dataframe(full_valid_schema, dataframe) + except DatasetValidationError as error: + print(error) + assert error.message == [ + "Column 'colname1' failed element-wise validator number 0: less_than(10) failure cases: 11" + ] + def test_invalid_column_names(self): dataframe = pd.DataFrame( { @@ -750,13 +816,13 @@ def test_converts_multiple_date_columns(self): ) transformed_df, _ = convert_date_columns(data, schema) - expected_date_column_1 = pd.to_datetime(pd.Series( - ["2008-01-30", "2008-01-31", "2008-02-01", "2008-02-02"] - )) + expected_date_column_1 = pd.to_datetime( + pd.Series(["2008-01-30", "2008-01-31", "2008-02-01", "2008-02-02"]) + ) expected_date_column_1.name = "date1" - expected_date_column_2 = pd.to_datetime(pd.Series( - ["2008-05-15", "2008-12-13", "2008-07-09", "2008-03-17"] - )) + expected_date_column_2 = pd.to_datetime( + pd.Series(["2008-05-15", "2008-12-13", "2008-07-09", "2008-03-17"]) + ) expected_date_column_2.name = "date2" assert transformed_df["date1"].equals(expected_date_column_1) @@ -849,7 +915,7 @@ def test_validate_with_pandera_in_range_check_valid(self): "year_range": { "check_type": "in_range", "parameters": {"min_value": 2000, "max_value": 2030}, - "error": "Year must be between 2000 and 2030" + "error": "Year must be between 2000 and 2030", } }, ), @@ -873,7 +939,7 @@ def test_validate_with_pandera_in_range_check_invalid(self): "year_range": { "check_type": "in_range", "parameters": {"min_value": 2000, "max_value": 2030}, - "error": "Year must be between 2000 and 2030" + "error": "Year must be between 2000 and 2030", } }, ), @@ -899,7 +965,7 @@ def test_validate_with_pandera_isin_check_valid(self): "status_check": { "check_type": "isin", "parameters": {"allowed_values": ["Carlos", "Ada"]}, - "error": "colname1 must be one of: Carlos, Ada" + "error": "colname1 must be one of: Carlos, Ada", } }, ), @@ -923,7 +989,7 @@ def test_validate_with_pandera_isin_check_invalid(self): "status_check": { "check_type": "isin", "parameters": {"allowed_values": ["Carlos", "Ada"]}, - "error": "colname1 must be one of: Carlos, Ada" + "error": "colname1 must be one of: Carlos, Ada", } }, ), @@ -954,13 +1020,13 @@ def test_validate_with_pandera_multiple_checks_on_column(self): "username_length": { "check_type": "str_length", "parameters": {"min_value": 5, "max_value": 20}, - "error": "Username must be between 5 and 20 characters" + "error": "Username must be between 5 and 20 characters", }, "username_pattern": { "check_type": "str_matches", "parameters": {"pattern": r"^[a-z]+\d+$"}, - "error": "Username must be lowercase letters followed by numbers" - } + "error": "Username must be lowercase letters followed by numbers", + }, }, ), Column( @@ -972,13 +1038,13 @@ def test_validate_with_pandera_multiple_checks_on_column(self): "age_minimum": { "check_type": "greater_than", "parameters": {"min_value": 18}, - "error": "Age must be greater than 18" + "error": "Age must be greater than 18", }, "age_maximum": { "check_type": "less_than", "parameters": {"max_value": 100}, - "error": "Age must be less than 100" - } + "error": "Age must be less than 100", + }, }, ), ], @@ -990,7 +1056,11 @@ def test_validate_with_pandera_multiple_checks_on_column(self): def test_validate_with_pandera_multiple_checks_on_column_invalid(self): df = pd.DataFrame( { - "colname1": ["ab", "BOB456", "carlosabcdefghijklmnop"], # Fails str_length and str_matches + "colname1": [ + "ab", + "BOB456", + "carlosabcdefghijklmnop", + ], # Fails str_length and str_matches "colname2": [15, 30, 105], # Fails greater_than and less_than } ) @@ -1006,13 +1076,13 @@ def test_validate_with_pandera_multiple_checks_on_column_invalid(self): "username_length": { "check_type": "str_length", "parameters": {"min_value": 5, "max_value": 20}, - "error": "Username must be between 5 and 20 characters" + "error": "Username must be between 5 and 20 characters", }, "username_pattern": { "check_type": "str_matches", "parameters": {"pattern": r"^[a-z]+\d+$"}, - "error": "Username must be lowercase letters followed by numbers" - } + "error": "Username must be lowercase letters followed by numbers", + }, }, ), Column( @@ -1024,13 +1094,13 @@ def test_validate_with_pandera_multiple_checks_on_column_invalid(self): "age_minimum": { "check_type": "greater_than", "parameters": {"min_value": 18}, - "error": "Age must be greater than 18" + "error": "Age must be greater than 18", }, "age_maximum": { "check_type": "less_than", "parameters": {"max_value": 100}, - "error": "Age must be less than 100" - } + "error": "Age must be less than 100", + }, }, ), ], diff --git a/backend/test/api/application/services/test_schema_service.py b/backend/test/api/application/services/test_schema_service.py index a9bc600a..c05ec400 100644 --- a/backend/test/api/application/services/test_schema_service.py +++ b/backend/test/api/application/services/test_schema_service.py @@ -19,6 +19,8 @@ from rapid.items.schema import Column, Owner from api.domain.schema_metadata import SchemaMetadata +import pandera.pandas as pandera_pandas + class TestUploadSchema: def setup_method(self): @@ -54,6 +56,36 @@ def setup_method(self): ], ) + self.valid_schema_pandera = Schema( + metadata=SchemaMetadata( + layer="raw", + domain="some", + dataset="other", + sensitivity="PUBLIC", + owners=[Owner(name="owner", email="owner@email.com")], + ), + columns=[ + Column( + name="colname1", + partition_index=0, + data_type="int", + allow_null=False, + ), + Column( + name="colname2", + partition_index=None, + data_type="string", + allow_null=True, + ), + ], + panderaDataFrameSchema=pandera_pandas.DataFrameSchema( + columns={ + "colname1": pandera_pandas.Column(int), + "colname2": pandera_pandas.Column(str), + }, + ), + ) + def test_upload_schema(self): self.schema_service.get_schema = Mock(return_value=None) @@ -63,6 +95,19 @@ def test_upload_schema(self): self.glue_adapter.create_table.assert_called_once_with(self.valid_schema) assert result == self.valid_schema.metadata.glue_table_name() + def test_upload_schema_pandera(self): + self.schema_service.get_schema = Mock(return_value=None) + + result = self.schema_service.upload_schema(self.valid_schema_pandera) + + self.dynamodb_adapter.store_schema.assert_called_once_with( + self.valid_schema_pandera + ) + self.glue_adapter.create_table.assert_called_once_with( + self.valid_schema_pandera + ) + assert result == self.valid_schema_pandera.metadata.glue_table_name() + def test_upload_schema_uppercase_domain(self): self.schema_service.get_schema = Mock(return_value=None) @@ -384,6 +429,20 @@ def setup_method(self): "key_only_tags": [], } + pandera_schema = pandera_pandas.DataFrameSchema( + columns={ + "colname1": pandera_pandas.Column(int), + "colname2": pandera_pandas.Column(str), + }, + ) + self.schema_pandera = Schema( + metadata=self.metadata, + columns=self.columns, + panderaDataFrameSchema=pandera_schema, + ) + self.schema_pandera_dict = self.schema_dict.copy() + self.schema_pandera_dict["panderaDataFrameSchema"] = pandera_schema.to_json() + def test_get_schema_success(self): self.dynamodb_adapter.get_schema = Mock(return_value=self.schema_dict) @@ -392,6 +451,14 @@ def test_get_schema_success(self): assert res == self.schema self.dynamodb_adapter.get_schema.assert_called_once_with(self.metadata) + def test_get_schema_pandera_success(self): + self.dynamodb_adapter.get_schema = Mock(return_value=self.schema_pandera_dict) + + res = self.schema_service.get_schema(self.metadata) + + assert res == self.schema_pandera + self.dynamodb_adapter.get_schema.assert_called_once_with(self.metadata) + def test_get_schema_success_latest(self): self.dynamodb_adapter.get_latest_schema = Mock(return_value=self.schema_dict) diff --git a/backend/test/api/application/services/test_schema_validation.py b/backend/test/api/application/services/test_schema_validation.py index 03353f5d..8532b508 100644 --- a/backend/test/api/application/services/test_schema_validation.py +++ b/backend/test/api/application/services/test_schema_validation.py @@ -15,6 +15,8 @@ from api.domain.schema_metadata import SchemaMetadata from rapid.items.schema import UpdateBehaviour, Owner, Column +import pandera.pandas as pandera_pandas + class TestSchemaValidation: def setup_method(self): @@ -48,6 +50,38 @@ def setup_method(self): ], ) + self.valid_schema_pandera = Schema( + metadata=SchemaMetadata( + layer="raw", + domain="somedomain", + dataset="otherDataset", + sensitivity="PUBLIC", + owners=[Owner(name="owner", email="owner@email.com")], + ), + columns=[ + Column( + name="colname1", + partition_index=0, + data_type="int", + allow_null=False, + ), + Column( + name="colname2", + partition_index=None, + data_type="string", + allow_null=False, + ), + ], + panderaDataFrameSchema=pandera_pandas.DataFrameSchema( + columns={ + "colname1": pandera_pandas.Column( + int, checks=[pandera_pandas.Check.less_than(10)] + ), + "colname2": pandera_pandas.Column(str), + }, + ), + ) + def _assert_validate_schema_raises_error( self, invalid_schema: Schema, message_pattern: str ): @@ -55,6 +89,12 @@ def _assert_validate_schema_raises_error( validate_schema(invalid_schema) def test_valid_schema(self): + try: + validate_schema(self.valid_schema_pandera) + except SchemaValidationError: + pytest.fail("Unexpected SchemaError was thrown") + + def test_valid_schema_pandera(self): try: validate_schema(self.valid_schema) except SchemaValidationError: diff --git a/backend/test/rapid/test_items/test_schema.py b/backend/test/rapid/test_items/test_schema.py index 024df744..51216d57 100644 --- a/backend/test/rapid/test_items/test_schema.py +++ b/backend/test/rapid/test_items/test_schema.py @@ -3,6 +3,7 @@ from rapid.items.schema import Schema, SchemaMetadata, Column, Owner, SensitivityLevel +import pandera.pandas as pandera_pandas DUMMY_COLUMNS = [ Column( @@ -318,5 +319,56 @@ def test_schema_returns_correct_dictionary(self): "checks": {}, }, ], + "panderaDataFrameSchema": None, + } + assert schema.model_dump() == expected_dict + + def test_schema_pandera_returns_correct_dictionary(self): + pandera_schema = pandera_pandas.DataFrameSchema( + columns={ + "colname1": pandera_pandas.Column(int), + "colname2": pandera_pandas.Column(str), + }, + ) + schema = Schema( + metadata=DUMMY_METADATA, + columns=DUMMY_COLUMNS, + panderaDataFrameSchema=pandera_schema, + ) + expected_dict = { + "metadata": { + "layer": "raw", + "domain": "test", + "dataset": "rapid_sdk", + "sensitivity": "PUBLIC", + "owners": [{"name": "Test", "email": "test@email.com"}], + "version": None, + "key_value_tags": {}, + "key_only_tags": [], + "description": "test", + "update_behaviour": "OVERWRITE", + "is_latest_version": True, + }, + "columns": [ + { + "name": "column_a", + "data_type": "object", + "partition_index": None, + "allow_null": True, + "format": None, + "unique": False, + "checks": {}, + }, + { + "name": "column_b", + "data_type": "object", + "partition_index": None, + "allow_null": True, + "format": None, + "unique": False, + "checks": {}, + }, + ], + "panderaDataFrameSchema": pandera_schema.to_json(), } assert schema.model_dump() == expected_dict diff --git a/infrastructure/modules/app-cluster/main.tf b/infrastructure/modules/app-cluster/main.tf index 6940c209..80a20263 100644 --- a/infrastructure/modules/app-cluster/main.tf +++ b/infrastructure/modules/app-cluster/main.tf @@ -1,4 +1,5 @@ locals { + pandera_files_json = jsonencode({ for file_path in var.pandera_files : trimprefix(trimprefix(file_path, "/"), "./") => file(file_path) }) environment_variables = merge({ "AWS_ACCOUNT" : var.aws_account, "DATA_BUCKET" : var.data_s3_bucket_name, @@ -9,7 +10,8 @@ locals { "COGNITO_USER_POOL_ID" : var.cognito_user_pool_id, "RESOURCE_PREFIX" : var.resource-name-prefix, "COGNITO_USER_LOGIN_APP_CREDENTIALS_SECRETS_NAME" : var.cognito_user_login_app_credentials_secrets_name, - "CUSTOM_USER_NAME_REGEX" : var.custom_user_name_regex == null ? "" : var.custom_user_name_regex + "CUSTOM_USER_NAME_REGEX" : var.custom_user_name_regex == null ? "" : var.custom_user_name_regex, + "PANDERA_FILES" : local.pandera_files_json }, var.project_information ) diff --git a/infrastructure/modules/app-cluster/variables.tf b/infrastructure/modules/app-cluster/variables.tf index 9232c48f..bd2422f4 100644 --- a/infrastructure/modules/app-cluster/variables.tf +++ b/infrastructure/modules/app-cluster/variables.tf @@ -221,3 +221,10 @@ variable "task_cpu" { description = "rAPId ecs task cpu" default = 256 } + +variable "pandera_files" { + type = list(string) + description = "Python files paths with pandera custom checks" + default = [] + nullable = false +} diff --git a/infrastructure/modules/rapid/main.tf b/infrastructure/modules/rapid/main.tf index bcc7ff00..bb93cef2 100644 --- a/infrastructure/modules/rapid/main.tf +++ b/infrastructure/modules/rapid/main.tf @@ -34,6 +34,7 @@ module "app_cluster" { custom_user_name_regex = var.custom_user_name_regex task_cpu = var.task_cpu task_memory = var.task_memory + pandera_files = var.pandera_files } module "auth" { diff --git a/infrastructure/modules/rapid/variables.tf b/infrastructure/modules/rapid/variables.tf index 79d1f29d..b0c2f69f 100644 --- a/infrastructure/modules/rapid/variables.tf +++ b/infrastructure/modules/rapid/variables.tf @@ -202,3 +202,10 @@ variable "task_cpu" { description = "rAPId ecs task cpu" default = 256 } + +variable "pandera_files" { + type = list(string) + description = "Python files paths with pandera custom checks" + default = [] + nullable = false +}