Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_coerce_without_dtype():
with pytest.raises(errors.SchemaInitError):
DataFrameSchema({"col": Column(coerce=True)})
with pytest.raises(errors.SchemaInitError):
DataFrameSchema({"col": Column()}, coerce=True)
# raise errors.SchemaInitError even when the schema doesn't specify column
# key for groupby column
def init_schema_no_groupby_column():
DataFrameSchema({
"col1": Column(Int, [
Check(lambda s: s["foo"] > 10, groupby=["col2"]),
]),
})
with pytest.raises(errors.SchemaInitError):
init_schema_no_groupby_column()
# can't use groupby argument in SeriesSchema or Index objects
for SchemaClass in [SeriesSchema, Index]:
with pytest.raises(
errors.SchemaInitError,
match="^Cannot use groupby checks with"):
SchemaClass(Int, Check(lambda s: s["bar"] == 1, groupby="foo"))
# raise errors.SchemaInitError even when the schema doesn't specify column
# key for groupby column
def init_schema_no_groupby_column():
DataFrameSchema({
"col1": Column(Int, [
Check(lambda s: s["foo"] > 10, groupby=["col2"]),
]),
})
with pytest.raises(errors.SchemaInitError):
init_schema_no_groupby_column()
# can't use groupby argument in SeriesSchema or Index objects
for SchemaClass in [SeriesSchema, Index]:
with pytest.raises(
errors.SchemaInitError,
match="^Cannot use groupby checks with"):
SchemaClass(Int, Check(lambda s: s["bar"] == 1, groupby="foo"))
def test_groupby_init_exceptions():
def init_schema_element_wise():
DataFrameSchema({
"col1": Column(Int, [
Check(lambda s: s["foo"] > 10,
element_wise=True,
groupby=["col2"]),
]),
"col2": Column(String, Check(lambda s: s.isin(["foo", "bar"]))),
})
# can't use groupby in Checks where element_wise == True
with pytest.raises(
errors.SchemaInitError,
match=r"^Cannot use groupby when element_wise=True."):
init_schema_element_wise()
# raise errors.SchemaInitError even when the schema doesn't specify column
# key for groupby column
def init_schema_no_groupby_column():
DataFrameSchema({
"col1": Column(Int, [
Check(lambda s: s["foo"] > 10, groupby=["col2"]),
]),
})
with pytest.raises(errors.SchemaInitError):
init_schema_no_groupby_column()
# can't use groupby argument in SeriesSchema or Index objects
def test_groupby_init_exceptions():
def init_schema_element_wise():
DataFrameSchema({
"col1": Column(Int, [
Check(lambda s: s["foo"] > 10,
element_wise=True,
groupby=["col2"]),
]),
"col2": Column(String, Check(lambda s: s.isin(["foo", "bar"]))),
})
# can't use groupby in Checks where element_wise == True
with pytest.raises(
errors.SchemaInitError,
match=r"^Cannot use groupby when element_wise=True."):
init_schema_element_wise()
# raise errors.SchemaInitError even when the schema doesn't specify column
# key for groupby column
def init_schema_no_groupby_column():
DataFrameSchema({
"col1": Column(Int, [
Check(lambda s: s["foo"] > 10, groupby=["col2"]),
]),
})
with pytest.raises(errors.SchemaInitError):
init_schema_no_groupby_column()
# can't use groupby argument in SeriesSchema or Index objects
def _validate_schema(self):
for column_name, column in self.columns.items():
for check in column.checks:
if check.groupby is None or callable(check.groupby):
continue
nonexistent_groupby_columns = [
c for c in check.groupby if c not in self.columns]
if nonexistent_groupby_columns:
raise errors.SchemaInitError(
"groupby argument %s in Check for Column %s not "
"specified in the DataFrameSchema." %
(nonexistent_groupby_columns, column_name))
... "measure_2": [2, 4, 6, 8],
... "group": ["B", "B", "A", "A"]
... })
>>>
>>> schema.validate(df)[["measure_1", "measure_2", "group"]]
measure_1 measure_2 group
0 10 2 B
1 12 4 B
2 14 6 A
3 16 8 A
See :ref:`here` for more usage details.
"""
if element_wise and groupby is not None:
raise errors.SchemaInitError(
"Cannot use groupby when element_wise=True.")
self.fn = fn
self.element_wise = element_wise
self.error = error
self.n_failure_cases = n_failure_cases
if groupby is None and groups is not None:
raise ValueError(
"`groupby` argument needs to be provided when `groups` "
"argument is defined")
if isinstance(groupby, str):
groupby = [groupby]
self.groupby = groupby
if isinstance(groups, str):
groups = [groups]
:type allow_duplicates: bool
"""
self._pandas_dtype = pandas_dtype
self._nullable = nullable
self._allow_duplicates = allow_duplicates
self._coerce = coerce
if checks is None:
checks = []
if isinstance(checks, Check):
checks = [checks]
self.checks = checks
self._name = name
for check in self.checks:
if check.groupby is not None and not self._allow_groupby:
raise errors.SchemaInitError(
"Cannot use groupby checks with type %s" % type(self))