how to use structlog to output

[johums-best]

taskiq can use structlog to output ? i have no idea to resolve it.
this is celery code example , i can output it like json

@signals.setup_logging.connect
def on_setup_logging(**kwargs):
    setup_logging(json_logs=LOG_JSON_FORMAT, log_level=setting.LOG_LEVEL)


@signals.task_prerun.connect
def on_task_prerun(sender, task_id, task, args, kwargs, **_):
    structlog.contextvars.bind_contextvars(task_id=task_id, task_name=task.name)

[sector119]

I'm trying to use middleware for that, "pre_send" is called, but taskiq_task_id and taskiq_task_name doesn't appear at json log (

class StructlogMiddleware(TaskiqMiddleware):
    def pre_send(
        self,
        message: TaskiqMessage,
    ) -> (
        TaskiqMessage | typing.Coroutine[typing.Any, typing.Any, TaskiqMessage]
    ):
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            taskiq_task_id=message.task_id, taskiq_task_name=message.task_name
        )

        return message

broker = ListQueueBroker(settings.TASKIQ_BACKEND_URL)

middlewares = [
    StructlogMiddleware(),
]

broker.add_middlewares(*middlewares)

[sector119]

oops, but it works as expected if I put all that stuff at pre_execute, not pre_send method!

{"event":"Syncing payments...","taskiq_task_name":"billing:sync-payments","taskiq_task_id":"43ad252452d4491fa2c941408f9ed0b1","logger":"epsilon-tasks","level":"info","timestamp":"2024-02-19 16:00:23"}

[s3rius]

Can you provide with more details about how does it work for future?

[sector119]

https://gist.github.com/sector119/7461d181830ec1232c242e5aa8110152

Here I load scheduled tasks from tasks.yaml config file, that is useful if you use docker, so you can easily change cron settings, add or remove existing tasks

And run it with
taskiq worker epsilon_tasks.main:broker -tp "epsilon_tasks/tasks/**/*.py" -fsd
taskiq scheduler epsilon_tasks.main:get_scheduler

Those env variables must be exported before worker and scheduler start
EPSILON_TASKS_ENV
EPSILON_TASKS_CONFIG
EPSILON_TASKS_LOG_CONFIG

get_scheduler is function that returns TaskiqScheduler, to use it I had to change scheduler/run.py a bit

if isinstance(args.scheduler, str):
    scheduler = import_object(args.scheduler)

    if scheduler is types.FunctionType:
        scheduler = await scheduler()
else:
    scheduler = args.scheduler

[sector119]

@s3rius maybe it would be possible to add ability to pass scheduler not only as variable but as function that returns it ?
So we can dynamically add existing tasks and use taskiq scheduler command

Or I have to subclass TaskiqScheduler and load my tasks at startup method ?

[sector119]

import typing

from taskiq import TaskiqScheduler
from taskiq.abc.broker import AsyncBroker
from taskiq.abc.schedule_source import ScheduleSource

from .loader import Task


class DynamicTaskiqScheduler(TaskiqScheduler):
    def __init__(
        self,
        broker: AsyncBroker,
        source: ScheduleSource,
        tasks: typing.Iterable[Task],
    ):
        super().__init__(broker, [source])

        self.tasks = tasks

    async def startup(self) -> None:
        for task in self.tasks:
            await task.callable.schedule_by_cron(
                self.sources[0], task.schedule, *task.args, **task.kwargs
            )

        await super().startup()