用户管理 - 第三方用户数据插件实现周期性同步 about bk-user HOT 3 CLOSED

def make_periodic_sync_task(category_id: int, interval_seconds: int):
    """创建同步周期任务"""
    try:
        schedule, _ = IntervalSchedule.objects.get_or_create(every=interval_seconds, period=IntervalSchedule.SECONDS)
    except IntervalSchedule.MultipleObjectsReturned:
        # 兼容目前已有 schedule
        schedule = IntervalSchedule.objects.filter(every=interval_seconds, period=IntervalSchedule.SECONDS)[0]

    # 通过 category_id 来做任务名
    PeriodicTask.objects.get_or_create(
        interval=schedule,
        name=str(category_id),
        task="bkuser_core.categories.tasks.adapter_sync",
        enabled=True,
        kwargs=json.dumps({"instance_id": category_id}),
    )

@receiver(post_category_create)
def create_sync_tasks(sender, category, **kwargs):
    if category.type not in [CategoryType.LDAP.value, CategoryType.MAD.value]:
        return

    logger.info("going to add periodic task for Category<%s>", category.id)
    make_periodic_sync_task(
        category_id=category.id,
        interval_seconds=get_plugin_by_category(category).extra_config["default_sync_period"],
    )