def sync_source(source):
    """
    This celery task will sync the specified source.

    :param source:
    :return:
    """

    function = f"{__name__}.{sys._getframe().f_code.co_name}"
    task_id = None
    if celery_app.current_task:
        task_id = celery_app.current_task.request.id

    log_data = {
        "function": function,
        "message": "Syncing source",
        "source": source,
        "task_id": task_id,
    }

    if task_id and is_task_active(function, task_id, (source,)):
        log_data["message"] = "Skipping task: Task is already active"
        current_app.logger.debug(log_data)
        return

    current_app.logger.debug(log_data)
    try:
        sync([source], current_app.config.get("CELERY_ENDPOINTS_EXPIRE_TIME_IN_HOURS", 2))
        metrics.send(
            f"{function}.success", "counter", 1, metric_tags={"source": source}
        )
    except SoftTimeLimitExceeded:
        log_data["message"] = "Error syncing source: Time limit exceeded."
        current_app.logger.error(log_data)
        capture_exception()
        metrics.send(
            "sync_source_timeout", "counter", 1, metric_tags={"source": source}
        )
        metrics.send("celery.timeout", "counter", 1, metric_tags={"function": function})
        return

    log_data["message"] = "Done syncing source"
    current_app.logger.debug(log_data)
    metrics.send(f"{function}.success", "counter", 1, metric_tags={"source": source})
    return log_data

    This celery task will sync the specified source.

    :param source:
    :return:
    
lemur/common/celery.py