Callbacks

Overview

Callbacks are used to run specific functions after the success and/or failure of your project run. Callbacks can be provided when instantiating your PrismProject or when calling the PrismProject.run() method.

Let's look at an example:

common/
  ├── utils.py
example_project/
  ├── main.py
  ├── callbacks.py
  ├── tasks/
  │   ├── extract.py
  │   ├── transfrom.py
  │   └── load.py

# example_project/main.py

def print_success():
    print("Success!")

def print_failure():
    print("Boo! Project failed!")
    
project = PrismProject(
    ...
    on_success=[print_success],
    on_failure=[print_failure],
)
    
 if __name__ == "__main__":
     project.run()

Here's what's happening in this project:

• In main.py, we see that the user wishes to run print_success upon the project's success and print_failure upon the project's failure. These are specified in the same module as the project instance (i.e., in main.py itself).

• Both print_success and print_failure do not accept any arguments. This is required.

What if the user wants to run functions that do not live inside inside main.py)? Let's take a look at two more examples.

Callbacks live inside a separate module in the project directory

Let's say that, instead of living in main.py, the callbacks live inside callbacks.py:

main.py

# example_project/main.py
    
project = PrismProject(
    ...
    on_success=[...],
    on_failure=[...],
)
    
 if __name__ == "__main__":
     project.run()

callbacks.py

# example_project/callbacks.py

def print_success():
    print("Success!")

def print_failure():
    print("Boo! Project failed!")

In this case, you should be able to import callbacks directly, since it lives in the same directory as main.py:

# example_project/main.py

import callbacks

project = PrismProject(
    ...
    on_success=[callbacks.print_success],
    on_failure=[callbacks.print_failure],
)
    
 if __name__ == "__main__":
     project.run()

Callbacks live outside project directory

Now, let's assume that the callbacks live inside common/utils.py:

main.py

# example_project/main.py
    
project = PrismProject(
    ...
    on_success=[...],
    on_failure=[...],
)
    
 if __name__ == "__main__":
     project.run()

utils.py

# common/callbacks.py

def print_success():
    print("Success!")

def print_failure():
    print("Boo! Project failed!")

In this case, we need to ensure that our project has access to the modules within common. We do this via the package_lookups keyword argument. This argument takes a list of strings or path-like objects. At runtime, Prism adds these paths to sys.path. After the project finishes, these paths are then removed.

When specifying the on_success and on_failure callbacks, we can use the string representation of the import path:

# example_project/main.py
    
project = PrismProject(
    package_lookups=[
        Path(__file__).parent.parent  # the folder that contains commons/
    ]
    on_success=["common.utils.print_success"],  # string repr of import path
    on_failure=["common.utils.print_failure"],  # string repr of import path
)
    
 if __name__ == "__main__":
     project.run()