Custom Datasource

custom_sql.py

from great_expectations.datasource.fluent import SQLDatasource
import sqlalchemy.engine
from typing import Literal, Optional


class CustomSQLDatasource(SQLDatasource):
    type: Literal["my_custom_sql"] = "my_custom_sql"
    connection_string: Optional[str] = None  # this makes this field optional
    some_new_param: dict = {}

    def get_engine(self) -> sqlalchemy.engine.Engine:
        # as long as this returns a sqlalchemy.engine.Engine everything should work
        raise NotImplementedError("use sqlalchemy.engine.create_engine()")
        self._engine = sqlalchemy.engine.create_engine()
        return self._engine


# methods will automagically be generated for attaching this custom Datasource to the context
# the name of this method is taken from the `type` attribute
import great_expectations as gx

context = gx.get_context()
my_ds = context.sources.add_my_custom_sql("my_datasource", some_new_param={"foo": "bar"})
my_asset = my_ds.add_table_asset("my_asset", table_name="MY_TABLE")