Александр Лебедев

Пока писал вопрос, подумал с другого конца и нашел решение :)

d = dict(p1=1, p2=2)
def f2(p1,p2):
    print p1, p2
f2(**d)

Была внутреняя проблема сервиса (редирект на предыдущую версию API), код заработал, все хорошо :)

Решил переносом соединения в базу и проверки результата в саму функцию-worker. Примерно так:

def worker_email(keyword, task_id):

    # Соединяюсь с базой
    client = MongoClient('mongodb://localhost:27017/')
    db = client.admetric
    stream = db.stream

    sleep(10)
    print('Yo:' + keyword)

    # Обновляю статус задачи на "Выполнено" (если все ок) или не меняю статус и отправляю на повторое выполнение (если не ок)
    if True:
        stream.update_one({'_id': task_id}, {"$set": {"status": True}}, upsert=False)

    # Отключаюсь от базы
    client.close()
    return True

UPD: Более разобранный вариант:

def update_status(task_id, func_result):
    # Соединяюсь с базой
    client = MongoClient('mongodb://localhost:27017/')
    db = client.admetric
    stream = db.stream

    # Обновляю статус задачи на "Выполнено" (если все ок) или не меняю статус и отправляю на повторое выполнение (если не ок)
    if func_result:
        stream.update_one({'_id': task_id}, {"$set": {"status": True}}, upsert=False)

    # Отключаюсь от базы
    client.close()

def yo_func(keyword):
    sleep(10)
    print('Yo:' + keyword)
    return True

def worker_email(keyword, task_id):
    update_status(task_id, yo_func(keyword))

Ответ от товарища enchantner:

Если тебе надо две корутины запустить “параллельно” - делай два инстанса через asyncio.ensure_future(), и поверх них делай asyncio.gather().

Если у тебя цикл в синхронном коде - то внутри цикла делай то же самое, только с шелухой сверху:

loop.run_until_complete(
  asyncio.gather(
       asyncio.ensure_future(first_cor()),
asyncio.ensure_future(second_cor())
  )
)