Vandana Malik
05/03/2023, 10:37 AMimport 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"}
any helpJuan Luis
05/03/2023, 10:50 AMmetadata = bootstrap_project(Path.cwd())
with KedroSession.create(metadata.package_name) as session:
session.run()
? (as offered in https://docs.kedro.org/en/0.17.7/04_kedro_project_setup/03_session.html)Vandana Malik
05/03/2023, 10:56 AMJuan Luis
05/03/2023, 11:24 AMVandana Malik
05/03/2023, 11:25 AMJuan Luis
05/03/2023, 11:38 AMProjectContext
was removed in 0.17.0 - could you share a code snippet to better understand your solution?Vandana Malik
05/03/2023, 1:22 PMclass 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)
Nok Lam Chan
05/03/2023, 1:55 PMcreate_session
function? It doesn’t belong to the KedroContext API IIRCJuan Luis
05/03/2023, 2:10 PMcreate_session
method so I suppose this is non-public code. I'm still interested in understanding your solution but let's refrain from sharing more code here just in case 🙂