1. Service initialization

Deduplication I - Service initialization

1. Introduction
2. Service initialization
   1. Get basic information on request origin
   2. Prepare namespace for compartimentalization
3. Custom error handling
4. Sending requests to the API
   1. General request process
   2. Request headers
      1. content_type
   3. Query-string arguments
      1. email
      2. action
      3. duplicates
      4. id and field headers
   4. Partial duplicate fields
   5. Sending the file itself
5. Storing the original file in Cloud Storage
6. Launching the workers
7. Retrieving the immediate response

Introduction

This document provides a detailed overview of the first step of the deduplication service: uploading the users' file to Google Cloud Storage for later parsing.

The code that handles this part can be found in the dedupe.DedupeAPI module.

Service initialization

Upon initialization, the service does two key things: get basic information on the request origin and prepare the namespace for compartimentalization.

This code is wrapped in the __init__ method.

Get basic information on request origin

Some headers will be sent directly in the request, dealing basically with file type. But some others are global to the request, and are gathered at this point. Specifically, coordinates, country and user-agent of request origin. The two first are provided by Google App Engine. The last is a standard header in all HTTP requests.

        # Get request headers
        self.cityLatLong = request.headers.get('X-AppEngine-CityLatLong')
        self.country = request.headers.get('X-AppEngine-Country')
        self.user_agent = request.headers.get('User-Agent')

Prepare namespace for compartimentalization

Deduping is a fragile task, and in order to provide consistent concurrency, requests must be effectively isolated. Otherwise, records from one dataset can be flagged as duplicates in a different dataset.

Google App Engine provides namespaces to isolate requests, memcache operations and database queries, so inter-request independence is assured. The service does not need any particular schema for namespace naming, but it does need the namespace to be unique for each request. Therefore, UUIDs are used as namespace identifiers. Also, the "default" namespace identifier is stored for later reference.

        # Handle namespace for this request
        self.previous_namespace = namespace_manager.get_namespace()
        self.request_namespace = str(uuid.uuid4())

Custom error handling

A special function, _err is used for consistent error notification. This is wrapped in the _err method.

    def _err(self, err_code=500, err_message="", err_explain=""):
        self.error(err_code)
        resp = {
            "status": "error",
            "error": err_message,
            "message": err_explain
        }
        logging.error(err_message)
        logging.error(err_explain)
        self.response.headers['Content-Type'] = "application/json"
        self.response.write(json.dumps(resp)+"\n")
        return

Sending requests to the API

General request process

It all begins with a POST request to the API URL. Currently (as of December 20th 2016), the base URL is:

http://dedupe.vertnet-portal.appspot.com/api/v0/dedupe

NOTE: Currently, only the development version is available, so the actual URL is dev.dedupe.vertnet-portal...

If a GET request is attempted, the API returns a 405 error ("Method not allowed").

    def get(self):
        err_message = "Method not allowed"
        err_explain = "Only POST requests are allowed"
        self._err(405, err_message, err_explain)
        return

All the rest of the code for this part can be found in the post method.

Request headers

content_type

This header must be specified so that the API knows how to handle the included file. The default content-type, which in common HTML forms and curl operations is application/x-www-form-urlencoded is not valid, since this does not represent a file object, but rather a data from an urlencoded form.

        # Determine file format via 'Content-Type'
        self.content_type = self.request.headers['Content-Type']
        logging.info("Content-Type: %s" % self.content_type)

        if self.content_type == "application/x-www-form-urlencoded":
            err_explain = "'Content-Type' is a required header for the" \
                          " proper working of the API." \
                          " Please read the documentation for examples on" \
                          " how to set this parameter"
            self._err(400, "No 'Content-Type' was provided", err_explain)
            return

