Function get_custom_url_supported_connector

shared/url_generation.py:165–198

Gets the connector object and checks if it supports custom signed urls with a custom service. :param session: :param log: :param connection_id: :param connection: :return:

(session: Session, log: dict, connection_id: int)

def get_custom_url_supported_connector(session: Session, log: dict, connection_id: int) -> [object, dict]:
    """
        Gets the connector object and checks if it supports custom signed urls with a custom service.
    :param session:
    :param log:
    :param connection_id:
    :param connection:
    :return:
    """
    connection = Connection.get_by_id(session = session, id = connection_id)
    if connection is None:
        msg = f'connection id: {connection_id} not found.'
        log['error']['connection_id'] = msg
        logger.error(msg)
        return None, log
    if connection.integration_name not in ALLOWED_CONNECTION_SIGNED_URL_PROVIDERS:
        msg = f'Unsupported connection provider for URL regeneration {connection.id}:{connection.integration_name}'
        log['error']['unsupported'] = msg
        log['error']['supported_providers'] = ALLOWED_CONNECTION_SIGNED_URL_PROVIDERS
        logger.error(msg)
        return None, log

    connection_strategy = ConnectionStrategy(
        connection_id = connection_id,
        session = session)

    client, success = connection_strategy.get_connector(connection_id = connection_id)
    if not success:
        msg = f'Failed to get connector for connection {connection_id}'
        log['error']['connector'] = msg
        logger.error(msg)
        return None, log

    return client, log