'add', 'aggregate', 'all', 'allow', 'alter', 'and', 'apply', 'as', 'asc', 'ascii', 'authorize', 'batch', 'begin',

'bigint', 'blob', 'boolean', 'by', 'called', 'clustering', 'columnfamily', 'compact', 'contains', 'count',

'counter', 'create', 'custom', 'date', 'decimal', 'delete', 'desc', 'describe', 'distinct', 'double', 'drop',

'entries', 'execute', 'exists', 'filtering', 'finalfunc', 'float', 'from', 'frozen', 'full', 'function',

'functions', 'grant', 'if', 'in', 'index', 'inet', 'infinity', 'initcond', 'input', 'insert', 'int', 'into', 'is', 'json',

'key', 'keys', 'keyspace', 'keyspaces', 'language', 'limit', 'list', 'login', 'map', 'materialized', 'modify', 'nan', 'nologin',

'norecursive', 'nosuperuser', 'not', 'null', 'of', 'on', 'options', 'or', 'order', 'password', 'permission',

'permissions', 'primary', 'rename', 'replace', 'returns', 'revoke', 'role', 'roles', 'schema', 'select', 'set',

'sfunc', 'smallint', 'static', 'storage', 'stype', 'superuser', 'table', 'text', 'time', 'timestamp', 'timeuuid',

'tinyint', 'to', 'token', 'trigger', 'truncate', 'ttl', 'tuple', 'type', 'unlogged', 'update', 'use', 'user',

'users', 'using', 'uuid', 'values', 'varchar', 'varint', 'view', 'where', 'with', 'writetime'

'aggregate', 'all', 'as', 'ascii', 'bigint', 'blob', 'boolean', 'called', 'clustering', 'compact', 'contains',

'count', 'counter', 'custom', 'date', 'decimal', 'distinct', 'double', 'exists', 'filtering', 'finalfunc', 'float',

'frozen', 'function', 'functions', 'inet', 'initcond', 'input', 'int', 'json', 'key', 'keys', 'keyspaces',

'language', 'list', 'login', 'map', 'nologin', 'nosuperuser', 'options', 'password', 'permission', 'permissions',

'returns', 'role', 'roles', 'sfunc', 'smallint', 'static', 'storage', 'stype', 'superuser', 'text', 'time',

'timestamp', 'timeuuid', 'tinyint', 'trigger', 'ttl', 'tuple', 'type', 'user', 'users', 'uuid', 'values', 'varchar',

'varint', 'writetime'

"""

cluster_name = None

""" The string name of the cluster. """

keyspaces = None

"""

partitioner = None

"""

token_map = None

""" A :class:`~.TokenMap` instance describing the ring topology. """

def __init__(self):

self.keyspaces = {}

self._hosts = {}

self._hosts_lock = RLock()

def export_schema_as_string(self):

"""

return "\n\n".join(ks.export_as_string() for ks in self.keyspaces.values())

def refresh(self, connection, timeout, target_type=None, change_type=None, **kwargs):

server_version = self.get_host(connection.host).release_version

parser = get_schema_parser(connection, server_version, timeout)

if not target_type:

self._rebuild_all(parser)

return

tt_lower = target_type.lower()

try:

parse_method = getattr(parser, 'get_' + tt_lower)

meta = parse_method(self.keyspaces, **kwargs)

if meta:

update_method = getattr(self, '_update_' + tt_lower)

update_method(meta)

else:

drop_method = getattr(self, '_drop_' + tt_lower)

drop_method(**kwargs)

except AttributeError:

raise ValueError("Unknown schema target_type: '%s'" % target_type)

def _rebuild_all(self, parser):

current_keyspaces = set()

for keyspace_meta in parser.get_all_keyspaces():

current_keyspaces.add(keyspace_meta.name)

old_keyspace_meta = self.keyspaces.get(keyspace_meta.name, None)

self.keyspaces[keyspace_meta.name] = keyspace_meta

if old_keyspace_meta:

self._keyspace_updated(keyspace_meta.name)

else:

self._keyspace_added(keyspace_meta.name)

# remove not-just-added keyspaces

removed_keyspaces = [name for name in self.keyspaces.keys()

if name not in current_keyspaces]

self.keyspaces = dict((name, meta) for name, meta in self.keyspaces.items()

if name in current_keyspaces)

