Reads, locks, and updates completed requests.

def _ReadLockAndUpdateCompletedRequests(self, request_keys, response_counts,
                                          cursor):
    """Reads, locks, and updates completed requests."""

    condition_template = """
      (flow_requests.client_id = %s AND
       flow_requests.flow_id = %s AND
       flow_requests.request_id = %s AND
       responses_expected = %s)"""

    args = []
    conditions = []
    completed_requests = {}

    for request_key in request_keys:
      client_id, flow_id, request_id = request_key
      if request_key in response_counts:
        conditions.append(condition_template)
        args.append(db_utils.ClientIDToInt(client_id))
        args.append(db_utils.FlowIDToInt(flow_id))
        args.append(request_id)
        args.append(response_counts[request_key])

    if not args:
      return completed_requests

    query = """
      SELECT client_id, flow_id, request_id, request
      FROM flow_requests
      WHERE ({conditions}) AND NOT needs_processing
      FOR UPDATE
    """

    query = query.format(conditions=" OR ".join(conditions))
    cursor.execute(query, args)
    for client_id_int, flow_id_int, request_id, request in cursor.fetchall():
      request_key = (db_utils.IntToClientID(client_id_int),
                     db_utils.IntToFlowID(flow_id_int), request_id)
      r = rdf_flow_objects.FlowRequest.FromSerializedString(request)
      completed_requests[request_key] = r

    query = """
    UPDATE flow_requests
    SET needs_processing = TRUE
    WHERE ({conditions}) AND NOT needs_processing
    """
    query = query.format(conditions=" OR ".join(conditions))
    cursor.execute(query, args)

    return completed_requests