Hi. tortoise orm is rasing configuration error inside taskiq.

[KAMRON-DEVELOPER]

my_taskiq.py

@broker.task()
async def distribute_restore_tasks(target: str):
batch_size = 1000
assert target in {"redis", "db"}, ValueError("Target must be either 'redis' or 'db'")
my_logger.debug(f"Starting distribute_restore_tasks with target: {target}")

models: list[tuple[type[Model], str]] = [
    (UserModel, "user:*"),
    (PostModel, "post:*"),
    (PostCommentModel, "post_comment:*"),
    (PostReactionModel, "post_reaction:*"),
    (PostViewModel, "post_view:*"),
    (PostCommentReactionModel, "post_comment_reaction:*"),
    (PostCommentViewModel, "post_comment_view:*"),
]

for model, pattern in models:
    my_logger.info(f"Processing model: {model.__name__} with pattern: {pattern}")
    await distribute_tasks_for_model(model, pattern, target, batch_size)

async def distribute_tasks_for_model(model: type[Model], pattern: str, target: str, batch_size: int):
db = Tortoise.get_connection("default")

total_count = await model.all().using_db(db).count()
my_logger.debug(f"{model.__name__} total count: {total_count}")
rounds = total_count // batch_size
remaining = total_count % batch_size

console where taskiq worker is running

