class_docstr: Bounds the total contributions of a privacy_id.
  
  If a privacy_id contributes more than max_contributions, then
  max_contributions contributions are uniformly sampled, otherwise all
  contributions are kept.
class_docstr_tok: ["Bounds","the","total","contributions","of","a","privacy_id",".","If","a","privacy_id","contributes","more","than","max_contributions",",","then","max_contributions","contributions","are","uniformly","sampled",",","otherwise","all","contributions","are","kept","."]
docstr: See docstrings for this class and the base class.
docstr_tok: ["See","docstrings","for","this","class","and","the","base","class","."]
code: def bound_contributions(
      self, col, params, backend, report_generator, aggregate_fn
  ):
      """See docstrings for this class and the base class."""
      max_contributions = params.max_contributions
      col = backend.map_tuple(
          col,
          lambda pid, pk, v: (pid, (pk, v)),
          "Rekey to ((privacy_id), (partition_key, value))",
      )
      col = backend.sample_fixed_per_key(
          col, max_contributions, "Sample per privacy_id"
      )
      report_generator.add_stage(
          f"User contribution bounding: randomly selected not "
          f"more than {max_contributions} contributions"
      )
  
      # (privacy_id, [(partition_key, value)])
  
      col = collect_values_per_partition_key_per_privacy_id(
          col, backend
      )
  
      # (privacy_id, [(partition_key, [value])])
  
      # Rekey it into values per privacy id and partition key.
      # Unnest the list per privacy id.
      def rekey_per_privacy_id_per_partition_key(pid_pk_v_values):
          privacy_id, partition_values = pid_pk_v_values
  
          for partition_key, values in partition_values:
              yield (privacy_id, partition_key), values
  
      # Unnest the list per privacy id.
      col = backend.flat_map(
          col, rekey_per_privacy_id_per_partition_key, "Unnest"
      )
      # ((privacy_id, partition_key), [value])
  
      return backend.map_values(
          col,
          aggregate_fn,
          "Apply aggregate_fn after per privacy_id contribution bounding",
      )
  
file_path: pipelinedp/pipeline_dp/contribution_bounders.py
imports: import abc
  import collections
  from typing import Callable, Iterable
  import pipeline_dp
  from pipeline_dp import pipeline_backend
  from pipeline_dp import sampling_utils
label_desc: Use image node_id 1 for calling the SamplingPerPrivacyIdContributionBounder obj's underlying member method code with example usage: obj.bound_contributions(col, params, backend, report_generator, aggregate_fn) and returns: backend