Function blob_regenerate_url

shared/url_generation.py:278–324

(blob_object: DiffgramBlobObjectType,
                        session: Session,
                        connection_id: int = None,
                        bucket_name: str = None,
                        access_token: str = None,
                        reference_file: File = None)

def blob_regenerate_url(blob_object: DiffgramBlobObjectType,
                        session: Session,
                        connection_id: int = None,
                        bucket_name: str = None,
                        access_token: str = None,
                        reference_file: File = None) -> list[object, dict]:

    if not blob_object.url_signed_blob_path:
        return

    strategy = determine_url_regenerate_strategy(
        connection_id = connection_id,
        bucket_name = bucket_name)

    should_regenerate, new_offset_in_seconds = data_tools.determine_if_should_regenerate_url(blob_object, session)
    if should_regenerate is not True and strategy != 'connection':
        return

    logger.debug(f"Regenerating with {strategy} strategy")

    if strategy == "default":
        blob_object, log = default_url_regenerate(
            session = session,
            blob_object = blob_object,
            new_offset_in_seconds = new_offset_in_seconds
        )

    if strategy == "connection":
        logger.debug(f'Generate Signed Url with connection {connection_id} on bucket {bucket_name}')
        blob_object, log = connection_url_regenerate(
            session = session,
            blob_object = blob_object,
            connection_id = connection_id,
            bucket_name = bucket_name,
            new_offset_in_seconds = new_offset_in_seconds,
            reference_file = reference_file,
            access_token = access_token
        )

    if regular_log.log_has_error(log):
        logger.error(f'Failed to regenerate Blob URL {log}')
        blob_object.url_signed = None
        blob_object.error = str(log)
        session.add(blob_object)
        return blob_object, log

    return blob_object, log


def determine_url_regenerate_strategy(connection_id,