How to develop a custom expectation and register like default expectations

GX Core Support
1

use gx 1.13；

I have developed a custom exception checking rule, and I need to register it as globally available. when i use @gx.expectations.register_expectation ，but failed；how can i do？

2

Hi @AlexXu2027, thanks for the question. For most custom checks against a SQL-backed source, UnexpectedRowsExpectation is your best supported option today. It’s documented here.

Beyond that, the public API in v1 doesn’t yet cover writing fully custom Expectations the way the older releases did. Bringing that back is a priority that I’m working on.

In the meantime, the legacy approach from GX ≤0.18 still works. It’s undocumented and leans on library internals that aren’t part of the public API, so consider it use-at-your-own-risk. The upcoming rework will likely change it, so treat it as a stopgap rather than something to build on long-term. With that said, here’s how you can build a custom Expectation with GX v1 today:

You’ll write two classes: a MetricProvider (the actual computation) and an Expectation (a shell that references the metric by name). Both auto-register via metaclasses when their module is imported, no need to call register_expectation. Put the code for your custom Expectation in the plugins directory.

# plugins/custom_expectations/expect_column_values_to_be_valid_email.py

import re
from great_expectations.execution_engine import PandasExecutionEngine
from great_expectations.expectations.expectation import ColumnMapExpectation
from great_expectations.expectations.metrics import ColumnMapMetricProvider, column_condition_partial

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")

class ColumnValuesToBeValidEmail(ColumnMapMetricProvider):  # the metric
    condition_metric_name = "column_values.valid_email"

    @column_condition_partial(engine=PandasExecutionEngine)
    def _pandas(cls, column, **kwargs):
        return column.astype("string").str.match(EMAIL_RE)
    # add @column_condition_partial(engine=SqlAlchemyExecutionEngine) / SparkDFExecutionEngine
    # for those backends — a metric only resolves for engines you implement.

class ExpectColumnValuesToBeValidEmail(ColumnMapExpectation):  # the expectation
    map_metric = "column_values.valid_email"  # this string must match the value of the condition_metric_name field defined on the custom metric

Then, import your custom Expectation anywhere it is referenced. The import must be after you call get_context, since that’s when GX puts the plugin dir on sys.path.

import great_expectations as gx
import pandas as pd

context = gx.get_context(mode="file")  # GX puts the plugins package on sys.path
from custom_expectations.expect_column_values_to_be_valid_email import ExpectColumnValuesToBeValidEmail  # <-- this registers metric + expectation                               

df = pd.DataFrame({"emails": ["a@b.com", "nope", "c@d.co.uk"]})
batch = context.data_sources.pandas_default.read_dataframe(df)

result = batch.validate(ExpectColumnValuesToBeValidEmail(
    column="emails", mostly=0.5)
)
print(result.success)   # True

You’ll need to import the custom Expectation anywhere it’s referenced, for instance prior to loading an ExpectationSuite or running a Checkpoint, otherwise you’ll hit ExpectationNotFoundError. This example is based on a ColumnMapMetricProvider, which evaluates against each row in the data. You can find the other available metric base classes in our legacy 0.18 documentation. I ran it against the current release, GX v1.18.1.