def rotate_endpoint()

in lemur/common/celery.py [0:0]

• 40 lines of code
• 6 McCabe index (conditional complexity)

def rotate_endpoint(self, endpoint_id, **kwargs):
    function = f"{__name__}.{sys._getframe().f_code.co_name}"
    logger = logging.getLogger(function)

    endpoint = endpoint_service.get(endpoint_id)

    if not endpoint:
        logger.info(f"Skipping rotation,due to {endpoint_id} did not exist")
        return

    old_certificate_id = endpoint.certificate.id

    remove_cert_args = (endpoint_id, old_certificate_id)
    delay_before_removal = current_app.config.get(
        "CELERY_ROTATE_ENDPOINT_DELAY_BEFORE_DETACH", 60
    )
    if is_task_scheduled(rotate_endpoint_remove_cert.name, remove_cert_args):
        # the remove task has already been scheduled so we skip this turn
        logger.info(
            f"{rotate_endpoint_remove_cert.name}{str(remove_cert_args)} already scheduled."
        )
        return

    new_cert = endpoint.certificate.replaced[0]
    new_cert_name = new_cert.name

    if self.request.retries > 0:
        extra_message = f"retry {self.request.retries} of {self.max_retries}"
    else:
        extra_message = None

    logger.info(f"Attaching {new_cert_name} to {endpoint.name}")

    # update with redis lock
    # will raise redis.exceptions.LockError Unable to acquire lock within the time specified
    with red.lock(endpoint.name.rsplit("/", 1)[0], blocking_timeout=10):
        endpoint.source.plugin.update_endpoint(endpoint, new_cert_name)

    # send notification taking notifications from both new and old certificate
    send_notifications(
        list(set(endpoint.certificate.notifications + new_cert.notifications)),
        "rotation",
        extra_message,
        endpoint=endpoint,
    )
    # schedule a task to remove the old certificate
    logger.info(
        f"Scheduling {rotate_endpoint_remove_cert.name}{str(remove_cert_args)} to execute in {delay_before_removal} seconds."
    )
    rotate_endpoint_remove_cert.apply_async(
        remove_cert_args, countdown=delay_before_removal
    )

    # sync source
    if not is_task_scheduled(sync_source, (endpoint.source.label,)):
        sync_source.delay(endpoint.source.label)