In this blog post, we'll be looking at a piece of code that verifies email addresses in batches. This can be useful in a variety of scenarios, such as cleaning up a database of email addresses or checking the validity of a list of email addresses that have been collected through a sign-up form.

The Code

The code is written in Python, and it uses the py3-validate-emailpackage to perform the email verification. Here's the code:

import csv
from io import StringIO
from os.path import exists
from validate_email import validate_email_or_fail, validate_email
from redis import Redis
from rq import Queue, Retry

r = Redis()
q = Queue(connection=r)


class VerifyEmails:
    def batch_emails(self, emails, batch_size, filename, with_smtp=False):
        batches_to_run = []
        for i in range(0, len(emails), batch_size):
            batch = emails[i:i + batch_size]
            print(batch)
            job = q.enqueue(self.queued_batch, batch, filename, with_smtp, job_timeout='20m', retry=Retry(max=3))
            batches_to_run.append(job.id)
        return batches_to_run

    async def queued_batch(self, emails, filename, with_smtp=False):
        print('about to verify batch')
        result = await self.verify(emails, with_smtp)

        path = 'validated/validated-%s.csv' % filename
        converted_list = []

        if not exists(path):
            converted_list.append(result[0].keys())

        for r in result:
            converted_list.append(r.values())
        si = StringIO()
        cw = csv.writer(si)
        cw.writerows(converted_list)
        with open(path, 'a+') as file:
            file.write(si.getvalue())

    async def verify(self, emails, with_smtp=False):
        # resolver = caching_resolver(timeout=10)
        response = []
        for email in emails:
            print("trying %s" % email)
            try:
                # Check that the email address is valid.
                # validation = validate_email(email, check_deliverability=True, dns_resolver=resolver)
                domain_name = email.split('@')[1]
                fake_address = "thisaddresshsouldnotexist@%s" % domain_name
                is_valid = validate_email_or_fail(
                    email_address=email,
                    check_format=True,
                    check_blacklist=True,
                    check_dns=True,
                    dns_timeout=10,
                    check_smtp=with_smtp,
                    smtp_timeout=10,
                    smtp_helo_host='my.host.name',
                    smtp_from_address='[email protected]',
                    smtp_skip_tls=False,
                    smtp_tls_context=None,
                    smtp_debug=False)

                is_catch_all = None
                if with_smtp:
                    is_catch_all = validate_email(
                        email_address=fake_address,
                        check_format=False,
                        check_blacklist=False,
                        check_dns=False,
                        dns_timeout=10,
                        check_smtp=True,
                        smtp_timeout=10,
                        smtp_helo_host='my.host.name',
                        smtp_from_address='[email protected]',
                        smtp_skip_tls=False,
                        smtp_tls_context=None,
                        smtp_debug=False)

                response.append({
                    'email': email,
                    'message': "",
                    'valid': is_valid,
                    'isCatchAll': is_catch_all
                })
            except Exception as e:
                response.append({
                    'email': email,
                    'message': e,
                    'valid': False,
                    'isCatchAll': None
                })
                print(e)

        return response

How Batching Works

The batch_emails method is the main entry point for the class. It takes a list of email addresses, a batch size, and a filename as input, and splits the list of emails into smaller batches of the specified size.

The batch_emails method then queues each batch of email addresses to be verified using the queued_batch method, which is used to create a queue of jobs that can be run in parallel. This can improve the performance of the email verification process by allowing multiple batches of emails to be verified at the same time.

The queued_batch method also writes the verified email addresses to a CSV file, which is saved in the validated directory. The filename is generated by appending the provided filename to the string "validated-".

The verify method is where the actual email verification happens. It uses the validate_email_or_fail function from the validate_email package to check whether the email addresses are valid. This function checks several different things, such as whether the email address has the correct format, whether the domain exists and has a valid DNS record and whether it's SMTP server accepts mail for it.

The verify method also checks whether the email address is a catch-all address. This is an email address that receives all emails sent to a domain, regardless of whether the recipient actually exists. To check this, the method sends a fake email to the domain and checks if it gets accepted by the email server. This is useful because catch-all addresses can be problematic for email senders because it can lead to a high rate of undeliverable emails, which can negatively impact the sender's reputation and deliverability.

Adding Tests

Lets add some tests:

import asyncio
import unittest
from unittest.mock import patch

from main import VerifyEmails, q


