diff --git a/bin/cqlsh.py b/bin/cqlsh.py index 9a561eb200..7e6db9bdb8 100755 --- a/bin/cqlsh.py +++ b/bin/cqlsh.py @@ -16,26 +16,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -import cmd -import codecs -import configparser -import csv -import errno -import getpass -import optparse import os import platform -import re -import stat -import subprocess import sys -import traceback -import warnings -import webbrowser -from contextlib import contextmanager from glob import glob -from io import StringIO -from uuid import UUID if sys.version_info < (3, 6): sys.exit("\ncqlsh requires Python 3.6+\n") @@ -44,50 +28,9 @@ if sys.version_info < (3, 6): if platform.python_implementation().startswith('Jython'): sys.exit("\nCQL Shell does not run on Jython\n") -UTF8 = 'utf-8' - -description = "CQL Shell for Apache Cassandra" -version = "6.2.0" - -readline = None -try: - # check if tty first, cause readline doesn't check, and only cares - # about $TERM. we don't want the funky escape code stuff to be - # output if not a tty. - if sys.stdin.isatty(): - import readline -except ImportError: - pass - CQL_LIB_PREFIX = 'cassandra-driver-internal-only-' CASSANDRA_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..') -CASSANDRA_CQL_HTML_FALLBACK = 'https://cassandra.apache.org/doc/latest/cql/index.html' - -# default location of local CQL.html -if os.path.exists(CASSANDRA_PATH + '/doc/cql3/CQL.html'): - # default location of local CQL.html - CASSANDRA_CQL_HTML = 'file://' + CASSANDRA_PATH + '/doc/cql3/CQL.html' -elif os.path.exists('/usr/share/doc/cassandra/CQL.html'): - # fallback to package file - CASSANDRA_CQL_HTML = 'file:///usr/share/doc/cassandra/CQL.html' -else: - # fallback to online version - CASSANDRA_CQL_HTML = CASSANDRA_CQL_HTML_FALLBACK - -# On Linux, the Python webbrowser module uses the 'xdg-open' executable -# to open a file/URL. But that only works, if the current session has been -# opened from _within_ a desktop environment. I.e. 'xdg-open' will fail, -# if the session's been opened via ssh to a remote box. -# -try: - webbrowser.register_standard_browsers() # registration is otherwise lazy in Python3 -except AttributeError: - pass -if webbrowser._tryorder and webbrowser._tryorder[0] == 'xdg-open' and os.environ.get('XDG_DATA_DIRS', '') == '': - # only on Linux (some OS with xdg-open) - webbrowser._tryorder.remove('xdg-open') - webbrowser._tryorder.append('xdg-open') # use bundled lib for python-cql if available. if there # is a ../lib dir, use bundled libs there preferentially. @@ -120,7 +63,6 @@ for lib in third_parties: if lib_zip: sys.path.insert(0, lib_zip) -warnings.filterwarnings("ignore", r".*blist.*") try: import cassandra except ImportError as e: @@ -130,14 +72,6 @@ except ImportError as e: 'Module load path: %r\n\n' 'Error: %s\n' % (sys.executable, sys.path, e)) -from cassandra.auth import PlainTextAuthProvider -from cassandra.cluster import Cluster -from cassandra.cqltypes import cql_typename -from cassandra.marshal import int64_unpack -from cassandra.metadata import (ColumnMetadata, KeyspaceMetadata, TableMetadata) -from cassandra.policies import WhiteListRoundRobinPolicy -from cassandra.query import SimpleStatement, ordered_dict_factory, TraceUnavailable -from cassandra.util import datetime_from_timestamp # cqlsh should run correctly when run out of a Cassandra source tree, # out of an unpacked Cassandra tarball, and after a proper package install. @@ -145,2261 +79,10 @@ cqlshlibdir = os.path.join(CASSANDRA_PATH, 'pylib') if os.path.isdir(cqlshlibdir): sys.path.insert(0, cqlshlibdir) -from cqlshlib import cql3handling, pylexotron, sslhandling, cqlshhandling, authproviderhandling -from cqlshlib.copyutil import ExportTask, ImportTask -from cqlshlib.displaying import (ANSI_RESET, BLUE, COLUMN_NAME_COLORS, CYAN, - RED, WHITE, FormattedValue, colorme) -from cqlshlib.formatting import (DEFAULT_DATE_FORMAT, DEFAULT_NANOTIME_FORMAT, - DEFAULT_TIMESTAMP_FORMAT, CqlType, DateTimeFormat, - format_by_type) -from cqlshlib.tracing import print_trace, print_trace_session -from cqlshlib.util import get_file_encoding_bomsize -from cqlshlib.util import is_file_secure - - -DEFAULT_HOST = '127.0.0.1' -DEFAULT_PORT = 9042 -DEFAULT_SSL = False -DEFAULT_CONNECT_TIMEOUT_SECONDS = 5 -DEFAULT_REQUEST_TIMEOUT_SECONDS = 10 - -DEFAULT_FLOAT_PRECISION = 5 -DEFAULT_DOUBLE_PRECISION = 5 -DEFAULT_MAX_TRACE_WAIT = 10 - -if readline is not None and readline.__doc__ is not None and 'libedit' in readline.__doc__: - DEFAULT_COMPLETEKEY = '\t' -else: - DEFAULT_COMPLETEKEY = 'tab' - -cqldocs = None -cqlruleset = None - -epilog = """Connects to %(DEFAULT_HOST)s:%(DEFAULT_PORT)d by default. These -defaults can be changed by setting $CQLSH_HOST and/or $CQLSH_PORT. When a -host (and optional port number) are given on the command line, they take -precedence over any defaults.""" % globals() - -parser = optparse.OptionParser(description=description, epilog=epilog, - usage="Usage: %prog [options] [host [port]]", - version='cqlsh ' + version) -parser.add_option("-C", "--color", action='store_true', dest='color', - help='Always use color output') -parser.add_option("--no-color", action='store_false', dest='color', - help='Never use color output') -parser.add_option("--browser", dest='browser', help="""The browser to use to display CQL help, where BROWSER can be: - - one of the supported browsers in https://docs.python.org/3/library/webbrowser.html. - - browser path followed by %s, example: /usr/bin/google-chrome-stable %s""") -parser.add_option('--ssl', action='store_true', help='Use SSL', default=False) -parser.add_option("-u", "--username", help="Authenticate as user.") -parser.add_option("-p", "--password", help="Authenticate using password.") -parser.add_option('-k', '--keyspace', help='Authenticate to the given keyspace.') -parser.add_option("-f", "--file", help="Execute commands from FILE, then exit") -parser.add_option('--debug', action='store_true', - help='Show additional debugging information') -parser.add_option('--coverage', action='store_true', - help='Collect coverage data') -parser.add_option("--encoding", help="Specify a non-default encoding for output." - + " (Default: %s)" % (UTF8,)) -parser.add_option("--cqlshrc", help="Specify an alternative cqlshrc file location.") -parser.add_option("--credentials", help="Specify an alternative credentials file location.") -parser.add_option('--cqlversion', default=None, - help='Specify a particular CQL version, ' - 'by default the highest version supported by the server will be used.' - ' Examples: "3.0.3", "3.1.0"') -parser.add_option("--protocol-version", type="int", default=None, - help='Specify a specific protcol version otherwise the client will default and downgrade as necessary') - -parser.add_option("-e", "--execute", help='Execute the statement and quit.') -parser.add_option("--connect-timeout", default=DEFAULT_CONNECT_TIMEOUT_SECONDS, dest='connect_timeout', - help='Specify the connection timeout in seconds (default: %default seconds).') -parser.add_option("--request-timeout", default=DEFAULT_REQUEST_TIMEOUT_SECONDS, dest='request_timeout', - help='Specify the default request timeout in seconds (default: %default seconds).') -parser.add_option("-t", "--tty", action='store_true', dest='tty', - help='Force tty mode (command prompt).') -parser.add_option('-v', action="version", help='Print the current version of cqlsh.') - -# This is a hidden option to suppress the warning when the -p/--password command line option is used. -# Power users may use this option if they know no other people has access to the system where cqlsh is run or don't care about security. -# Use of this option in scripting is discouraged. Please use a (temporary) credentials file where possible. -# The Cassandra distributed tests (dtests) also use this option in some tests when a well-known password is supplied via the command line. -parser.add_option("--insecure-password-without-warning", action='store_true', dest='insecure_password_without_warning', - help=optparse.SUPPRESS_HELP) - -opt_values = optparse.Values() -(options, arguments) = parser.parse_args(sys.argv[1:], values=opt_values) - -# BEGIN history/config definition - - -def mkdirp(path): - """Creates all parent directories up to path parameter or fails when path exists, but it is not a directory.""" - - try: - os.makedirs(path) - except OSError: - if not os.path.isdir(path): - raise - - -def resolve_cql_history_file(): - default_cql_history = os.path.expanduser(os.path.join('~', '.cassandra', 'cqlsh_history')) - if 'CQL_HISTORY' in os.environ: - return os.environ['CQL_HISTORY'] - else: - return default_cql_history - - -HISTORY = resolve_cql_history_file() -HISTORY_DIR = os.path.dirname(HISTORY) - -try: - mkdirp(HISTORY_DIR) -except OSError: - print('\nWarning: Cannot create directory at `%s`. Command history will not be saved. Please check what was the environment property CQL_HISTORY set to.\n' % HISTORY_DIR) - -DEFAULT_CQLSHRC = os.path.expanduser(os.path.join('~', '.cassandra', 'cqlshrc')) - -if hasattr(options, 'cqlshrc'): - CONFIG_FILE = os.path.expanduser(options.cqlshrc) - if not os.path.exists(CONFIG_FILE): - print('\nWarning: Specified cqlshrc location `%s` does not exist. Using `%s` instead.\n' % (CONFIG_FILE, DEFAULT_CQLSHRC)) - CONFIG_FILE = DEFAULT_CQLSHRC -else: - CONFIG_FILE = DEFAULT_CQLSHRC - -CQL_DIR = os.path.dirname(CONFIG_FILE) - -CQL_ERRORS = ( - cassandra.AlreadyExists, cassandra.AuthenticationFailed, cassandra.CoordinationFailure, - cassandra.InvalidRequest, cassandra.Timeout, cassandra.Unauthorized, cassandra.OperationTimedOut, - cassandra.cluster.NoHostAvailable, - cassandra.connection.ConnectionBusy, cassandra.connection.ProtocolError, cassandra.connection.ConnectionException, - cassandra.protocol.ErrorMessage, cassandra.protocol.InternalError, cassandra.query.TraceUnavailable -) - -debug_completion = bool(os.environ.get('CQLSH_DEBUG_COMPLETION', '') == 'YES') - - -class NoKeyspaceError(Exception): - pass - - -class KeyspaceNotFound(Exception): - pass - - -class ColumnFamilyNotFound(Exception): - pass - - -class IndexNotFound(Exception): - pass - - -class MaterializedViewNotFound(Exception): - pass - - -class ObjectNotFound(Exception): - pass - - -class VersionNotSupported(Exception): - pass - - -class UserTypeNotFound(Exception): - pass - - -class FunctionNotFound(Exception): - pass - - -class AggregateNotFound(Exception): - pass - - -class DecodeError(Exception): - verb = 'decode' - - def __init__(self, thebytes, err, colname=None): - self.thebytes = thebytes - self.err = err - self.colname = colname - - def __str__(self): - return str(self.thebytes) - - def message(self): - what = 'value %r' % (self.thebytes,) - if self.colname is not None: - what = 'value %r (for column %r)' % (self.thebytes, self.colname) - return 'Failed to %s %s : %s' \ - % (self.verb, what, self.err) - - def __repr__(self): - return '<%s %s>' % (self.__class__.__name__, self.message()) - - -def maybe_ensure_text(val): - return str(val) if val else val - - -class FormatError(DecodeError): - verb = 'format' - - -def full_cql_version(ver): - while ver.count('.') < 2: - ver += '.0' - ver_parts = ver.split('-', 1) + [''] - vertuple = tuple(list(map(int, ver_parts[0].split('.'))) + [ver_parts[1]]) - return ver, vertuple - - -def format_value(val, cqltype, encoding, addcolor=False, date_time_format=None, - float_precision=None, colormap=None, nullval=None): - if isinstance(val, DecodeError): - if addcolor: - return colorme(repr(val.thebytes), colormap, 'error') - else: - return FormattedValue(repr(val.thebytes)) - return format_by_type(val, cqltype=cqltype, encoding=encoding, colormap=colormap, - addcolor=addcolor, nullval=nullval, date_time_format=date_time_format, - float_precision=float_precision) - - -def show_warning_without_quoting_line(message, category, filename, lineno, file=None, line=None): - if file is None: - file = sys.stderr - try: - file.write(warnings.formatwarning(message, category, filename, lineno, line='')) - except IOError: - pass - - -warnings.showwarning = show_warning_without_quoting_line -warnings.filterwarnings('always', category=cql3handling.UnexpectedTableStructure) - - -def insert_driver_hooks(): - - class DateOverFlowWarning(RuntimeWarning): - pass - - # Native datetime types blow up outside of datetime.[MIN|MAX]_YEAR. We will fall back to an int timestamp - def deserialize_date_fallback_int(byts, protocol_version): - timestamp_ms = int64_unpack(byts) - try: - return datetime_from_timestamp(timestamp_ms / 1000.0) - except OverflowError: - warnings.warn(DateOverFlowWarning("Some timestamps are larger than Python datetime can represent. " - "Timestamps are displayed in milliseconds from epoch.")) - return timestamp_ms - - cassandra.cqltypes.DateType.deserialize = staticmethod(deserialize_date_fallback_int) - - if hasattr(cassandra, 'deserializers'): - del cassandra.deserializers.DesDateType - - # Return cassandra.cqltypes.EMPTY instead of None for empty values - cassandra.cqltypes.CassandraType.support_empty_values = True - - -class Shell(cmd.Cmd): - custom_prompt = os.getenv('CQLSH_PROMPT', '') - if custom_prompt != '': - custom_prompt += "\n" - default_prompt = custom_prompt + "cqlsh> " - continue_prompt = " ... " - keyspace_prompt = custom_prompt + "cqlsh:{}> " - keyspace_continue_prompt = "{} ... " - show_line_nums = False - debug = False - coverage = False - coveragerc_path = None - stop = False - last_hist = None - shunted_query_out = None - use_paging = True - - default_page_size = 100 - - def __init__(self, hostname, port, color=False, - username=None, encoding=None, stdin=None, tty=True, - completekey=DEFAULT_COMPLETEKEY, browser=None, use_conn=None, - cqlver=None, keyspace=None, - tracing_enabled=False, expand_enabled=False, - display_nanotime_format=DEFAULT_NANOTIME_FORMAT, - display_timestamp_format=DEFAULT_TIMESTAMP_FORMAT, - display_date_format=DEFAULT_DATE_FORMAT, - display_float_precision=DEFAULT_FLOAT_PRECISION, - display_double_precision=DEFAULT_DOUBLE_PRECISION, - display_timezone=None, - max_trace_wait=DEFAULT_MAX_TRACE_WAIT, - ssl=False, - single_statement=None, - request_timeout=DEFAULT_REQUEST_TIMEOUT_SECONDS, - protocol_version=None, - connect_timeout=DEFAULT_CONNECT_TIMEOUT_SECONDS, - is_subshell=False, - auth_provider=None): - cmd.Cmd.__init__(self, completekey=completekey) - self.hostname = hostname - self.port = port - self.auth_provider = auth_provider - self.username = username - - if isinstance(auth_provider, PlainTextAuthProvider): - self.username = auth_provider.username - if not auth_provider.password: - # if no password is provided, we need to query the user to get one. - password = getpass.getpass() - self.auth_provider = PlainTextAuthProvider(username=auth_provider.username, password=password) - - self.keyspace = keyspace - self.ssl = ssl - self.tracing_enabled = tracing_enabled - self.page_size = self.default_page_size - self.expand_enabled = expand_enabled - if use_conn: - self.conn = use_conn - else: - kwargs = {} - if protocol_version is not None: - kwargs['protocol_version'] = protocol_version - self.conn = Cluster(contact_points=(self.hostname,), port=self.port, cql_version=cqlver, - auth_provider=self.auth_provider, - ssl_options=sslhandling.ssl_settings(hostname, CONFIG_FILE) if ssl else None, - load_balancing_policy=WhiteListRoundRobinPolicy([self.hostname]), - control_connection_timeout=connect_timeout, - connect_timeout=connect_timeout, - **kwargs) - self.owns_connection = not use_conn - - if keyspace: - self.session = self.conn.connect(keyspace) - else: - self.session = self.conn.connect() - - if browser == "": - browser = None - self.browser = browser - self.color = color - - self.display_nanotime_format = display_nanotime_format - self.display_timestamp_format = display_timestamp_format - self.display_date_format = display_date_format - - self.display_float_precision = display_float_precision - self.display_double_precision = display_double_precision - - self.display_timezone = display_timezone - - self.session.default_timeout = request_timeout - self.session.row_factory = ordered_dict_factory - self.session.default_consistency_level = cassandra.ConsistencyLevel.ONE - self.get_connection_versions() - self.set_expanded_cql_version(self.connection_versions['cql']) - - self.current_keyspace = keyspace - - self.max_trace_wait = max_trace_wait - self.session.max_trace_wait = max_trace_wait - - self.tty = tty - self.encoding = encoding - - self.output_codec = codecs.lookup(encoding) - - self.statement = StringIO() - self.lineno = 1 - self.in_comment = False - - self.prompt = '' - if stdin is None: - stdin = sys.stdin - - if tty: - self.reset_prompt() - self.report_connection() - print('Use HELP for help.') - else: - self.show_line_nums = True - self.stdin = stdin - self.query_out = sys.stdout - self.consistency_level = cassandra.ConsistencyLevel.ONE - self.serial_consistency_level = cassandra.ConsistencyLevel.SERIAL - - self.empty_lines = 0 - self.statement_error = False - self.single_statement = single_statement - self.is_subshell = is_subshell - - @property - def batch_mode(self): - return not self.tty - - def set_expanded_cql_version(self, ver): - ver, vertuple = full_cql_version(ver) - self.cql_version = ver - self.cql_ver_tuple = vertuple - - def cqlver_atleast(self, major, minor=0, patch=0): - return self.cql_ver_tuple[:3] >= (major, minor, patch) - - def myformat_value(self, val, cqltype=None, **kwargs): - if isinstance(val, DecodeError): - self.decoding_errors.append(val) - try: - dtformats = DateTimeFormat(timestamp_format=self.display_timestamp_format, - date_format=self.display_date_format, nanotime_format=self.display_nanotime_format, - timezone=self.display_timezone) - precision = self.display_double_precision if cqltype is not None and cqltype.type_name == 'double' \ - else self.display_float_precision - return format_value(val, cqltype=cqltype, encoding=self.output_codec.name, - addcolor=self.color, date_time_format=dtformats, - float_precision=precision, **kwargs) - except Exception as e: - err = FormatError(val, e) - self.decoding_errors.append(err) - return format_value(err, cqltype=cqltype, encoding=self.output_codec.name, addcolor=self.color) - - def myformat_colname(self, name, table_meta=None): - column_colors = COLUMN_NAME_COLORS.copy() - # check column role and color appropriately - if table_meta: - if name in [col.name for col in table_meta.partition_key]: - column_colors.default_factory = lambda: RED - elif name in [col.name for col in table_meta.clustering_key]: - column_colors.default_factory = lambda: CYAN - elif name in table_meta.columns and table_meta.columns[name].is_static: - column_colors.default_factory = lambda: WHITE - return self.myformat_value(name, colormap=column_colors) - - def report_connection(self): - self.show_host() - self.show_version() - - def show_host(self): - print("Connected to {0} at {1}:{2}" - .format(self.applycolor(self.get_cluster_name(), BLUE), - self.hostname, - self.port)) - - def show_version(self): - vers = self.connection_versions.copy() - vers['shver'] = version - # system.Versions['cql'] apparently does not reflect changes with - # set_cql_version. - vers['cql'] = self.cql_version - print("[cqlsh %(shver)s | Cassandra %(build)s | CQL spec %(cql)s | Native protocol v%(protocol)s]" % vers) - - def show_session(self, sessionid, partial_session=False): - print_trace_session(self, self.session, sessionid, partial_session) - - def show_replicas(self, token_value, keyspace=None): - ks = self.current_keyspace if keyspace is None else keyspace - token_map = self.conn.metadata.token_map - nodes = token_map.get_replicas(ks, token_map.token_class(token_value)) - addresses = [x.address for x in nodes] - print(f"{addresses}") - - def get_connection_versions(self): - result, = self.session.execute("select * from system.local where key = 'local'") - vers = { - 'build': result['release_version'], - 'protocol': self.conn.protocol_version, - 'cql': result['cql_version'], - } - self.connection_versions = vers - - def get_keyspace_names(self): - return list(self.conn.metadata.keyspaces) - - def get_columnfamily_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return list(self.get_keyspace_meta(ksname).tables) - - def get_materialized_view_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return list(self.get_keyspace_meta(ksname).views) - - def get_index_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return list(self.get_keyspace_meta(ksname).indexes) - - def get_column_names(self, ksname, cfname): - if ksname is None: - ksname = self.current_keyspace - layout = self.get_table_meta(ksname, cfname) - return list(layout.columns) - - def get_usertype_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return list(self.get_keyspace_meta(ksname).user_types) - - def get_usertype_layout(self, ksname, typename): - if ksname is None: - ksname = self.current_keyspace - - ks_meta = self.get_keyspace_meta(ksname) - - try: - user_type = ks_meta.user_types[typename] - except KeyError: - raise UserTypeNotFound("User type {!r} not found".format(typename)) - - return list(zip(user_type.field_names, user_type.field_types)) - - def get_userfunction_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return [f.name for f in list(self.get_keyspace_meta(ksname).functions.values())] - - def get_useraggregate_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return [f.name for f in list(self.get_keyspace_meta(ksname).aggregates.values())] - - def get_cluster_name(self): - return self.conn.metadata.cluster_name - - def get_partitioner(self): - return self.conn.metadata.partitioner - - def get_keyspace_meta(self, ksname): - if ksname in self.conn.metadata.keyspaces: - return self.conn.metadata.keyspaces[ksname] - - raise KeyspaceNotFound('Keyspace %r not found.' % ksname) - - def get_keyspaces(self): - return list(self.conn.metadata.keyspaces.values()) - - def get_ring(self, ks): - self.conn.metadata.token_map.rebuild_keyspace(ks, build_if_absent=True) - return self.conn.metadata.token_map.tokens_to_hosts_by_ks[ks] - - def get_table_meta(self, ksname, tablename): - if ksname is None: - ksname = self.current_keyspace - ksmeta = self.get_keyspace_meta(ksname) - if tablename not in ksmeta.tables: - if ksname == 'system_auth' and tablename in ['roles', 'role_permissions']: - self.get_fake_auth_table_meta(ksname, tablename) - else: - raise ColumnFamilyNotFound("Column family {} not found".format(tablename)) - else: - return ksmeta.tables[tablename] - - def get_fake_auth_table_meta(self, ksname, tablename): - # may be using external auth implementation so internal tables - # aren't actually defined in schema. In this case, we'll fake - # them up - if tablename == 'roles': - ks_meta = KeyspaceMetadata(ksname, True, None, None) - table_meta = TableMetadata(ks_meta, 'roles') - table_meta.columns['role'] = ColumnMetadata(table_meta, 'role', cassandra.cqltypes.UTF8Type) - table_meta.columns['is_superuser'] = ColumnMetadata(table_meta, 'is_superuser', cassandra.cqltypes.BooleanType) - table_meta.columns['can_login'] = ColumnMetadata(table_meta, 'can_login', cassandra.cqltypes.BooleanType) - elif tablename == 'role_permissions': - ks_meta = KeyspaceMetadata(ksname, True, None, None) - table_meta = TableMetadata(ks_meta, 'role_permissions') - table_meta.columns['role'] = ColumnMetadata(table_meta, 'role', cassandra.cqltypes.UTF8Type) - table_meta.columns['resource'] = ColumnMetadata(table_meta, 'resource', cassandra.cqltypes.UTF8Type) - table_meta.columns['permission'] = ColumnMetadata(table_meta, 'permission', cassandra.cqltypes.UTF8Type) - else: - raise ColumnFamilyNotFound("Column family {} not found".format(tablename)) - - def get_index_meta(self, ksname, idxname): - if ksname is None: - ksname = self.current_keyspace - ksmeta = self.get_keyspace_meta(ksname) - - if idxname not in ksmeta.indexes: - raise IndexNotFound("Index {} not found".format(idxname)) - - return ksmeta.indexes[idxname] - - def get_view_meta(self, ksname, viewname): - if ksname is None: - ksname = self.current_keyspace - ksmeta = self.get_keyspace_meta(ksname) - - if viewname not in ksmeta.views: - raise MaterializedViewNotFound("Materialized view '{}' not found".format(viewname)) - return ksmeta.views[viewname] - - def get_object_meta(self, ks, name): - if name is None: - if ks and ks in self.conn.metadata.keyspaces: - return self.conn.metadata.keyspaces[ks] - elif self.current_keyspace is None: - raise ObjectNotFound("'{}' not found in keyspaces".format(ks)) - else: - name = ks - ks = self.current_keyspace - - if ks is None: - ks = self.current_keyspace - - ksmeta = self.get_keyspace_meta(ks) - - if name in ksmeta.tables: - return ksmeta.tables[name] - elif name in ksmeta.indexes: - return ksmeta.indexes[name] - elif name in ksmeta.views: - return ksmeta.views[name] - - raise ObjectNotFound("'{}' not found in keyspace '{}'".format(name, ks)) - - def get_trigger_names(self, ksname=None): - if ksname is None: - ksname = self.current_keyspace - - return [trigger.name - for table in list(self.get_keyspace_meta(ksname).tables.values()) - for trigger in list(table.triggers.values())] - - def reset_statement(self): - self.reset_prompt() - self.statement.truncate(0) - self.statement.seek(0) - self.empty_lines = 0 - - def reset_prompt(self): - if self.current_keyspace is None: - self.set_prompt(self.default_prompt, True) - else: - self.set_prompt(self.keyspace_prompt.format(self.current_keyspace), True) - - def set_continue_prompt(self): - if self.empty_lines >= 3: - self.set_prompt("Statements are terminated with a ';'. You can press CTRL-C to cancel an incomplete statement.") - self.empty_lines = 0 - return - if self.current_keyspace is None: - self.set_prompt(self.continue_prompt) - else: - spaces = ' ' * len(str(self.current_keyspace)) - self.set_prompt(self.keyspace_continue_prompt.format(spaces)) - self.empty_lines = self.empty_lines + 1 if not self.lastcmd else 0 - - @contextmanager - def prepare_loop(self): - readline = None - if self.tty and self.completekey: - try: - import readline - except ImportError: - pass - else: - old_completer = readline.get_completer() - readline.set_completer(self.complete) - if readline.__doc__ is not None and 'libedit' in readline.__doc__: - readline.parse_and_bind("bind -e") - readline.parse_and_bind("bind '" + self.completekey + "' rl_complete") - readline.parse_and_bind("bind ^R em-inc-search-prev") - else: - readline.parse_and_bind(self.completekey + ": complete") - # start coverage collection if requested, unless in subshell - if self.coverage and not self.is_subshell: - # check for coveragerc file, write it if missing - if os.path.exists(CQL_DIR): - self.coveragerc_path = os.path.join(CQL_DIR, '.coveragerc') - covdata_path = os.path.join(CQL_DIR, '.coverage') - if not os.path.isfile(self.coveragerc_path): - with open(self.coveragerc_path, 'w') as f: - f.writelines(["[run]\n", - "concurrency = multiprocessing\n", - "data_file = {}\n".format(covdata_path), - "parallel = true\n"] - ) - # start coverage - import coverage - self.cov = coverage.Coverage(config_file=self.coveragerc_path) - self.cov.start() - try: - yield - finally: - if readline is not None: - readline.set_completer(old_completer) - if self.coverage and not self.is_subshell: - self.stop_coverage() - - def get_input_line(self, prompt=''): - if self.tty: - self.lastcmd = input(str(prompt)) - line = self.lastcmd + '\n' - else: - self.lastcmd = self.stdin.readline() - line = self.lastcmd - if not len(line): - raise EOFError - self.lineno += 1 - return line - - def use_stdin_reader(self, until='', prompt=''): - until += '\n' - while True: - try: - newline = self.get_input_line(prompt=prompt) - except EOFError: - return - if newline == until: - return - yield newline - - def cmdloop(self, intro=None): - """ - Adapted from cmd.Cmd's version, because there is literally no way with - cmd.Cmd.cmdloop() to tell the difference between "EOF" showing up in - input and an actual EOF. - """ - with self.prepare_loop(): - while not self.stop: - try: - if self.single_statement: - line = self.single_statement - self.stop = True - else: - line = self.get_input_line(self.prompt) - self.statement.write(line) - if self.onecmd(self.statement.getvalue()): - self.reset_statement() - except EOFError: - self.handle_eof() - except CQL_ERRORS as cqlerr: - self.printerr(cqlerr.message) - except KeyboardInterrupt: - self.reset_statement() - print('') - - def strip_comment_blocks(self, statementtext): - comment_block_in_literal_string = re.search('["].*[/][*].*[*][/].*["]', statementtext) - if not comment_block_in_literal_string: - result = re.sub('[/][*].*[*][/]', "", statementtext) - if '*/' in result and '/*' not in result and not self.in_comment: - raise SyntaxError("Encountered comment block terminator without being in comment block") - if '/*' in result: - result = re.sub('[/][*].*', "", result) - self.in_comment = True - if '*/' in result: - result = re.sub('.*[*][/]', "", result) - self.in_comment = False - if self.in_comment and not re.findall('[/][*]|[*][/]', statementtext): - result = '' - return result - return statementtext - - def onecmd(self, statementtext): - """ - Returns true if the statement is complete and was handled (meaning it - can be reset). - """ - statementtext = self.strip_comment_blocks(statementtext) - try: - statements, endtoken_escaped = cqlruleset.cql_split_statements(statementtext) - except pylexotron.LexingError as e: - if self.show_line_nums: - self.printerr('Invalid syntax at line {0}, char {1}' - .format(e.linenum, e.charnum)) - else: - self.printerr('Invalid syntax at char {0}'.format(e.charnum)) - statementline = statementtext.split('\n')[e.linenum - 1] - self.printerr(' {0}'.format(statementline)) - self.printerr(' {0}^'.format(' ' * e.charnum)) - return True - - while statements and not statements[-1]: - statements = statements[:-1] - if not statements: - return True - if endtoken_escaped or statements[-1][-1][0] != 'endtoken': - self.set_continue_prompt() - return - for st in statements: - try: - self.handle_statement(st, statementtext) - except Exception as e: - if self.debug: - traceback.print_exc() - else: - self.printerr(e) - return True - - def handle_eof(self): - if self.tty: - print('') - statement = self.statement.getvalue() - if statement.strip(): - if not self.onecmd(statement): - self.printerr('Incomplete statement at end of file') - self.do_exit() - - def handle_statement(self, tokens, srcstr): - # Concat multi-line statements and insert into history - if readline is not None: - nl_count = srcstr.count("\n") - - new_hist = srcstr.replace("\n", " ").rstrip() - - if nl_count > 1 and self.last_hist != new_hist: - readline.add_history(new_hist) - - self.last_hist = new_hist - cmdword = tokens[0][1] - if cmdword == '?': - cmdword = 'help' - custom_handler = getattr(self, 'do_' + cmdword.lower(), None) - if custom_handler: - parsed = cqlruleset.cql_whole_parse_tokens(tokens, srcstr=srcstr, - startsymbol='cqlshCommand') - if parsed and not parsed.remainder: - # successful complete parse - return custom_handler(parsed) - else: - return self.handle_parse_error(cmdword, tokens, parsed, srcstr) - return self.perform_statement(cqlruleset.cql_extract_orig(tokens, srcstr)) - - def handle_parse_error(self, cmdword, tokens, parsed, srcstr): - if cmdword.lower() in ('select', 'insert', 'update', 'delete', 'truncate', - 'create', 'drop', 'alter', 'grant', 'revoke', - 'batch', 'list'): - # hey, maybe they know about some new syntax we don't. type - # assumptions won't work, but maybe the query will. - return self.perform_statement(cqlruleset.cql_extract_orig(tokens, srcstr)) - if parsed: - self.printerr('Improper %s command (problem at %r).' % (cmdword, parsed.remainder[0])) - else: - self.printerr(f'Improper {cmdword} command.') - - def do_use(self, parsed): - ksname = parsed.get_binding('ksname') - success, _ = self.perform_simple_statement(SimpleStatement(parsed.extract_orig())) - if success: - if ksname[0] == '"' and ksname[-1] == '"': - self.current_keyspace = self.cql_unprotect_name(ksname) - else: - self.current_keyspace = ksname.lower() - - def do_select(self, parsed): - tracing_was_enabled = self.tracing_enabled - ksname = parsed.get_binding('ksname') - stop_tracing = ksname == 'system_traces' or (ksname is None and self.current_keyspace == 'system_traces') - self.tracing_enabled = self.tracing_enabled and not stop_tracing - statement = parsed.extract_orig() - self.perform_statement(statement) - self.tracing_enabled = tracing_was_enabled - - def perform_statement(self, statement): - - stmt = SimpleStatement(statement, consistency_level=self.consistency_level, serial_consistency_level=self.serial_consistency_level, fetch_size=self.page_size if self.use_paging else None) - success, future = self.perform_simple_statement(stmt) - - if future: - if future.warnings: - self.print_warnings(future.warnings) - - if self.tracing_enabled: - try: - for trace in future.get_all_query_traces(max_wait_per=self.max_trace_wait, query_cl=self.consistency_level): - print_trace(self, trace) - except TraceUnavailable: - msg = "Statement trace did not complete within %d seconds; trace data may be incomplete." % (self.session.max_trace_wait,) - self.writeresult(msg, color=RED) - for trace_id in future.get_query_trace_ids(): - self.show_session(trace_id, partial_session=True) - except Exception as err: - self.printerr("Unable to fetch query trace: %s" % (str(err),)) - - return success - - def parse_for_select_meta(self, query_string): - try: - parsed = cqlruleset.cql_parse(query_string)[1] - except IndexError: - return None - ks = self.cql_unprotect_name(parsed.get_binding('ksname', None)) - name = self.cql_unprotect_name(parsed.get_binding('cfname', None)) - try: - return self.get_table_meta(ks, name) - except ColumnFamilyNotFound: - try: - return self.get_view_meta(ks, name) - except MaterializedViewNotFound: - raise ObjectNotFound("'{}' not found in keyspace '{}'".format(name, ks)) - - def parse_for_update_meta(self, query_string): - try: - parsed = cqlruleset.cql_parse(query_string)[1] - except IndexError: - return None - ks = self.cql_unprotect_name(parsed.get_binding('ksname', None)) - cf = self.cql_unprotect_name(parsed.get_binding('cfname')) - return self.get_table_meta(ks, cf) - - def perform_simple_statement(self, statement): - if not statement: - return False, None - - future = self.session.execute_async(statement, trace=self.tracing_enabled) - result = None - try: - result = future.result() - except CQL_ERRORS as err: - err_msg = err.message if hasattr(err, 'message') else str(err) - self.printerr(str(err.__class__.__name__) + ": " + err_msg) - except Exception: - import traceback - self.printerr(traceback.format_exc()) - - # Even if statement failed we try to refresh schema if not agreed (see CASSANDRA-9689) - if not future.is_schema_agreed: - try: - self.conn.refresh_schema_metadata(5) # will throw exception if there is a schema mismatch - except Exception: - self.printerr("Warning: schema version mismatch detected; check the schema versions of your " - "nodes in system.local and system.peers.") - self.conn.refresh_schema_metadata(-1) - - if result is None: - return False, None - - if statement.query_string[:6].lower() == 'select': - self.print_result(result, self.parse_for_select_meta(statement.query_string)) - elif statement.query_string.lower().startswith("list users") or statement.query_string.lower().startswith("list roles"): - self.print_result(result, self.get_table_meta('system_auth', 'roles')) - elif statement.query_string.lower().startswith("list"): - self.print_result(result, self.get_table_meta('system_auth', 'role_permissions')) - elif result: - # CAS INSERT/UPDATE - self.writeresult("") - self.print_static_result(result, self.parse_for_update_meta(statement.query_string), with_header=True, tty=self.tty) - self.flush_output() - return True, future - - def print_result(self, result, table_meta): - self.decoding_errors = [] - - self.writeresult("") - - def print_all(result, table_meta, tty): - # Return the number of rows in total - num_rows = 0 - is_first = True - while True: - # Always print for the first page even it is empty - if result.current_rows or is_first: - with_header = is_first or tty - self.print_static_result(result, table_meta, with_header, tty, num_rows) - num_rows += len(result.current_rows) - if result.has_more_pages: - if self.shunted_query_out is None and tty: - # Only pause when not capturing. - input("---MORE---") - result.fetch_next_page() - else: - if not tty: - self.writeresult("") - break - is_first = False - return num_rows - - num_rows = print_all(result, table_meta, self.tty) - self.writeresult("(%d rows)" % num_rows) - - if self.decoding_errors: - for err in self.decoding_errors[:2]: - self.writeresult(err.message(), color=RED) - if len(self.decoding_errors) > 2: - self.writeresult('%d more decoding errors suppressed.' - % (len(self.decoding_errors) - 2), color=RED) - - def print_static_result(self, result, table_meta, with_header, tty, row_count_offset=0): - if not result.column_names and not table_meta: - return - - column_names = result.column_names or list(table_meta.columns.keys()) - formatted_names = [self.myformat_colname(name, table_meta) for name in column_names] - if not result.current_rows: - # print header only - self.print_formatted_result(formatted_names, None, with_header=True, tty=tty) - return - - cql_types = [] - if result.column_types: - ks_name = table_meta.keyspace_name if table_meta else self.current_keyspace - ks_meta = self.conn.metadata.keyspaces.get(ks_name, None) - cql_types = [CqlType(cql_typename(t), ks_meta) for t in result.column_types] - - formatted_values = [list(map(self.myformat_value, [row[c] for c in column_names], cql_types)) for row in result.current_rows] - - if self.expand_enabled: - self.print_formatted_result_vertically(formatted_names, formatted_values, row_count_offset) - else: - self.print_formatted_result(formatted_names, formatted_values, with_header, tty) - - def print_formatted_result(self, formatted_names, formatted_values, with_header, tty): - # determine column widths - widths = [n.displaywidth for n in formatted_names] - if formatted_values is not None: - for fmtrow in formatted_values: - for num, col in enumerate(fmtrow): - widths[num] = max(widths[num], col.displaywidth) - - # print header - if with_header: - header = ' | '.join(hdr.ljust(w, color=self.color) for (hdr, w) in zip(formatted_names, widths)) - self.writeresult(' ' + header.rstrip()) - self.writeresult('-%s-' % '-+-'.join('-' * w for w in widths)) - - # stop if there are no rows - if formatted_values is None: - self.writeresult("") - return - - # print row data - for row in formatted_values: - line = ' | '.join(col.rjust(w, color=self.color) for (col, w) in zip(row, widths)) - self.writeresult(' ' + line) - - if tty: - self.writeresult("") - - def print_formatted_result_vertically(self, formatted_names, formatted_values, row_count_offset): - max_col_width = max([n.displaywidth for n in formatted_names]) - max_val_width = max([n.displaywidth for row in formatted_values for n in row]) - - # for each row returned, list all the column-value pairs - for i, row in enumerate(formatted_values): - self.writeresult("@ Row %d" % (row_count_offset + i + 1)) - self.writeresult('-%s-' % '-+-'.join(['-' * max_col_width, '-' * max_val_width])) - for field_id, field in enumerate(row): - column = formatted_names[field_id].ljust(max_col_width, color=self.color) - value = field.ljust(field.displaywidth, color=self.color) - self.writeresult(' ' + " | ".join([column, value])) - self.writeresult('') - - def print_warnings(self, warnings): - if warnings is None or len(warnings) == 0: - return - - self.writeresult('') - self.writeresult('Warnings :') - for warning in warnings: - self.writeresult(warning) - self.writeresult('') - - def emptyline(self): - pass - - def parseline(self, line): - # this shouldn't be needed - raise NotImplementedError - - def complete(self, text, state): - if readline is None: - return - if state == 0: - try: - self.completion_matches = self.find_completions(text) - except Exception: - if debug_completion: - import traceback - traceback.print_exc() - else: - raise - try: - return self.completion_matches[state] - except IndexError: - return None - - def find_completions(self, text): - curline = readline.get_line_buffer() - prevlines = self.statement.getvalue() - wholestmt = prevlines + curline - begidx = readline.get_begidx() + len(prevlines) - stuff_to_complete = wholestmt[:begidx] - return cqlruleset.cql_complete(stuff_to_complete, text, cassandra_conn=self, - debug=debug_completion, startsymbol='cqlshCommand') - - def set_prompt(self, prompt, prepend_user=False): - if prepend_user and self.username: - self.prompt = "{0}@{1}".format(self.username, prompt) - return - self.prompt = prompt - - def cql_unprotect_name(self, namestr): - if namestr is None: - return - return cqlruleset.dequote_name(namestr) - - def cql_unprotect_value(self, valstr): - if valstr is not None: - return cqlruleset.dequote_value(valstr) - - def _columnize_unicode(self, name_list): - """ - Used when columnizing identifiers that may contain unicode - """ - names = [n for n in name_list] - cmd.Cmd.columnize(self, names) - print('') - - def do_describe(self, parsed): - - """ - DESCRIBE [cqlsh only] - - (DESC may be used as a shorthand.) - - Outputs information about the connected Cassandra cluster, or about - the data objects stored in the cluster. Use in one of the following ways: - - DESCRIBE KEYSPACES - - Output the names of all keyspaces. - - DESCRIBE KEYSPACE [] - - Output CQL commands that could be used to recreate the given keyspace, - and the objects in it (such as tables, types, functions, etc.). - In some cases, as the CQL interface matures, there will be some metadata - about a keyspace that is not representable with CQL. That metadata will not be shown. - The '' argument may be omitted, in which case the current - keyspace will be described. - - DESCRIBE TABLES - - Output the names of all tables in the current keyspace, or in all - keyspaces if there is no current keyspace. - - DESCRIBE TABLE [.] - - Output CQL commands that could be used to recreate the given table. - In some cases, as above, there may be table metadata which is not - representable and which will not be shown. - - DESCRIBE INDEX - - Output the CQL command that could be used to recreate the given index. - In some cases, there may be index metadata which is not representable - and which will not be shown. - - DESCRIBE MATERIALIZED VIEW - - Output the CQL command that could be used to recreate the given materialized view. - In some cases, there may be materialized view metadata which is not representable - and which will not be shown. - - DESCRIBE CLUSTER - - Output information about the connected Cassandra cluster, such as the - cluster name, and the partitioner and snitch in use. When you are - connected to a non-system keyspace, also shows endpoint-range - ownership information for the Cassandra ring. - - DESCRIBE [FULL] SCHEMA - - Output CQL commands that could be used to recreate the entire (non-system) schema. - Works as though "DESCRIBE KEYSPACE k" was invoked for each non-system keyspace - k. Use DESCRIBE FULL SCHEMA to include the system keyspaces. - - DESCRIBE TYPES - - Output the names of all user-defined-types in the current keyspace, or in all - keyspaces if there is no current keyspace. - - DESCRIBE TYPE [.] - - Output the CQL command that could be used to recreate the given user-defined-type. - - DESCRIBE FUNCTIONS - - Output the names of all user-defined-functions in the current keyspace, or in all - keyspaces if there is no current keyspace. - - DESCRIBE FUNCTION [.] - - Output the CQL command that could be used to recreate the given user-defined-function. - - DESCRIBE AGGREGATES - - Output the names of all user-defined-aggregates in the current keyspace, or in all - keyspaces if there is no current keyspace. - - DESCRIBE AGGREGATE [.] - - Output the CQL command that could be used to recreate the given user-defined-aggregate. - - DESCRIBE - - Output CQL commands that could be used to recreate the entire object schema, - where object can be either a keyspace or a table or an index or a materialized - view (in this order). - """ - stmt = SimpleStatement(parsed.extract_orig(), consistency_level=cassandra.ConsistencyLevel.LOCAL_ONE, fetch_size=self.page_size if self.use_paging else None) - future = self.session.execute_async(stmt) - - if self.connection_versions['build'][0] < '4': - print('\nWARN: DESCRIBE|DESC was moved to server side in Cassandra 4.0. As a consequence DESRIBE|DESC ' - 'will not work in cqlsh %r connected to Cassandra %r, the version that you are connected to. ' - 'DESCRIBE does not exist server side prior Cassandra 4.0.' - % (version, self.connection_versions['build'])) - else: - try: - result = future.result() - - what = parsed.matched[1][1].lower() - - if what in ('columnfamilies', 'tables', 'types', 'functions', 'aggregates'): - self.describe_list(result) - elif what == 'keyspaces': - self.describe_keyspaces(result) - elif what == 'cluster': - self.describe_cluster(result) - elif what: - self.describe_element(result) - - except CQL_ERRORS as err: - err_msg = err.message if hasattr(err, 'message') else str(err) - self.printerr(err_msg.partition("message=")[2].strip('"')) - except Exception: - import traceback - self.printerr(traceback.format_exc()) - - if future: - if future.warnings: - self.print_warnings(future.warnings) - - do_desc = do_describe - - def describe_keyspaces(self, rows): - """ - Print the output for a DESCRIBE KEYSPACES query - """ - names = [r['name'] for r in rows] - - print('') - cmd.Cmd.columnize(self, names) - print('') - - def describe_list(self, rows): - """ - Print the output for all the DESCRIBE queries for element names (e.g DESCRIBE TABLES, DESCRIBE FUNCTIONS ...) - """ - keyspace = None - names = list() - for row in rows: - if row['keyspace_name'] != keyspace: - if keyspace is not None: - self.print_keyspace_element_names(keyspace, names) - - keyspace = row['keyspace_name'] - names = list() - - names.append(str(row['name'])) - - if keyspace is not None: - self.print_keyspace_element_names(keyspace, names) - print('') - - def print_keyspace_element_names(self, keyspace, names): - print('') - if self.current_keyspace is None: - print('Keyspace %s' % (keyspace)) - print('---------%s' % ('-' * len(keyspace))) - cmd.Cmd.columnize(self, names) - - def describe_element(self, rows): - """ - Print the output for all the DESCRIBE queries where an element name as been specified (e.g DESCRIBE TABLE, DESCRIBE INDEX ...) - """ - for row in rows: - print('') - self.query_out.write(row['create_statement']) - print('') - - def describe_cluster(self, rows): - """ - Print the output for a DESCRIBE CLUSTER query. - - If a specified keyspace was in use the returned ResultSet will contains a 'range_ownership' column, - otherwise not. - """ - for row in rows: - print('\nCluster: %s' % row['cluster']) - print('Partitioner: %s' % row['partitioner']) - print('Snitch: %s\n' % row['snitch']) - if 'range_ownership' in row: - print("Range ownership:") - for entry in list(row['range_ownership'].items()): - print(' %39s [%s]' % (entry[0], ', '.join([host for host in entry[1]]))) - print('') - - def do_copy(self, parsed): - r""" - COPY [cqlsh only] - - COPY x FROM: Imports CSV data into a Cassandra table - COPY x TO: Exports data from a Cassandra table in CSV format. - - COPY [ ( column [, ...] ) ] - FROM ( '' | STDIN ) - [ WITH