Dependencies

Ergate has a very similar dependency injection system to FastAPI, which is designed to be very simple to use, and to make it very easy for any developer to integrate other components with their application.

What is dependency injection?

Dependency injection is a way through which your code declares what dependencies it needs to run, and where your code doesn't explicitly create that dependency when it needs to use it, but rather lets the underlying system/application (in this case, Ergate) automatically create those dependencies for it.

Dependency injection is an extremely powerful tool, since it allows you to do things like easily having common shared logic in your code and easily sharing common dependencies across different surfaces of your application, amongst other things; all of it while reducing code repetition.

Creating a dependency

To use dependencies in Ergate, you'll need a function that yields a dependency. It's important to note that the function must yield one time only. As an example, let's create a dependency that returns a datetime object representing the current time.

my_dependency.py

from collections.abc import Generator
from datetime import datetime

def create_current_time() -> Generator[datetime, None, None]:
    yield datetime.now()

Info

Ergate uses contextmanagers under the hood, so any function that works with them can be used as a dependency.

Using a dependency

Now, to use this dependency from within a step, you must first understand the use of an Annotated type. To put it simply, the Annotated type is a way of adding metadata to a type annotation. In this case, the metadata we want to add to our type is that it's a dependency, and that it's generated by a certain callable.

To do so, you need to use the Depends class from Ergate, which takes a callable as its first argument. Let's modify the workflow from the previous section to use the dependency we just created.

my_workflow.py

from datetime import datetime
from typing import Annotated
from ergate import Depends, Workflow
from my_dependency import create_current_time

workflow = Workflow(unique_name="my_first_workflow")

@workflow.step
def step_1(
    input_value: int,
    now: Annotated[
        datetime, # (1)!
        Depends(create_current_time), # (2)!
    ],
) -> int:
    print(f"Hello, I am step 1 and I've received {input_value} at {now}")
    return input_value + 1

@workflow.step
def step_2(input_value: int) -> None:
    print(f"Hello, I am step 2 and I've received {input_value}")

1. The first argument is the actual type of the now argument. This is what type checkers will see it as.
2. Any subsequent arguments are the metadata we want to add to the type. In this case, we're saying that now is a dependency that is generated by the create_current_time callable.

Dependency arguments

Dependencies can take any arguments that workflow steps can take. This means that they can also "see" the input values and they can make use of other dependencies too, and same goes for those other subdependencies, and so on to infinity and beyooond.

---

And now, time for another challenge! Can you modify the create_current_time dependency to receive the current timestamp from another dependency and then yield a datetime object created with it. Give it a try and then check our solution below!

Solution

my_dependency.py

from collections.abc import Generator
from datetime import datetime

def create_current_timestamp() -> Generator[float, None, None]:
    yield datetime.now().timestamp()

def create_current_time(
    timestamp: Annotated[
        float,
        Depends(create_current_timestamp),
    ]
) -> Generator[datetime, None, None]:
    yield datetime.fromtimestamp(timestamp)