publishing on PyPI

[imtoopunkforyou]

Hello everyone,
First of all, I would like to express my gratitude for all your hard work.
I noticed that there is currently no library available on PyPI. Are you planning on posting one soon?

[imtoopunkforyou]

I was able to test everything during the development process.
My application, built locally, uses valkey (standalone), and we have valkey (cluster) deployed in our booth.
I decided to use the redis and taskiq-redislibraries, despite the fact that we use valkey.
I didn't find any problems either for cron tasks or for regular deferred tasks.
below is a small sample code that you can use if you also have valkey installed instead of redis:

if settings.redis_in_cluster_mode:
    result_backend: AsyncResultBackend[Any] = RedisAsyncClusterResultBackend(
        redis_url=str(settings.redis_url),
    )
    broker: AsyncBroker = ListQueueClusterBroker(
        str(settings.redis_url),
        queue_name=settings.taskiq_queue_name,
    ).with_result_backend(result_backend)
    source = RedisClusterScheduleSource(
        str(settings.redis_url),
        prefix=settings.taskiq_schedule_prefix,
    )
elif not settings.redis_in_cluster_mode:
    result_backend = RedisAsyncResultBackend(
        redis_url=str(settings.redis_url.with_path('/1')),
    )
    broker = ListQueueBroker(
        str(settings.redis_url.with_path('/1')),
        queue_name=settings.taskiq_queue_name,
    ).with_result_backend(result_backend)
    source = RedisScheduleSource(
        str(settings.redis_url.with_path('/1')),
        prefix=settings.taskiq_schedule_prefix,
    )


scheduler = TaskiqScheduler(
    broker,
    sources=[source],
)


async def register_scheduled_tasks() -> None:
    """Register scheduled tasks in the schedule source."""
    scheduled_tasks_config = [
        {
            'task_name': 'proj.tasks.every_min:task_every_min',
            'cron': '*/1 * * * *',
        },
    ]

    # Remove old schedules for tasks we're about to register
    # to avoid duplicates
    existing_schedules = await source.get_schedules()
    task_names_to_register = {config['task_name'] for config in scheduled_tasks_config}
    for schedule in existing_schedules:
        if schedule.task_name in task_names_to_register:
            await source.delete_schedule(schedule.schedule_id)

    # Register new schedules
    for task_config in scheduled_tasks_config:
        scheduled_task = ScheduledTask(
            task_name=task_config['task_name'],
            labels={},
            args=[],
            kwargs={},
            cron=task_config['cron'],
        )
        await source.add_schedule(scheduled_task)

    logger.info(f'Registered {len(scheduled_tasks_config)} scheduled task(s)')


taskiq_fastapi.init(
    broker,
    'proj.application:get_app',
)

# Import all tasks to register them with the broker and scheduler
# This is necessary for the scheduler to know about scheduled tasks
from proj.tasks import every_min  # noqa: E402, F401