#!/usr/bin/env python
# -- coding: utf-8 --
# Licensed to Cloudera, Inc. under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  Cloudera, Inc. licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import logging

from django.utils.translation import ugettext as _

from desktop.lib.django_util import JsonResponse
from desktop.lib.i18n import force_unicode
from metadata.manager_client import ManagerApi

from kafka.conf import has_kafka_api
from kafka.kafka_client import KafkaApi, KafkaApiException


LOG = logging.getLogger(__name__)


def error_handler(view_fn):
  def decorator(*args, **kwargs):
    status = 500
    response = {
      'message': ''
    }

    try:
      return view_fn(*args, **kwargs)
    except KafkaApiException as e:
      try:
        response['message'] = json.loads(e.message)
      except Exception:
        response['message'] = force_unicode(e.message)
    except Exception as e:
      message = force_unicode(e)
      response['message'] = message
      LOG.exception(message)

    return JsonResponse(response, status=status)
  return decorator


@error_handler
def list_topics(request):
  return JsonResponse({
    'status': 0,
    'topics': [
      {'name': topic} for topic in get_topics(request.user)
    ]
  })


@error_handler
def list_topic(request):
  name = request.POST.get('name')

  topic = get_topic(name)
  topic['name'] = name
  topic['type'] = 'topic'

  return JsonResponse({
    'status': 0,
    'topic': topic
  })


@error_handler
def create_topic(request):
  name = request.POST.get('name')
  partitions = request.POST.get('partitions', 1)
  replication_factor = request.POST.get('replication_factor', 1)

  status = KafkaApi().create_topic(name, partitions, replication_factor)

  return JsonResponse({
    'status': status,
    'topic': {
      'name': name,
      'partitions': partitions,
      'replication_factor': replication_factor
    }
  })


def get_topics(user):
  if has_kafka_api():
    return KafkaApi().topics()
  else:
    from metadata.models.bigquery_client import _get_notebook_api
    data = {
      'snippet': {},
      'database': 'topics'
    }

    return [
      topic['name']
      for topic in _get_notebook_api(user, connector_id=56).autocomplete(**data)['tables_meta']
      if not topic['name'].startswith('__')
    ]


def get_topic_data(user, name):
  from metadata.models.bigquery_client import _get_notebook_api

  data = _get_notebook_api(user, connector_id=56).get_sample_data(snippet={})
  print(data)
  return data


def get_topic(name):
  if has_kafka_api():
    pass
  else:
    manager = ManagerApi()
    broker_host = manager.get_kafka_brokers().split(',')[0].split(':')[0]
    return manager.get_kafka_topics(broker_host)[name]

@@ -1025,16 +1025,62 @@
 nagerApi
+%0Afrom notebook.models import _get_notebook_api
 %0A%0Afrom k
@@ -2664,74 +2664,8 @@
 se:%0A
-    from metadata.models.bigquery_client import _get_notebook_api%0A
     
@@ -2945,73 +2945,8 @@
 e):%0A
-  from metadata.models.bigquery_client import _get_notebook_api%0A%0A
   da

Use the proper notebook api client