According to the content-type, the service will determine the dialect of the csv reader:

        # Establish field separator based on Content-Type
        if self.content_type == "text/csv":
            self.delimiter = ","
            self.extension = "csv"
        elif self.content_type == "text/tab-separated-values":
            self.delimiter = "\t"
            self.extension = "txt"
        else:
            err_explain = "The value of 'Content-Type' is not among the" \
                          " accepted values for this header. Should be one" \
                          " of: %s" % ", ".join(ALLOWED_TYPES)
            self._err(400, "Wrong 'Content-Type' header", err_explain)
            return

The list of allowed types can be found in the config module. Currently, the list of ALLOWED_TYPES is:

ALLOWED_TYPES = ["text/csv", "text/tab-separated-values"]

Query-string arguments

email

The Dedupe API is, in essence, an asynchronous task. Therefore, the user cannot know immediately if the process finished or not. An email is required so that the user receives notifications and links during the process.

        # Check email exists in parameters
        self.email = self.request.get("email", None)
        if self.email is None:
            self._err(400, "Please provide an email address")
            return
        logging.info("Results will be sent to %s" % self.email)

action

This parameter indicates the action to perform on the detected duplicates. Descriptions of the available actions can be found in the "Handling actions" page. flag is the default action.

        # Determine action ("flag" by default)
        self.action = self.request.get("action", "flag")
        if self.action not in ALLOWED_ACTIONS:
            err_explain = "Action %s is not valid. Should be one of: %s" % (
                self.action, ", ".join(ALLOWED_ACTIONS))
            self._err(400, "Action not allowed", err_explain)
            return
        logging.info("Action: %s" % self.action)

The list of available actions can be found in the config module. Currently, the value of the ALLOWED_ACTIONS variable is:

ALLOWED_ACTIONS = ["report", "flag", "remove"]

duplicates

The service can detect two types of duplicates, depending on the duplicated content. strict duplicates are those that have all the values of all the fields identical. partial duplicates are those that have the values of certain key fields identical. The service can try to detect only strict duplicates, only partial duplicates or all (this last by default).

        # Determine duplicate types to be checked ("all" by default)
        self.duplicates = self.request.get("duplicates", "all")
        if self.duplicates not in ALLOWED_DUPLICATES:
            err_explain = "Value of 'duplicates' parameter %s is not valid."
            err_explain += " Should be one of: %s"
            err_explain = err_explain % (
                self.duplicates, ", ".join(ALLOWED_DUPLICATES)
            )
            self._err(400, "Duplicate detection type not allowed", err_explain)
            return
        logging.info("Looking for %s duplicates" % self.duplicates)

The list of available duplicate types can be found in the config module. Currently, the value of the ALLOWED_DUPLICATES is:

ALLOWED_DUPLICATES = ["strict", "partial", "all"]

id and field headers

All records should have an id field to uniquely identify them, at least within the dataset. This is especially useful for identifying duplicates (record X is duplicate of record Y), but not so much for removing them.

By default, the service follows the DarwinCore standard recommendations and will look for an id field and, if not present, an occurrenceid field. However, users can specify the name of the field that will act as an id in the id parameter.

To do this, the service first "sniffs" the headers and stores a lowercase version of them. Also, this sniffing can check if the provided content-type matches the actual file type: if the length of the list object that has the headers allegedly separated by the delimiter character is 1, it means the specified delimiter is not the actual delimiter, so the content-type is wrong.

        # Sniff headers
        self.reader = csv.reader(self.file, delimiter=self.delimiter)
        self.headers = self.reader.next()
        self.headers_lower = [x.lower() for x in self.headers]

        # Check if proper field delimiter
        if len(self.headers) == 1:
            err_explain = "The system ended up with 1-field rows. Please" \
                          " check the 'Content-Type' parameter"
            self._err(400, "Wrong 'Content-Type' header", err_explain)
            return

