def LeaseFlowForProcessing(self,
                             client_id,
                             flow_id,
                             processing_time,
                             cursor=None):
    """Marks a flow as being processed on this worker and returns it."""
    query = ("SELECT " + self.FLOW_DB_FIELDS +
             "FROM flows WHERE client_id=%s AND flow_id=%s")
    cursor.execute(
        query,
        [db_utils.ClientIDToInt(client_id),
         db_utils.FlowIDToInt(flow_id)])
    response = cursor.fetchall()
    if not response:
      raise db.UnknownFlowError(client_id, flow_id)

    row, = response
    rdf_flow = self._FlowObjectFromRow(row)

    now = rdfvalue.RDFDatetime.Now()
    if rdf_flow.processing_on and rdf_flow.processing_deadline > now:
      raise ValueError("Flow %s on client %s is already being processed." %
                       (client_id, flow_id))

    if (rdf_flow.parent_hunt_id is not None and
        # TODO(user): remove the check for a legacy hunt prefix as soon as
        # AFF4 is gone.
        not rdf_flow.parent_hunt_id.startswith("H:")):

      query = "SELECT hunt_state FROM hunts WHERE hunt_id=%s"
      args = [db_utils.HuntIDToInt(rdf_flow.parent_hunt_id)]
      rows_found = cursor.execute(query, args)
      if rows_found == 1:
        hunt_state, = cursor.fetchone()
        if (hunt_state is not None and
            not rdf_hunt_objects.IsHuntSuitableForFlowProcessing(hunt_state)):
          raise db.ParentHuntIsNotRunningError(client_id, flow_id,
                                               rdf_flow.parent_hunt_id,
                                               hunt_state)

    update_query = ("UPDATE flows SET "
                    "processing_on=%s, "
                    "processing_since=FROM_UNIXTIME(%s), "
                    "processing_deadline=FROM_UNIXTIME(%s) "
                    "WHERE client_id=%s and flow_id=%s")
    processing_deadline = now + processing_time
    process_id_string = utils.ProcessIdString()

    args = [
        process_id_string,
        mysql_utils.RDFDatetimeToTimestamp(now),
        mysql_utils.RDFDatetimeToTimestamp(processing_deadline),
        db_utils.ClientIDToInt(client_id),
        db_utils.FlowIDToInt(flow_id)
    ]
    cursor.execute(update_query, args)

    # This needs to happen after we are sure that the write has succeeded.
    rdf_flow.processing_on = process_id_string
    rdf_flow.processing_since = now
    rdf_flow.processing_deadline = processing_deadline
    return rdf_flow