ksql.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. from __future__ import absolute_import
  18. import logging
  19. from django.core.urlresolvers import reverse
  20. from django.utils.translation import ugettext as _
  21. from desktop.lib.i18n import force_unicode
  22. from kafka.ksql_client import KSqlApi
  23. from notebook.connectors.base import Api, QueryError
  24. LOG = logging.getLogger(__name__)
  25. def query_error_handler(func):
  26. def decorator(*args, **kwargs):
  27. try:
  28. return func(*args, **kwargs)
  29. except Exception as e:
  30. message = force_unicode(str(e))
  31. raise QueryError(message)
  32. return decorator
  33. class KafkaApi(Api):
  34. def __init__(self, user, interpreter=None):
  35. Api.__init__(self, user, interpreter=interpreter)
  36. self.db = KSqlApi(user=user)
  37. @query_error_handler
  38. def execute(self, notebook, snippet):
  39. if self.db is None:
  40. raise AuthenticationRequired()
  41. data, description = query_and_fetch(self.db, snippet['statement'], 1000)
  42. has_result_set = data is not None
  43. return {
  44. 'sync': True,
  45. 'has_result_set': has_result_set,
  46. 'result': {
  47. 'has_more': False,
  48. 'data': data if has_result_set else [],
  49. 'meta': [{
  50. 'name': col[0],
  51. 'type': col[1],
  52. 'comment': ''
  53. } for col in description] if has_result_set else [],
  54. 'type': 'table'
  55. }
  56. }
  57. @query_error_handler
  58. def check_status(self, notebook, snippet):
  59. return {'status': 'available'}
  60. @query_error_handler
  61. def autocomplete(self, snippet, database=None, table=None, column=None, nested=None):
  62. response = {}
  63. try:
  64. if database is None:
  65. response['databases'] = ['default']
  66. elif table is None:
  67. response['tables_meta'] = self.db.show_tables()
  68. else:
  69. response = {
  70. u'status': 0,
  71. u'comment': u'test test test 22',
  72. u'hdfs_link': u'/filebrowser/view=/user/hive/warehouse/web_logs',
  73. u'extended_columns': [
  74. {u'comment': u'', u'type': u'bigint', u'name': u'_version_'},
  75. {u'comment': u'The app', u'type': u'string', u'name': u'app'},
  76. {u'comment': u'test test test 22', u'type': u'smallint', u'name': u'bytes'},
  77. {u'comment': u'The citi', u'type': u'string', u'name': u'city'},
  78. {u'comment': u'', u'type': u'string', u'name': u'client_ip'},
  79. {u'comment': u'', u'type': u'tinyint', u'name': u'code'},
  80. {u'comment': u'', u'type': u'string', u'name': u'country_code'},
  81. {u'comment': u'', u'type': u'string', u'name': u'country_code3'},
  82. {u'comment': u'', u'type': u'string', u'name': u'country_name'},
  83. {u'comment': u'', u'type': u'string', u'name': u'device_family'},
  84. {u'comment': u'', u'type': u'string', u'name': u'extension'},
  85. {u'comment': u'', u'type': u'float', u'name': u'latitude'},
  86. {u'comment': u'', u'type': u'float', u'name': u'longitude'},
  87. {u'comment': u'', u'type': u'string', u'name': u'method'},
  88. {u'comment': u'', u'type': u'string', u'name': u'os_family'},
  89. {u'comment': u'', u'type': u'string', u'name': u'os_major'},
  90. {u'comment': u'', u'type': u'string', u'name': u'protocol'},
  91. {u'comment': u'', u'type': u'string', u'name': u'record'},
  92. {u'comment': u'', u'type': u'string', u'name': u'referer'},
  93. {u'comment': u'', u'type': u'bigint', u'name': u'region_code'},
  94. {u'comment': u'', u'type': u'string', u'name': u'request'},
  95. {u'comment': u'', u'type': u'string', u'name': u'subapp'},
  96. {u'comment': u'', u'type': u'string', u'name': u'time'},
  97. {u'comment': u'', u'type': u'string', u'name': u'url'},
  98. {u'comment': u'', u'type': u'string', u'name': u'user_agent'},
  99. {u'comment': u'', u'type': u'string', u'name': u'user_agent_family'},
  100. {u'comment': u'', u'type': u'string', u'name': u'user_agent_major'},
  101. {u'comment': u'', u'type': u'string', u'name': u'id'},
  102. {u'comment': u'', u'type': u'string', u'name': u'date'}
  103. ],
  104. u'support_updates': False,
  105. u'partition_keys': [
  106. {u'type': u'string', u'name': u'date'}
  107. ],
  108. u'columns': [u'_version_', u'app', u'bytes', u'city', u'client_ip', u'code', u'country_code', u'country_code3', u'country_name', u'device_family', u'extension', u'latitude', u'longitude', u'method', u'os_family', u'os_major', u'protocol', u'record', u'referer', u'region_code', u'request', u'subapp', u'time', u'url', u'user_agent', u'user_agent_family', u'user_agent_major', u'id', u'date'],
  109. u'is_view': False
  110. }
  111. except Exception as e:
  112. LOG.warn('Autocomplete data fetching error: %s' % e)
  113. response['code'] = 500
  114. response['error'] = e.message
  115. return response