Returns a :class:`dict` with the keys grouped per host. This can be used to more accurately group by IN clause or to batch the keys per host. If a valid replica is not found for a particular key it will be grouped under :class:`~.NO_VALID_REPLICA` Example usage:: resul
(session, keyspace, table, keys)
| 3323 | |
| 3324 | |
| 3325 | def group_keys_by_replica(session, keyspace, table, keys): |
| 3326 | """ |
| 3327 | Returns a :class:`dict` with the keys grouped per host. This can be |
| 3328 | used to more accurately group by IN clause or to batch the keys per host. |
| 3329 | |
| 3330 | If a valid replica is not found for a particular key it will be grouped under |
| 3331 | :class:`~.NO_VALID_REPLICA` |
| 3332 | |
| 3333 | Example usage:: |
| 3334 | result = group_keys_by_replica( |
| 3335 | session, "system", "peers", |
| 3336 | (("127.0.0.1", ), ("127.0.0.2", )) |
| 3337 | ) |
| 3338 | """ |
| 3339 | cluster = session.cluster |
| 3340 | |
| 3341 | partition_keys = cluster.metadata.keyspaces[keyspace].tables[table].partition_key |
| 3342 | |
| 3343 | serializers = list(types._cqltypes[partition_key.cql_type] for partition_key in partition_keys) |
| 3344 | keys_per_host = defaultdict(list) |
| 3345 | distance = cluster._default_load_balancing_policy.distance |
| 3346 | |
| 3347 | for key in keys: |
| 3348 | serialized_key = [serializer.serialize(pk, cluster.protocol_version) |
| 3349 | for serializer, pk in zip(serializers, key)] |
| 3350 | if len(serialized_key) == 1: |
| 3351 | routing_key = serialized_key[0] |
| 3352 | else: |
| 3353 | routing_key = b"".join(struct.pack(">H%dsB" % len(p), len(p), p, 0) for p in serialized_key) |
| 3354 | all_replicas = cluster.metadata.get_replicas(keyspace, routing_key) |
| 3355 | # First check if there are local replicas |
| 3356 | valid_replicas = [host for host in all_replicas if |
| 3357 | host.is_up and distance(host) == HostDistance.LOCAL] |
| 3358 | if not valid_replicas: |
| 3359 | valid_replicas = [host for host in all_replicas if host.is_up] |
| 3360 | |
| 3361 | if valid_replicas: |
| 3362 | keys_per_host[random.choice(valid_replicas)].append(key) |
| 3363 | else: |
| 3364 | # We will group under this statement all the keys for which |
| 3365 | # we haven't found a valid replica |
| 3366 | keys_per_host[NO_VALID_REPLICA].append(key) |
| 3367 | |
| 3368 | return dict(keys_per_host) |
| 3369 | |
| 3370 | |
| 3371 | # TODO next major reorg |