44
55import copy
66import traceback
7- from typing import TYPE_CHECKING , Any , Optional
7+ from typing import TYPE_CHECKING , Any , Callable , Optional
88from collections .abc import Iterable
99
1010import ibis
11- import ibis . selectors as s
11+ from ibis import _ , selectors as s
1212from ibis .common .exceptions import IbisError
1313
1414from pandera .api .base .error_handler import ErrorHandler
1515from pandera .config import ValidationScope
1616from pandera .backends .base import CoreCheckResult , ColumnInfo
17+ from pandera .backends .utils import convert_uniquesettings
1718from pandera .backends .ibis .base import IbisSchemaBackend
1819from pandera .errors import (
20+ ParserError ,
1921 SchemaDefinitionError ,
2022 SchemaError ,
2123 SchemaErrorReason ,
@@ -51,6 +53,22 @@ def validate(
5153
5254 column_info = self .collect_column_info (check_obj , schema )
5355
56+ core_parsers : list [tuple [Callable [..., Any ], tuple [Any , ...]]] = [
57+ (self .strict_filter_columns , (schema , column_info )),
58+ ]
59+
60+ for parser , args in core_parsers :
61+ try :
62+ check_obj = parser (check_obj , * args )
63+ except SchemaError as exc :
64+ error_handler .collect_error (
65+ validation_type (exc .reason_code ),
66+ exc .reason_code ,
67+ exc ,
68+ )
69+ except SchemaErrors as exc :
70+ error_handler .collect_errors (exc .schema_errors )
71+
5472 # collect schema components
5573 components = self .collect_schema_components (
5674 check_obj , schema , column_info
@@ -62,6 +80,7 @@ def validate(
6280 # run the checks
6381 core_checks = [
6482 (self .check_column_presence , (check_obj , schema , column_info )),
83+ (self .check_column_values_are_unique , (check_obj , schema )),
6584 (
6685 self .run_schema_component_checks ,
6786 (sample , schema , components , lazy ),
@@ -213,7 +232,7 @@ def collect_column_info(
213232 regex_match_patterns .append (col_schema .name )
214233 except SchemaError :
215234 pass
216- elif col_name in check_obj . columns :
235+ elif col_name in check_obj :
217236 column_names .append (col_name )
218237
219238 # Ibis tables cannot have duplicated column names
@@ -265,6 +284,60 @@ def collect_schema_components(
265284
266285 return schema_components
267286
287+ ###########
288+ # Parsers #
289+ ###########
290+
291+ def strict_filter_columns (
292+ self ,
293+ check_obj : ibis .Table ,
294+ schema : DataFrameSchema ,
295+ column_info : ColumnInfo ,
296+ ) -> ibis .Table :
297+ """Filter columns that aren't specified in the schema."""
298+ # dataframe strictness check makes sure all columns in the dataframe
299+ # are specified in the dataframe schema
300+ if not (schema .strict or schema .ordered ):
301+ return check_obj
302+
303+ filter_out_columns = []
304+ sorted_column_names = iter (column_info .sorted_column_names )
305+ for column in column_info .destuttered_column_names :
306+ is_schema_col = column in column_info .expanded_column_names
307+ if schema .strict is True and not is_schema_col :
308+ raise SchemaError (
309+ schema = schema ,
310+ data = check_obj ,
311+ message = (
312+ f"column '{ column } { schema .__class__ .__name__ }
313+ f" { schema .columns }
314+ ),
315+ failure_cases = column ,
316+ check = "column_in_schema" ,
317+ reason_code = SchemaErrorReason .COLUMN_NOT_IN_SCHEMA ,
318+ )
319+ if schema .strict == "filter" and not is_schema_col :
320+ filter_out_columns .append (column )
321+ if schema .ordered and is_schema_col :
322+ try :
323+ next_ordered_col = next (sorted_column_names )
324+ except StopIteration :
325+ pass
326+ if next_ordered_col != column :
327+ raise SchemaError (
328+ schema = schema ,
329+ data = check_obj ,
330+ message = f"column '{ column } ,
331+ failure_cases = column ,
332+ check = "column_ordered" ,
333+ reason_code = SchemaErrorReason .COLUMN_NOT_ORDERED ,
334+ )
335+
336+ if schema .strict == "filter" :
337+ check_obj = check_obj .drop (filter_out_columns )
338+
339+ return check_obj
340+
268341 ##########
269342 # Checks #
270343 ##########
@@ -312,3 +385,51 @@ def check_column_presence(
312385 )
313386 )
314387 return results
388+
389+ @validate_scope (scope = ValidationScope .DATA )
390+ def check_column_values_are_unique (
391+ self ,
392+ check_obj : ibis .Table ,
393+ schema : DataFrameSchema ,
394+ ) -> CoreCheckResult :
395+ """Check that column values are unique."""
396+
397+ passed = True
398+ message = None
399+ failure_cases = None
400+
401+ if not schema .unique :
402+ return CoreCheckResult (
403+ passed = passed ,
404+ check = "multiple_fields_uniqueness" ,
405+ )
406+
407+ keep_setting = convert_uniquesettings (schema .report_duplicates )
408+ temp_unique : list [list ] = (
409+ [schema .unique ]
410+ if all (isinstance (x , str ) for x in schema .unique )
411+ else schema .unique
412+ )
413+ for lst in temp_unique :
414+ subset = [x for x in lst if x in check_obj ]
415+ if keep_setting == "first" :
416+ duplicated = ibis .row_number ().over (group_by = subset ) > 0
417+ elif keep_setting == "last" :
418+ duplicated = (_ .count () - ibis .row_number ()).over (
419+ group_by = subset
420+ ) > 1
421+ else :
422+ duplicated = _ .count ().over (group_by = subset ) > 1
423+ duplicates = check_obj .select (duplicated = duplicated ).duplicated
424+ if duplicates .any ().execute ():
425+ failure_cases = check_obj .filter (duplicated )
426+ passed = False
427+ message = f"columns '{ * subset ,} \n { failure_cases }
428+ break
429+ return CoreCheckResult (
430+ passed = passed ,
431+ check = "multiple_fields_uniqueness" ,
432+ reason_code = SchemaErrorReason .DUPLICATES ,
433+ message = message ,
434+ failure_cases = failure_cases ,
435+ )
0 commit comments