raise ConfigurationError(
tortoise.exceptions.ConfigurationError: default_connection for the model <class 'app.users_app.models.UserModel'> cannot be None

[s3rius]

Can you share more info? How do you initialize database? What do you do on startup events?

[KAMRON-DEVELOPER]

can I get websocket connection? when I print connections it doesn't show websocet connections.

import asyncio
from datetime import datetime
from typing import Annotated, Optional
from uuid import UUID

from taskiq import Context, TaskiqDepends, TaskiqEvents, TaskiqScheduler, TaskiqState
from taskiq.schedule_sources import LabelScheduleSource
from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend, RedisScheduleSource
from tortoise import BaseDBAsyncClient, Model, Tortoise
from tortoise.expressions import F

from app.community_app.models import PostCommentModel, PostCommentReactionModel, PostCommentViewModel, PostModel, PostReactionModel, PostViewModel
from app.services.zepto_service import ZeptoMail
from app.settings.my_config import get_settings
from app.settings.my_redis import cache_manager, my_redis
from app.settings.my_websocket import feed_connection_manager, metrics_connection_manager
from app.users_app.models import UserModel
from app.utility.my_logger import my_logger

settings = get_settings()

broker = ListQueueBroker(
    url=settings.TASKIQ_WORKER_URL,
).with_result_backend(result_backend=RedisAsyncResultBackend(redis_url=settings.TASKIQ_SCHEDULER_URL, result_ex_time=600))

redis_schedule_source = RedisScheduleSource(url=settings.TASKIQ_REDIS_SCHEDULE_SOURCE_URL)

scheduler = TaskiqScheduler(
    broker=broker,
    sources=[LabelScheduleSource(broker=broker), redis_schedule_source],
)


@broker.on_event(TaskiqEvents.WORKER_STARTUP)
async def startup(state: TaskiqState) -> None:
    await Tortoise.init(config=settings.get_tortoise_orm())
    db: BaseDBAsyncClient = Tortoise.get_connection(connection_name="default")
    state.db = db


@broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
async def shutdown(_: TaskiqState) -> None:
    await Tortoise.close_connections()


@broker.task(task_name="send_email_task")
async def send_email_task(to_email: str, username: str, code: str = "0000", for_reset_password: bool = False, for_thanks_signing_up: bool = False):
    my_logger.debug(f"send_email_task is starting")
    zepto = ZeptoMail()
    await zepto.send_email(to_email, username, code, for_reset_password, for_thanks_signing_up)


@broker.task(task_name="broadcast_stats_to_settings_task")
async def broadcast_statistics_to_settings_task() -> None:
    try:
        statistics: dict = await cache_manager.get_statistics()
        await metrics_connection_manager.broadcast(data=statistics)
        my_logger.info("📊 Open statistics broadcasted to settings screen clients.")
    except Exception as e:
        my_logger.critical(f"Exception in broadcast_stats_to_settings_task: {e}")


@broker.task(schedule=[{"cron": "*/60 * * * *"}], task_name="sync_post_stats_task")
async def sync_post_statistics_to_db_task() -> None:
    print("🗓️ process_sync_events is started...")
    await asyncio.sleep(delay=60)
    print("🗓️ process_sync_events is finished...")

    try:
        ready = "not"
        if ready == "ready":
            async with my_redis.pipeline() as _:
                keys = await my_redis.keys("post:*:stats")  # Get all tracked posts
                updates = {}

                for key in keys:
                    post_id = key.split(":")[1]
                    stats = await my_redis.hgetall(key)

                    updates[post_id] = {
                        "views": int(stats.get("views", 0)),
                        "likes": int(stats.get("likes", 0)),
                        "dislikes": int(stats.get("dislikes", 0)),
                    }

                # Bulk update database
                for post_id, data in updates.items():
                    await PostModel.filter(id=post_id).update(
                        views_count=F("views_count") + data["views"],
                        likes_count=F("likes_count") + data["likes"],
                        dislikes_count=F("dislikes_count") + data["dislikes"],
                    )

                # Clear Redis counters after syncing
                await my_redis.delete(*keys)
    except Exception as e:
        print(f"Error updating view count: {e}")


@broker.task(task_name="send_new_post_notification_task")
async def send_new_post_notification_task(user_id: str) -> None:
    user_avatar_url: Optional[str] = await cache_manager.get_profile_avatar_url(user_id=user_id)
    my_logger.debug(f"send_new_post_notification_task user_avatar_url: {user_avatar_url}")
    if user_avatar_url is None:
        user_avatar_url = "defaults/default-avatar.jpg"
    followers: set[str] = await cache_manager.get_followers(user_id=user_id)
    my_logger.debug(f"followers: {followers}")
    await feed_connection_manager.broadcast(user_ids=list(followers), data={"user_avatar_url": user_avatar_url})


@broker.task(task_name="send_new_follower_notification_task")
async def send_new_follower_notification_task(user_id: str) -> None:
    user_avatar_url: Optional[str] = await cache_manager.get_profile_avatar_url(user_id=user_id)
    if user_avatar_url is not None:
        followers: set[str] = await cache_manager.get_followers(user_id=user_id)
        await feed_connection_manager.broadcast(user_ids=list(followers), data={"user_avatar_url": user_avatar_url})


@broker.task()
async def distribute_restore_tasks(target: str, _: Annotated[Context, TaskiqDepends()], batch_size: int = 1000):

    models: list[tuple[type[Model], str, str]] = [
        (UserModel, "user:*", "profile"),
        (PostModel, "post:*", "suffix"),
        (PostCommentModel, "post_comment:*", "suffix"),
        (PostReactionModel, "post_reaction:*", "suffix"),
        (PostViewModel, "post_view:*", "suffix"),
        (PostCommentReactionModel, "post_comment_reaction:*", "suffix"),
        (PostCommentViewModel, "post_comment_view:*", "suffix"),
    ]
    
    
    task_map = {
        "redis": sync_batch_from_db_to_redis,
        "db": sync_batch_from_redis_to_db,
    }

    task = task_map.get(target)
    if not task:
        raise ValueError("Target must be either 'redis' or 'db'")
                    
    for model, pattern, suffix in models:
        total = await model.all().count()
        if target == "redis":
            pattern = pattern[:-1]
        if total > 0:
            for offset in range(0, total, batch_size):
                await task.kiq(model.__name__, pattern, suffix, offset, min(batch_size, total - offset))


@broker.task()
async def sync_batch_from_db_to_redis(model_name: str, pattern: str, suffix: str, offset: int, limit: int):
    my_logger.debug(f"model_name: {model_name}, pattern: {pattern}, offset: {offset}, limit: {limit}")
    model = get_model_by_name(model_name)
    if not model:
        my_logger.error(f"Model not found: {model_name}")
        return

    items = await model.all().offset(offset).limit(limit)
    if not items:
        return

    serialized = [serialize_model_instance(item) for item in items]
    my_logger.debug(f"serialized: {serialized}")
    await cache_manager.bulk_set(pattern, suffix=suffix, items=serialized)


@broker.task()
async def sync_batch_from_redis_to_db(model_name: str, pattern: str, suffix: str, offset: int, limit: int):
    model = get_model_by_name(model_name)
    if not model:
        my_logger.error(f"Model not found: {model_name}")
        return

    redis_data = await cache_manager.bulk_get(pattern, offset, limit)
    if not redis_data:
        return

    # Add your logic to write redis_data back to PostgreSQL using Tortoise ORM


def serialize_model_instance(instance: Model) -> dict:
    return {
        key: (value.hex if isinstance(value, UUID) else value.isoformat() if isinstance(value, datetime) else int(value) if isinstance(value, bool) else value)
        for key, value in instance.__dict__.items()
        if not key.startswith("_") and value is not None
    }


def get_model_by_name(model_name: str) -> type[Model] | None:
    return MODEL_MAPPING.get(model_name)


MODEL_MAPPING: dict[str, type[Model]] = {
    "UserModel": UserModel,
    "PostModel": PostModel,
    "PostCommentModel": PostCommentModel,
    "PostReactionModel": PostReactionModel,
    "PostViewModel": PostViewModel,
    "PostCommentReactionModel": PostCommentReactionModel,
    "PostCommentViewModel": PostCommentViewModel,
}

[s3rius]

No you cannot. Because worker doesn't have an app and therefore doesn't have any client connections. Your web application can have connections, which should be handled on the app side only.

Since you send messages over redis, only data that you send to your worker can be accessible from tasks.

[KAMRON-DEVELOPER]

Hi. I left a comment on github. Thank you.

…

On Mon, Apr 7, 2025 at 7:42 PM Pavel Kirilin ***@***.***> wrote: Can you share more info? How do you initialize database? What do you do on startup events? — Reply to this email directly, view it on GitHub <#440 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/A4RA73LESKWYSNQ2N5WUDA32YKFFFAVCNFSM6AAAAAB2ROPJF2VHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTENZVGIZTMOI> . You are receiving this because you authored the thread.Message ID: ***@***.***>

[KAMRON-DEVELOPER]

Ok, I understand.

…

On Wed, Apr 9, 2025 at 11:25 PM Pavel Kirilin ***@***.***> wrote: No you cannot. Because worker doesn't have an app and therefore doesn't have any client connections. Your web application can have connections, which should be handled on the app side only. Since you send messages over redis, only data that you send to your worker can be accessible from tasks. — Reply to this email directly, view it on GitHub <#440 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/A4RA73OJE5BJTLGLHN5TXS32YVQZ5AVCNFSM6AAAAAB2ROPJF2VHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTENZYGIZDCNY> . You are receiving this because you authored the thread.Message ID: ***@***.***>