How to inject dependencies into flows or tasks?

Is it possible to set arguments to a flow / task that are not parameters (non serializable)?
I mean arbitrary object: a service, client, etc with its own lifecycle:

@task
def task_with_injected_dependency(
    param: int,
    service_a: ServiceA = Inject(ServiceA),
):
    return service_a.some_method(param)

# or

@flow
def flow_with_injected_dependency(
    params: list[int],
    service_b: Annotated[ServiceB, Depends()],
):
    nums = service_b.preprocess(params)
    task_with_injected_dependency.map(nums)

currently I do it like this:

# separate function
def call_service(
    param: int,
    service_a: ServiceA = Inject(ServiceA),
):
    return service_a.some_method(param)


@task
def call_service_task(param: int):
    # call the function
    return call_service(param)


@flow
def call_service_flow(params: list[int]):
    service_b = ServiceB()
    nums = service_b.preprocess(params)
    task_with_injected_dependency.map(nums)

1 Like