for ksname in removed_keyspaces:

self._keyspace_removed(ksname)

def _update_keyspace(self, keyspace_meta, new_user_types=None):

ks_name = keyspace_meta.name

old_keyspace_meta = self.keyspaces.get(ks_name, None)

self.keyspaces[ks_name] = keyspace_meta

if old_keyspace_meta:

keyspace_meta.tables = old_keyspace_meta.tables

keyspace_meta.user_types = new_user_types if new_user_types is not None else old_keyspace_meta.user_types

keyspace_meta.indexes = old_keyspace_meta.indexes

keyspace_meta.functions = old_keyspace_meta.functions

keyspace_meta.aggregates = old_keyspace_meta.aggregates

keyspace_meta.views = old_keyspace_meta.views

if (keyspace_meta.replication_strategy != old_keyspace_meta.replication_strategy):

self._keyspace_updated(ks_name)

else:

self._keyspace_added(ks_name)

def _drop_keyspace(self, keyspace):

if self.keyspaces.pop(keyspace, None):

self._keyspace_removed(keyspace)

def _update_table(self, meta):

try:

keyspace_meta = self.keyspaces[meta.keyspace_name]

# this is unfortunate, but protocol v4 does not differentiate

# between events for tables and views. <parser>.get_table will

# return one or the other based on the query results.

# Here we deal with that.

if isinstance(meta, TableMetadata):

keyspace_meta._add_table_metadata(meta)

else:

keyspace_meta._add_view_metadata(meta)

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _drop_table(self, keyspace, table):

try:

keyspace_meta = self.keyspaces[keyspace]

keyspace_meta._drop_table_metadata(table) # handles either table or view

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _update_type(self, type_meta):

try:

self.keyspaces[type_meta.keyspace].user_types[type_meta.name] = type_meta

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _drop_type(self, keyspace, type):

try:

self.keyspaces[keyspace].user_types.pop(type, None)

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _update_function(self, function_meta):

try:

self.keyspaces[function_meta.keyspace].functions[function_meta.signature] = function_meta

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _drop_function(self, keyspace, function):

try:

self.keyspaces[keyspace].functions.pop(function.signature, None)

except KeyError:

pass

def _update_aggregate(self, aggregate_meta):

try:

self.keyspaces[aggregate_meta.keyspace].aggregates[aggregate_meta.signature] = aggregate_meta

except KeyError:

pass

def _drop_aggregate(self, keyspace, aggregate):

try:

self.keyspaces[keyspace].aggregates.pop(aggregate.signature, None)

except KeyError:

pass

def _keyspace_added(self, ksname):

if self.token_map:

self.token_map.rebuild_keyspace(ksname, build_if_absent=False)

def _keyspace_updated(self, ksname):

if self.token_map:

self.token_map.rebuild_keyspace(ksname, build_if_absent=False)

def _keyspace_removed(self, ksname):

if self.token_map:

self.token_map.remove_keyspace(ksname)

def rebuild_token_map(self, partitioner, token_map):

"""

self.partitioner = partitioner

if partitioner.endswith('RandomPartitioner'):

token_class = MD5Token

elif partitioner.endswith('Murmur3Partitioner'):

token_class = Murmur3Token

elif partitioner.endswith('ByteOrderedPartitioner'):

token_class = BytesToken

else:

self.token_map = None

return

token_to_host_owner = {}

ring = []

for host, token_strings in six.iteritems(token_map):

for token_string in token_strings:

token = token_class.from_string(token_string)

ring.append(token)

token_to_host_owner[token] = host

all_tokens = sorted(ring)

self.token_map = TokenMap(

token_class, token_to_host_owner, all_tokens, self)

def get_replicas(self, keyspace, key):

"""

t = self.token_map

if not t:

return []

try:

return t.get_replicas(keyspace, t.token_class.from_key(key))

except NoMurmur3:

return []

def can_support_partitioner(self):

if self.partitioner.endswith('Murmur3Partitioner') and murmur3 is None:

return False

else:

return True

def add_or_return_host(self, host):

"""

with self._hosts_lock:

try:

return self._hosts[host.address], False

except KeyError:

self._hosts[host.address] = host

return host, True

def remove_host(self, host):

with self._hosts_lock:

return bool(self._hosts.pop(host.address, False))

def get_host(self, address):

return self._hosts.get(address)

def all_hosts(self):

"""

