import os

from kedro.framework.session import KedroSession
from kedro.framework.startup import bootstrap_project
from kedro.runner import SequentialRunner
from hooks import ControlTableHooks

if __name__ == "__main__":
    bootstrap_project(os.path.abspath(os.environ.get("PROJECT_PATH")))
    os.chdir(os.environ.get("PROJECT_PATH"))
    with KedroSession.create(env=os.environ.get("kedro_environment")) as session:
        runner = SequentialRunner()
        context = session.load_context()
        pipeline = context.pipelines[os.environ.get("pipeline_name")]
        catalog = context.catalog
        runner.run(pipeline, catalog)
        result_dict = {"message": "Success"}

metadata = bootstrap_project(Path.cwd())
with KedroSession.create(metadata.package_name) as session:
    session.run()

ProjectContext

class ProjectContext(KedroContext):
    def __init__(
        self,
        package_name: str,
        project_path: Union[Path, str],
        env: str = None,
        extra_params: Dict[str, Any] = None,
    ):
        """Init class."""
        super().__init__(
            package_name, project_path, env, extra_params,
        )
    def create_session(self, *args, **kwargs):
        hooks = [ControlTableHooks()]
        return super().create_session(*args, hooks=hooks, **kwargs)

create_session

create_session