def check_suppressed_email_confirmation()

in src/olympia/users/utils.py [0:0]

• 65 lines of code
• 11 McCabe index (conditional complexity)

def check_suppressed_email_confirmation(verification, page_size=5):
    from olympia.users.models import SuppressedEmailVerification

    assert_socket_labs_settings_defined()

    assert isinstance(verification, SuppressedEmailVerification)

    email = verification.suppressed_email.email

    current_count = 0
    total = 0

    code_snippet = str(verification.confirmation_code)[-5:]
    path = f'servers/{settings.SOCKET_LABS_SERVER_ID}/reports/recipient-search/'

    # socketlabs might set the queued time any time of day
    # so we need to check to midnight, one day before the verification was created
    # and to midnight of tomorrow
    before = verification.created - datetime.timedelta(days=1)
    start_date = datetime.datetime(
        year=before.year,
        month=before.month,
        day=before.day,
    )
    end_date = datetime.datetime.now() + datetime.timedelta(days=1)
    date_format = '%Y-%m-%d'

    params = {
        'toEmailAddress': email,
        'startDate': start_date.strftime(date_format),
        'endDate': end_date.strftime(date_format),
        'pageNumber': 0,
        'pageSize': page_size,
        'sortField': 'queuedTime',
        'sortDirection': 'dsc',
    }

    is_first_page = True

    found_emails = []

    while current_count < total or is_first_page:
        if not is_first_page:
            params['pageNumber'] = params['pageNumber'] + 1

        url = (
            urllib.parse.urljoin(settings.SOCKET_LABS_HOST, path)
            + '?'
            + urllib.parse.urlencode(params)
        )

        headers = {
            'authorization': f'Bearer {settings.SOCKET_LABS_TOKEN}',
        }

        utils_log.info(f'checking for {code_snippet} with params {params}')

        response = requests.get(url, headers=headers)
        response.raise_for_status()
        json_data = response.json()

        utils_log.info(f'recieved data {json_data} for {code_snippet}')

        if is_first_page:
            total = json_data['total']

            if total == 0:
                return found_emails

            is_first_page = False

        data = json_data['data']
        current_count += len(data)

        utils_log.info(f'found emails {data} for {code_snippet}')

        ## TODO: check if we can set `customMessageId` to replace code snippet
        for item in data:
            found_emails.append(
                {
                    'from': item['from'],
                    'to': item['to'],
                    'subject': item['subject'],
                    'status': item['status'],
                    'statusDate': item['statusDate'],
                }
            )

            if code_snippet in item['subject'] and item['status'] == 'Delivered':
                verification.mark_as_delivered()

    return found_emails