with self._hosts_lock:

def __new__(metacls, name, bases, dct):

dct.setdefault('name', name)

cls = type.__new__(metacls, name, bases, dct)

if not name.startswith('_'):

_replication_strategies[name] = cls

options_map = None

@classmethod

def create(cls, strategy_class, options_map):

if not strategy_class:

return None

strategy_name = trim_if_startswith(strategy_class, REPLICATION_STRATEGY_CLASS_PREFIX)

rs_class = _replication_strategies.get(strategy_name, None)

if rs_class is None:

rs_class = _UnknownStrategyBuilder(strategy_name)

_replication_strategies[strategy_name] = rs_class

try:

rs_instance = rs_class(options_map)

except Exception as exc:

log.warning("Failed creating %s with options %s: %s", strategy_name, options_map, exc)

return None

return rs_instance

def make_token_replica_map(self, token_to_host_owner, ring):

raise NotImplementedError()

def export_for_schema(self):

def __init__(self, name):

self.name = name

def __call__(self, options_map):

strategy_instance = _UnknownStrategy(self.name, options_map)

def __init__(self, name, options_map):

self.name = name

self.options_map = options_map.copy() if options_map is not None else dict()

self.options_map['class'] = self.name

def __eq__(self, other):

return (isinstance(other, _UnknownStrategy)

and self.name == other.name

and self.options_map == other.options_map)

def export_for_schema(self):

"""

if self.options_map:

return dict((str(key), str(value)) for key, value in self.options_map.items())

return "{'class': '%s'}" % (self.name, )

def make_token_replica_map(self, token_to_host_owner, ring):

replication_factor = None

"""

def __init__(self, options_map):

try:

self.replication_factor = int(options_map['replication_factor'])

except Exception:

raise ValueError("SimpleStrategy requires an integer 'replication_factor' option")

def make_token_replica_map(self, token_to_host_owner, ring):

replica_map = {}

for i in range(len(ring)):

j, hosts = 0, list()

while len(hosts) < self.replication_factor and j < len(ring):

token = ring[(i + j) % len(ring)]

host = token_to_host_owner[token]

if host not in hosts:

hosts.append(host)

j += 1

replica_map[ring[i]] = hosts

return replica_map

def export_for_schema(self):

"""

return "{'class': 'SimpleStrategy', 'replication_factor': '%d'}" \

% (self.replication_factor,)

def __eq__(self, other):

if not isinstance(other, SimpleStrategy):

return False

dc_replication_factors = None

"""

def __init__(self, dc_replication_factors):

self.dc_replication_factors = dict(

(str(k), int(v)) for k, v in dc_replication_factors.items())

def make_token_replica_map(self, token_to_host_owner, ring):

dc_rf_map = dict((dc, int(rf))

for dc, rf in self.dc_replication_factors.items() if rf > 0)

# build a map of DCs to lists of indexes into `ring` for tokens that

# belong to that DC

dc_to_token_offset = defaultdict(list)

dc_racks = defaultdict(set)

hosts_per_dc = defaultdict(set)

for i, token in enumerate(ring):

host = token_to_host_owner[token]

dc_to_token_offset[host.datacenter].append(i)

if host.datacenter and host.rack:

dc_racks[host.datacenter].add(host.rack)

hosts_per_dc[host.datacenter].add(host)

# A map of DCs to an index into the dc_to_token_offset value for that dc.

# This is how we keep track of advancing around the ring for each DC.

dc_to_current_index = defaultdict(int)

replica_map = defaultdict(list)

for i in range(len(ring)):

replicas = replica_map[ring[i]]

# go through each DC and find the replicas in that DC

for dc in dc_to_token_offset.keys():

if dc not in dc_rf_map:

continue

# advance our per-DC index until we're up to at least the

# current token in the ring

token_offsets = dc_to_token_offset[dc]

index = dc_to_current_index[dc]

num_tokens = len(token_offsets)

while index < num_tokens and token_offsets[index] < i:

index += 1

dc_to_current_index[dc] = index

replicas_remaining = dc_rf_map[dc]

replicas_this_dc = 0

skipped_hosts = []

racks_placed = set()

racks_this_dc = dc_racks[dc]

hosts_this_dc = len(hosts_per_dc[dc])

