Как передать параметр шаблона Jinja2 задаче с помощью API Taskflow? ⇐ Python

[Anonymous]

У меня есть задача, которая неправильно отображает один из своих параметров, если я передаю ее как "{{ var.json.my_api_key.api_key }}":

Код: Выделить всё

@task
def generate_signature_headers(api_key="{{ var.json.knmi_api_key.api_key }}",
hmac_secret_key="{{ var.json.knmi_api_key.hmac_secret_key }}".encode('utf-8')):

hmac_digest_transformed = ...some code..

return {
"Date": now_utc,
"Authorization": f'Signature keyId="{api_key}",algorithm="hmac-sha512",'
f'signature="{hmac_digest_transformed}" ',
}

Параметр api_key возвращается в виде строки «{{ var.json.knmi_api_key.api_key }}» без обработки, но hmac_secret_key работает как положено.

Все работает, если я передаю параметр api_key как Variable.get('knmi_api_key', deserialize_json=True)['api_key'], но при этом будет создаваться соединение с базой данных метахранилища каждый раз при анализе дага.
Как передать параметр с помощью шаблонизатора Jinja2?

Подробнее здесь: https://stackoverflow.com/questions/701 ... skflow-api