If no id is specified and none of the default values is found, the process will continue, but all id-related information will be omitted.

        # Check "id" parameter
        self.id_field = self.request.get("id", None)
        # If not given
        if self.id_field is None:
            # Find "id" field
            if 'id' in self.headers_lower:
                self.id_field = 'id'
            # Otherwise find "occurrenceid" field
            elif 'occurrenceid' in self.headers_lower:
                self.id_field = 'occurrenceid'
            # Otherwise, show warning and don't show "id"-related info
            else:
                warning_msg = "No 'id' field could be determined"
                self.warnings.append(warning_msg)
                logging.warning(warning_msg)
                self.id_field = None
        # Otherwise, check if field exists in headers
        elif self.id_field.lower() not in self.headers_lower:
            self._err(400, "Couldn't find field '%s'" % self.id_field)
            return

Partial duplicate fields

As said earlier, the service can detect strict (full) duplicates and partial duplicates. These partial duplicates are based on four key fields, here shown as their DarwinCore standard term.

• locality
• scientificName
• recordedBy
• eventDate

So, if two records refer to the same taxon, were collected in the same time and place and by the same person, the service assumes they are the same record.

        # Get positions for partial duplicates
        self.loc = self.headers_lower.index(LOC.lower())
        self.sci = self.headers_lower.index(SCI.lower())
        self.col = self.headers_lower.index(COL.lower())
        self.dat = self.headers_lower.index(DAT.lower())

The list of key fields can be found in the config module. Currently, the value of the LOC, SCI, COL and DAT variables are:

# Names of default fields for partial duplicate detection
LOC = "locality"
SCI = "scientificName"
COL = "recordedBy"
DAT = "eventDate"

Sending the file itself

The file in which duplicates will be detected must be sent in the body of the request. To properly get the contents of the file, the two-step process requires: (a) get the file-type object from the request and, (b) extract the file content itself.

        # Get content from request body
        self.body_file = self.request.body_file
        self.file = self.body_file.file

Storing the original file in Cloud Storage

The submitted file will be uploaded to a specific bucket in Google Cloud Storage. From there, service tasks will take it and parse the content for duplicates. The path to the file includes the UUID used as namespace identifier, so cross-request isolation is assured.

        # Store original file in GCS
        self.file_path = "/".join(["", BUCKET, self.request_namespace])
        self.file_name = "%s/orig.%s" % (self.file_path, self.extension)
        try:
            f = gcs.open(self.file_name, 'w', content_type=self.content_type)
            logging.info("File %s created" % self.file_name)
            f.write(self.file.read())
            logging.info("Successfully wrote file to GCS")
            f.close()
            logging.info("File closed")
        except Exception, e:
            logging.error("Something went wrong opening the file:\n"
                          "f: %s\nerror: %s" % (self.file_name, e))

Launching the workers

The work itself is not done here. This handler only takes the file, its associated parameters and prepares the content so that the workers can scan the document and extract the duplicates. The workers are called asynchronously and all available parameters are sent so that they can perform their job.

        # Launch async task with parameters
        params = {
            "latlon": self.cityLatLong,
            "country": self.country,
            "user_agent": self.user_agent,
            "email": self.email,
            "request_namespace": self.request_namespace,
            "previous_namesapce": self.previous_namespace,
            "content_type": self.content_type,
            "delimiter": self.delimiter,
            "extension": self.extension,
            "action": self.action,
            "duplicates": self.duplicates,
            "file_path": self.file_path,
            "file_name": self.file_name,
            "headers": json.dumps(self.headers),
            "loc": self.loc,
            "sci": self.sci,
            "dat": self.dat,
            "col": self.col,
            "id_field": self.id_field
        }
        taskqueue.add(
                url=TASKURL,
                params=params
            )

Retrieving the immediate response

The workers will need some time to finish parsing the file. But meanwhile, the user receives a message indicating whether or not this initial step finished successfully.

        # Build response
        msg = "De-duplication successfully initiated. Please check your email"
        msg += " address for notifications"
        resp = {
            "status": "success",
            "message": msg,
            "email": self.email
        }
        self.response.headers['Content-Type'] = "application/json"
        self.response.write(json.dumps(resp)+"\n")
        return