for token_offset in islice(cycle(token_offsets), index, index + num_tokens):

host = token_to_host_owner[ring[token_offset]]

if replicas_remaining == 0 or replicas_this_dc == hosts_this_dc:

break

if host in replicas:

continue

if host.rack in racks_placed and len(racks_placed) < len(racks_this_dc):

skipped_hosts.append(host)

continue

replicas.append(host)

replicas_this_dc += 1

replicas_remaining -= 1

racks_placed.add(host.rack)

if len(racks_placed) == len(racks_this_dc):

for host in skipped_hosts:

if replicas_remaining == 0:

break

replicas.append(host)

replicas_remaining -= 1

del skipped_hosts[:]

return replica_map

def export_for_schema(self):

"""

ret = "{'class': 'NetworkTopologyStrategy'"

for dc, repl_factor in sorted(self.dc_replication_factors.items()):

ret += ", '%s': '%d'" % (dc, repl_factor)

return ret + "}"

def __eq__(self, other):

if not isinstance(other, NetworkTopologyStrategy):

return False

def __init__(self, options_map):

pass

def make_token_replica_map(self, token_to_host_owner, ring):

return {}

def export_for_schema(self):

"""

return "{'class': 'LocalStrategy'}"

def __eq__(self, other):

"""

name = None

""" The string name of the keyspace. """

durable_writes = True

"""

replication_strategy = None

"""

tables = None

"""

indexes = None

"""

user_types = None

"""

functions = None

"""

aggregates = None

"""

views = None

"""

_exc_info = None

""" set if metadata parsing failed """

def __init__(self, name, durable_writes, strategy_class, strategy_options):

self.name = name

self.durable_writes = durable_writes

self.replication_strategy = ReplicationStrategy.create(strategy_class, strategy_options)

self.tables = {}

self.indexes = {}

self.user_types = {}

self.functions = {}

self.aggregates = {}

self.views = {}

def export_as_string(self):

"""

cql = "\n\n".join([self.as_cql_query() + ';']

+ self.user_type_strings()

+ [f.export_as_string() for f in self.functions.values()]

+ [a.export_as_string() for a in self.aggregates.values()]

+ [t.export_as_string() for t in self.tables.values()])

if self._exc_info:

import traceback

ret = "/*\nWarning: Keyspace %s is incomplete because of an error processing metadata.\n" % \

(self.name)

for line in traceback.format_exception(*self._exc_info):

ret += line

ret += "\nApproximate structure, for reference:\n(this should not be used to reproduce this schema)\n\n%s\n*/" % cql

return ret

return cql

def as_cql_query(self):

"""

ret = "CREATE KEYSPACE %s WITH replication = %s " % (

protect_name(self.name),

self.replication_strategy.export_for_schema())

return ret + (' AND durable_writes = %s' % ("true" if self.durable_writes else "false"))

def user_type_strings(self):

user_type_strings = []

user_types = self.user_types.copy()

keys = sorted(user_types.keys())

for k in keys:

if k in user_types:

self.resolve_user_types(k, user_types, user_type_strings)

return user_type_strings

def resolve_user_types(self, key, user_types, user_type_strings):

user_type = user_types.pop(key)

for type_name in user_type.field_types:

for sub_type in types.cql_types_from_string(type_name):

if sub_type in user_types:

self.resolve_user_types(sub_type, user_types, user_type_strings)

user_type_strings.append(user_type.export_as_string())

def _add_table_metadata(self, table_metadata):

old_indexes = {}

old_meta = self.tables.get(table_metadata.name, None)

if old_meta:

# views are not queried with table, so they must be transferred to new

table_metadata.views = old_meta.views

# indexes will be updated with what is on the new metadata

old_indexes = old_meta.indexes

# note the intentional order of add before remove

# this makes sure the maps are never absent something that existed before this update

for index_name, index_metadata in six.iteritems(table_metadata.indexes):

self.indexes[index_name] = index_metadata

for index_name in (n for n in old_indexes if n not in table_metadata.indexes):

self.indexes.pop(index_name, None)

self.tables[table_metadata.name] = table_metadata

def _drop_table_metadata(self, table_name):

table_meta = self.tables.pop(table_name, None)

if table_meta:

for index_name in table_meta.indexes:

self.indexes.pop(index_name, None)

for view_name in table_meta.views:

self.views.pop(view_name, None)

return

# we can't tell table drops from views, so drop both

# (name is unique among them, within a keyspace)

view_meta = self.views.pop(table_name, None)

if view_meta:

try:

self.tables[view_meta.base_table_name].views.pop(table_name, None)

except KeyError:

pass

def _add_view_metadata(self, view_metadata):

try:

self.tables[view_metadata.base_table_name].views[view_metadata.name] = view_metadata

self.views[view_metadata.name] = view_metadata

except KeyError:

"""