class TestStringMethods(unittest.TestCase):
    def test_batch_emails_method_exists(self):
        verify_emails = VerifyEmails()
        self.assertTrue(hasattr(verify_emails, "batch_emails"))
        method = getattr(verify_emails, "batch_emails")
        self.assertEqual(method.__code__.co_argcount, 5)

    def test_batch_emails_enqueues_job(self):
        verify_emails = VerifyEmails()
        emails = ["[email protected]", "[email protected]"]
        batch_size = 1
        filename = "test_file"
        with_smtp = False
        result = verify_emails.batch_emails(emails, batch_size, filename, with_smtp)
        self.assertIsNotNone(result)
        self.assertEqual(len(result), 2)
        self.assertEqual((["[email protected]"], "test_file", False), q.fetch_job(result[0]).args)
        for job_id in result:
            job = q.fetch_job(job_id)
            self.assertIsNotNone(job)
            self.assertEqual(job.func_name, "queued_batch")
            self.assertLessEqual(batch_size, len(job.args[0]))

    @patch.object(VerifyEmails, "verify")
    def test_queued_batch_calls_verify_method(self, mock_verify):
        verify_emails = VerifyEmails()
        emails = ["[email protected]", "[email protected]"]
        filename = "test_file"
        with_smtp = False
        asyncio.run(verify_emails.queued_batch(emails, filename, with_smtp))
        mock_verify.assert_called_once_with(emails, with_smtp)

    def test_verify_method_returns_expected_keys(self):
        verify_emails = VerifyEmails()
        emails = ["[email protected]", "[email protected]"]
        with_smtp = False
        result = asyncio.run(verify_emails.verify(emails, with_smtp))
        self.assertIsNotNone(result)
        self.assertEqual(len(result), 2)
        for email_dict in result:
            self.assertIsInstance(email_dict, dict)

• test_batch_emails_method_exists checks if the VerifyEmails class has a batch_emails method and if the method takes the correct number of arguments.
• test_batch_emails_enqueues_job checks if the batch_emails method returns the correct number of jobs, and if the jobs are enqueued with the correct arguments.
• test_queued_batch_calls_verify_method checks if the queued_batch method calls the verify method with the correct arguments.
• test_verify_method_returns_expected_keys checks if the verify method returns the expected keys in the resulting dictionaries.

In summary, the VerifyEmails class is a simple but effective way to verify a large number of email addresses in batches. By using the rq package to queue the email verification jobs, it can run the email verification process in parallel, making it more efficient and scalable. It also provides the ability to save the verified email addresses to a CSV file, which can be useful in a variety of scenarios.

A Limit Worth Knowing

💡

It should be noted, that verifying too many emails this way will get the host IP address flagged by spam lists like spamhaus.

Exposing It as an API

In addition to the VerifyEmails class that we should also include an API that provides a simple way to use the email verification functionality. The API is implemented using the Flask framework, and it exposes several endpoints that can be called using HTTP requests.

import datetime
import uuid

from flask import Flask, request, send_from_directory, jsonify
from redis import Redis
from rq import Queue
from rq.exceptions import NoSuchJobError
from rq.job import Job
from flask_cors import CORS
from main import VerifyEmails

r = Redis()
q = Queue(connection=r)
app = Flask(__name__)

CORS(app)


@app.route('/batched-validation/<int:batch_size>', methods=['POST'])
def batch(batch_size):
    request_data = request.get_json()
    emails = request_data['emails']
    with_smtp = request_data['withSmtp']
    filename = str(uuid.uuid4())
    jobs = VerifyEmails().batch_emails(emails, batch_size, filename, with_smtp)
    return jsonify({"task": {
        "filename": 'validated-' + filename + '.csv',
        "enqueuedAt": datetime.datetime.now(),
        "jobs": jobs
    }}), 202


@app.route('/job/status', methods=['POST'])
def get_job_status():
    request_data = request.get_json()
    jobs = request_data
    status = []
    for job in jobs:
        try:
            s = Job.fetch(job, connection=r).get_status()
            status.append({"job": job, "status": s})
        except NoSuchJobError as e:
            status.append({job: 'none'})
    return jsonify(status)


@app.route("/validated/<path:name>")
def download_file(name):
    return send_from_directory(
        'validated', name, as_attachment=True
    )


@app.route('/')
def home():
    return "hello"


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

The first endpoint is /batched-validation/<int:batch_size>, which is used to start the email verification process. It takes a batch_size parameter that specifies the size of the batches in which the email addresses should be verified. It also takes a JSON object as the request body, which contains the list of email addresses to be verified and a withSmtp flag that indicates whether the smtp and catch-all checks should be performed.

The batch function then creates an instance of the VerifyEmails class and calls the batch_emails method to start the email verification process. It also generates a unique filename for the CSV file that will be used to save the verified email addresses. The function returns a JSON object that contains the filename and the list of job IDs that were enqueued for email verification.

The second endpoint is /job/status, which can be used to check the status of the email verification jobs. It takes a JSON object as the request body, which contains a list of job IDs. The get_job_status function then retrieves the status of each job using the rq package, and returns a JSON object that contains the job ID and the corresponding status.

The third endpoint is /validated/<path:name>, which can be used to download the CSV file that contains the verified email addresses. It takes the filename as a path parameter, and the download_file function uses the Flask send_from_directory function to serve the file to the client.

In summary, the VerifyEmails class is a simple but effective way to verify a large number of email addresses in batches. By using the rq package to queue the email verification jobs, it can run the email verification process in parallel, making it more efficient and scalable. It also provides the ability to save the verified email addresses to a CSV file. The API that is included in the code provides a convenient way to use the email verification functionality, and the tests ensure that the code is working as expected. Overall, this is a great example of how Python can be used to solve a common problem in a efficient and straightforward way.