def UpdateFlow(self,
                 client_id,
                 flow_id,
                 flow_obj=db.Database.unchanged,
                 flow_state=db.Database.unchanged,
                 client_crash_info=db.Database.unchanged,
                 pending_termination=db.Database.unchanged,
                 processing_on=db.Database.unchanged,
                 processing_since=db.Database.unchanged,
                 processing_deadline=db.Database.unchanged,
                 cursor=None):
    """Updates flow objects in the database."""
    updates = []
    args = []
    if flow_obj != db.Database.unchanged:
      updates.append("flow=%s")
      args.append(flow_obj.SerializeToString())
      updates.append("flow_state=%s")
      args.append(int(flow_obj.flow_state))
      updates.append("user_cpu_time_used_micros=%s")
      args.append(
          db_utils.SecondsToMicros(flow_obj.cpu_time_used.user_cpu_time))
      updates.append("system_cpu_time_used_micros=%s")
      args.append(
          db_utils.SecondsToMicros(flow_obj.cpu_time_used.system_cpu_time))
      updates.append("network_bytes_sent=%s")
      args.append(flow_obj.network_bytes_sent)
      updates.append("num_replies_sent=%s")
      args.append(flow_obj.num_replies_sent)

    if flow_state != db.Database.unchanged:
      updates.append("flow_state=%s")
      args.append(int(flow_state))
    if client_crash_info != db.Database.unchanged:
      updates.append("client_crash_info=%s")
      args.append(client_crash_info.SerializeToString())
    if pending_termination != db.Database.unchanged:
      updates.append("pending_termination=%s")
      args.append(pending_termination.SerializeToString())
    if processing_on != db.Database.unchanged:
      updates.append("processing_on=%s")
      args.append(processing_on)
    if processing_since != db.Database.unchanged:
      updates.append("processing_since=FROM_UNIXTIME(%s)")
      args.append(mysql_utils.RDFDatetimeToTimestamp(processing_since))
    if processing_deadline != db.Database.unchanged:
      updates.append("processing_deadline=FROM_UNIXTIME(%s)")
      args.append(mysql_utils.RDFDatetimeToTimestamp(processing_deadline))

    if not updates:
      return

    query = "UPDATE flows SET last_update=NOW(6), "
    query += ", ".join(updates)
    query += " WHERE client_id=%s AND flow_id=%s"

    args.append(db_utils.ClientIDToInt(client_id))
    args.append(db_utils.FlowIDToInt(flow_id))
    updated = cursor.execute(query, args)
    if updated == 0:
      raise db.UnknownFlowError(client_id, flow_id)
Updates flow objects in the database.