keyspace = None

"""

name = None

"""

field_names = None

"""

field_types = None

"""

def __init__(self, keyspace, name, field_names, field_types):

self.keyspace = keyspace

self.name = name

# non-frozen collections can return None

self.field_names = field_names or []

self.field_types = field_types or []

def as_cql_query(self, formatted=False):

"""

ret = "CREATE TYPE %s.%s (%s" % (

protect_name(self.keyspace),

protect_name(self.name),

"\n" if formatted else "")

if formatted:

field_join = ",\n"

padding = " "

else:

field_join = ", "

padding = ""

fields = []

for field_name, field_type in zip(self.field_names, self.field_types):

fields.append("%s %s" % (protect_name(field_name), field_type))

ret += field_join.join("%s%s" % (padding, field) for field in fields)

ret += "\n)" if formatted else ")"

return ret

def export_as_string(self):

"""

keyspace = None

"""

name = None

"""

argument_types = None

"""

final_func = None

"""

initial_condition = None

"""

return_type = None

"""

state_func = None

"""

state_type = None

"""

def __init__(self, keyspace, name, argument_types, state_func,

state_type, final_func, initial_condition, return_type):

self.keyspace = keyspace

self.name = name

self.argument_types = argument_types

self.state_func = state_func

self.state_type = state_type

self.final_func = final_func

self.initial_condition = initial_condition

self.return_type = return_type

def as_cql_query(self, formatted=False):

"""

sep = '\n ' if formatted else ' '

keyspace = protect_name(self.keyspace)

name = protect_name(self.name)

type_list = ', '.join(self.argument_types)

state_func = protect_name(self.state_func)

state_type = self.state_type

ret = "CREATE AGGREGATE %(keyspace)s.%(name)s(%(type_list)s)%(sep)s" \

"SFUNC %(state_func)s%(sep)s" \

"STYPE %(state_type)s" % locals()

ret += ''.join((sep, 'FINALFUNC ', protect_name(self.final_func))) if self.final_func else ''

ret += ''.join((sep, 'INITCOND ', self.initial_condition)) if self.initial_condition is not None else ''

return ret

def export_as_string(self):

return self.as_cql_query(formatted=True) + ';'

@property

def signature(self):

"""

keyspace = None

"""

name = None

"""

argument_types = None

"""

argument_names = None

"""

return_type = None

"""

language = None

"""

body = None

"""

called_on_null_input = None

"""

def __init__(self, keyspace, name, argument_types, argument_names,

return_type, language, body, called_on_null_input):

self.keyspace = keyspace

self.name = name

self.argument_types = argument_types

# argument_types (frozen<list<>>) will always be a list

# argument_name is not frozen in C* < 3.0 and may return None

self.argument_names = argument_names or []

self.return_type = return_type

self.language = language

self.body = body

self.called_on_null_input = called_on_null_input

def as_cql_query(self, formatted=False):

"""

sep = '\n ' if formatted else ' '

keyspace = protect_name(self.keyspace)

name = protect_name(self.name)

arg_list = ', '.join(["%s %s" % (protect_name(n), t)

for n, t in zip(self.argument_names, self.argument_types)])

typ = self.return_type

lang = self.language

body = self.body

on_null = "CALLED" if self.called_on_null_input else "RETURNS NULL"

return "CREATE FUNCTION %(keyspace)s.%(name)s(%(arg_list)s)%(sep)s" \

"%(on_null)s ON NULL INPUT%(sep)s" \

"RETURNS %(typ)s%(sep)s" \

"LANGUAGE %(lang)s%(sep)s" \

"AS $$%(body)s$$" % locals()

def export_as_string(self):

return self.as_cql_query(formatted=True) + ';'

@property

def signature(self):

"""

keyspace_name = None

""" String name of this Table's keyspace """

name = None

""" The string name of the table. """