code: def _LeaseFlowProcessingReqests(self, cursor=None):
      
      now = rdfvalue.RDFDatetime.Now()
      expiry = now + rdfvalue.Duration("10m")
  
      query = """
        UPDATE flow_processing_requests
        SET leased_until=FROM_UNIXTIME(%(expiry)s), leased_by=%(id)s
        WHERE
         (delivery_time IS NULL OR
          delivery_time <= NOW(6)) AND
         (leased_until IS NULL OR
          leased_until < NOW(6))
        LIMIT %(limit)s
      """
  
      id_str = utils.ProcessIdString()
      args = {
          "expiry": mysql_utils.RDFDatetimeToTimestamp(expiry),
          "id": id_str,
          "limit": 50,
      }
  
      updated = cursor.execute(query, args)
  
      if updated == 0:
        return []
  
      query = """
        SELECT UNIX_TIMESTAMP(timestamp), request
        FROM flow_processing_requests
        FORCE INDEX (flow_processing_requests_by_lease)
        WHERE leased_by=%(id)s AND leased_until=FROM_UNIXTIME(%(expiry)s)
        LIMIT %(updated)s
      """
  
      args = {
          "expiry": mysql_utils.RDFDatetimeToTimestamp(expiry),
          "id": id_str,
          "updated": updated,
      }
  
      cursor.execute(query, args)
  
      res = []
      for timestamp, request in cursor.fetchall():
        req = rdf_flows.FlowProcessingRequest.FromSerializedString(request)
        req.timestamp = mysql_utils.TimestampToRDFDatetime(timestamp)
        req.leased_until = expiry
        req.leased_by = id_str
        res.append(req)
  
      return res
docs: Leases a number of flow processing requests.