docs: def LeaseFlowForProcessing(self, client_id, flow_id, processing_time):
      """Marks a flow as being processed on this worker and returns it."""
      rdf_flow = self.ReadFlowObject(client_id, flow_id)
      # TODO(user): remove the check for a legacy hunt prefix as soon as
      # AFF4 is gone.
      if rdf_flow.parent_hunt_id and not rdf_flow.parent_hunt_id.startswith("H:"):
        rdf_hunt = self.ReadHuntObject(rdf_flow.parent_hunt_id)
        if not rdf_hunt_objects.IsHuntSuitableForFlowProcessing(
            rdf_hunt.hunt_state):
          raise db.ParentHuntIsNotRunningError(client_id, flow_id,
                                               rdf_hunt.hunt_id,
                                               rdf_hunt.hunt_state)
  
      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))
      processing_deadline = now + processing_time
      process_id_string = utils.ProcessIdString()
      self.UpdateFlow(
          client_id,
          flow_id,
          processing_on=process_id_string,
          processing_since=now,
          processing_deadline=processing_deadline)
      rdf_flow.processing_on = process_id_string
      rdf_flow.processing_since = now
      rdf_flow.processing_deadline = processing_deadline
      return rdf_flow