datahub/implement.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF 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 time
import urllib3

from .batch.schema_registry_client import SchemaRegistryClient
from .models.params import *
from .models.results import *
from .auth import AliyunAccount
from .exceptions import InvalidParameterException, InvalidOperationException
from .models import ShardState, OffsetBase, SubscriptionState, FieldType, OffsetWithSession
from .rest import Path
from .rest import RestClient
from .utils import check_project_name_valid, check_topic_name_valid, check_type, check_positive, \
    to_text, ErrorMessage, check_empty, check_negative

urllib3.disable_warnings()


class DataHubJson(object):
    """
    Datahub json client
    """

    def __init__(self, access_id, access_key, endpoint=None, compress_format=None, enable_schema_register=False, **kwargs):
        self._account = kwargs.pop('account', None)
        if self._account is None:
            security_token = kwargs.pop('security_token', '')
            self._account = AliyunAccount(access_id=access_id, access_key=access_key, security_token=security_token)
        self._endpoint = endpoint
        self._compress_format = compress_format
        self._rest_client = RestClient(self._account, self._endpoint, **kwargs)

    def list_project(self):
        url = Path.PROJECTS

        content = self._rest_client.get(url)

        result = ListProjectResult.parse_content(content)
        return result

    def create_project(self, project_name, comment):
        if not check_project_name_valid(project_name):
            raise InvalidParameterException(ErrorMessage.INVALID_PROJECT_NAME)

        url = Path.PROJECT % project_name
        request_param = CreateProjectRequestParams(comment)

        self._rest_client.post(url, data=request_param.content())

    def get_project(self, project_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')

        url = Path.PROJECT % project_name

        content = self._rest_client.get(url)
        result = GetProjectResult.parse_content(to_text(content), project_name=project_name)
        return result

    def update_project(self, project_name, comment):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')

        url = Path.PROJECT % project_name
        request_param = UpdateProjectRequestParams(comment)

        self._rest_client.put(url, data=request_param.content())

    def delete_project(self, project_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')

        url = Path.PROJECT % project_name

        self._rest_client.delete(url)

    def list_topic(self, project_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')

        url = Path.TOPICS % project_name

        content = self._rest_client.get(url)

        result = ListTopicResult.parse_content(content)
        return result

    def create_blob_topic(self, project_name, topic_name, shard_count, life_cycle, extend_mode, comment):
        self.__create_topic(project_name, topic_name, shard_count, life_cycle, RecordType.BLOB, comment, extend_mode)

    def create_tuple_topic(self, project_name, topic_name, shard_count, life_cycle, record_schema, extend_mode,
                           comment):
        self.__create_topic(project_name, topic_name, shard_count, life_cycle, RecordType.TUPLE, comment, extend_mode,
                            record_schema)

    def get_topic(self, project_name, topic_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.TOPIC % (project_name, topic_name)

        content = self._rest_client.get(url)

        result = GetTopicResult.parse_content(to_text(content), project_name=project_name,
                                              topic_name=topic_name)
        return result

    def update_topic(self, project_name, topic_name, life_cycle, comment):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if not check_positive(life_cycle):
            raise InvalidParameterException(ErrorMessage.PARAMETER_NOT_POSITIVE % 'life_cycle')

        url = Path.TOPIC % (project_name, topic_name)
        request_param = UpdateTopicRequestParams(life_cycle, comment)

        self._rest_client.put(url, data=request_param.content())

    def delete_topic(self, project_name, topic_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.TOPIC % (project_name, topic_name)

        self._rest_client.delete(url)

    def append_field(self, project_name, topic_name, field_name, field_type):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(field_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'field_name')
        if not check_type(field_type, FieldType):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE %
                                            ('field_type', FieldType.__name__))

        url = Path.TOPIC % (project_name, topic_name)
        request_param = AppendFieldParams(field_name, field_type)

        self._rest_client.post(url, data=request_param.content())

    def wait_shards_ready(self, project_name, topic_name, timeout=30):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_negative(timeout):
            raise InvalidParameterException(ErrorMessage.PARAMETER_NEGATIVE % 'timeout')

        current_time = time.time()
        end_time = current_time + timeout

        while current_time <= end_time:
            if self.__is_shard_load_completed(project_name, topic_name):
                return
            time.sleep(1)
            current_time = time.time()

        raise DatahubException(ErrorMessage.WAIT_SHARD_TIMEOUT)

    def list_shard(self, project_name, topic_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.SHARDS % (project_name, topic_name)

        content = self._rest_client.get(url)

        result = ListShardResult.parse_content(content)
        return result

    def merge_shard(self, project_name, topic_name, shard_id, adj_shard_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')
        if check_empty(adj_shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'adj_shard_id')

        url = Path.SHARDS % (project_name, topic_name)
        request_param = MergeShardRequestParams(shard_id, adj_shard_id)

        content = self._rest_client.post(url, data=request_param.content())

        result = MergeShardResult.parse_content(content)
        return result

    def split_shard(self, project_name, topic_name, shard_id, split_key=''):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')
        if check_empty(split_key):
            shard_info = self.__get_shard(project_name, topic_name, shard_id)
            if not check_empty(shard_info):
                begin_hash_key = int(shard_info.begin_hash_key, 16)
                end_hash_key = int(shard_info.end_hash_key, 16)
                split_key = hex(int((begin_hash_key + end_hash_key) // 2))[2:34]

        url = Path.SHARDS % (project_name, topic_name)
        request_param = SplitShardRequestParams(shard_id, split_key)

        content = self._rest_client.post(url, data=request_param.content())

        result = SplitShardResult.parse_content(content)
        return result

    def extend_shard(self, project_name, topic_name, shard_count):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if not check_positive(shard_count):
            raise InvalidParameterException(ErrorMessage.PARAMETER_NOT_POSITIVE % 'shard_count')

        url = Path.SHARDS % (project_name, topic_name)
        request_param = ExtendShardRequestParams(shard_count)

        self._rest_client.post(url, data=request_param.content())

    def get_cursor(self, project_name, topic_name, shard_id, cursor_type, param=-1):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')
        if not check_type(cursor_type, CursorType):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE % ('cursor_type', CursorType.__name__))
        if CursorType.SYSTEM_TIME == cursor_type and param < 0:
            raise InvalidParameterException(ErrorMessage.MISSING_SYSTEM_TIME)
        elif CursorType.SEQUENCE == cursor_type and param < 0:
            raise InvalidParameterException(ErrorMessage.MISSING_SEQUENCE)

        url = Path.SHARD % (project_name, topic_name, shard_id)
        request_param = GetCursorRequestParams(cursor_type, param)

        content = self._rest_client.post(url, data=request_param.content())

        result = GetCursorResult.parse_content(content)
        return result

    def put_records(self, project_name, topic_name, record_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.SHARDS % (project_name, topic_name)

        request_param = PutRecordsRequestParams(record_list)

        content = self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers(),
                                         compress_format=self._compress_format)

        result = PutRecordsResult.parse_content(content)

        return result

    def put_records_by_shard(self, project_name, topic_name, shard_id, record_list):
        raise DatahubException('put_records_by_shard api only support pb mode')

    def get_blob_records(self, project_name, topic_name, sub_id, shard_id, cursor, limit_num):
        return self.__get_records(project_name, topic_name, sub_id, shard_id, cursor, limit_num)

    def get_tuple_records(self, project_name, topic_name, sub_id, shard_id, record_schema, cursor, limit_num):
        return self.__get_records(project_name, topic_name, sub_id, shard_id, cursor, limit_num, record_schema)

    def get_metering_info(self, project_name, topic_name, shard_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')

        url = Path.SHARD % (project_name, topic_name, shard_id)
        request_param = GetMeteringInfoRequestParams()

        content = self._rest_client.post(url, data=request_param.content())

        result = GetMeteringInfoResult.parse_content(content)
        return result

    def list_connector(self, project_name, topic_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTORS % (project_name, topic_name) + "?mode=id"

        content = self._rest_client.get(url)

        result = ListConnectorResult.parse_content(content)
        return result

    def create_connector(self, project_name, topic_name, connector_type, column_fields, config, start_time):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_type.value)
        request_param = CreateConnectorParams(column_fields, config, start_time)

        content = self._rest_client.post(url, data=request_param.content())

        result = CreateConnectorResult.parse_content(content)
        return result

    def update_connector(self, project_name, topic_name, connector_id, config):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)
        request_param = UpdateConnectorParams(config)

        self._rest_client.post(url, data=request_param.content())

    def get_connector(self, project_name, topic_name, connector_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)

        content = self._rest_client.get(url)

        result = GetConnectorResult.parse_content(content)
        return result

    def delete_connector(self, project_name, topic_name, connector_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)

        self._rest_client.delete(url)

    def get_connector_shard_status(self, project_name, topic_name, connector_id, shard_id=''):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)
        request_param = GetConnectorShardStatusParams(shard_id)

        content = self._rest_client.post(url, data=request_param.content())

        result = GetConnectorShardStatusResult.parse_content(content)
        return result

    def reload_connector(self, project_name, topic_name, connector_id, shard_id=''):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)
        request_param = ReloadConnectorParams(shard_id)

        self._rest_client.post(url, data=request_param.content())

    def append_connector_field(self, project_name, topic_name, connector_id, field_name):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(field_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'field_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)
        request_param = AppendConnectorFieldParams(field_name)

        self._rest_client.post(url, data=request_param.content())

    def get_connector_done_time(self, project_name, topic_name, connector_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.DONE_TIME % (project_name, topic_name, connector_id)

        content = self._rest_client.get(url)

        result = GetConnectorDoneTimeResult.parse_content(content)
        return result

    def update_connector_state(self, project_name, topic_name, connector_id, connector_state):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)
        request_param = UpdateConnectorStateParams(connector_state)

        self._rest_client.post(url, data=request_param.content())

    def update_connector_offset(self, project_name, topic_name, connector_id, shard_id, connector_offset):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.CONNECTOR % (project_name, topic_name, connector_id)
        request_param = UpdateConnectorOffsetParams(shard_id, connector_offset)

        self._rest_client.post(url, data=request_param.content())

    def init_and_get_subscription_offset(self, project_name, topic_name, sub_id, shard_ids):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')
        if check_empty(shard_ids):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_ids')

        if isinstance(shard_ids, six.string_types):
            shard_ids = [shard_ids]
        if not check_type(shard_ids, list):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE % ('shard_ids', list.__name__))

        url = Path.OFFSETS % (project_name, topic_name, sub_id)
        request_param = InitAndGetSubscriptionOffsetParams(shard_ids)

        content = self._rest_client.post(url, data=request_param.content())

        result = InitAndGetSubscriptionOffsetResult.parse_content(content)
        return result

    def get_subscription_offset(self, project_name, topic_name, sub_id, shard_ids=None):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')

        if isinstance(shard_ids, six.string_types):
            shard_ids = [shard_ids]

        url = Path.OFFSETS % (project_name, topic_name, sub_id)
        request_param = GetSubscriptionOffsetParams(shard_ids)

        content = self._rest_client.post(url, data=request_param.content())

        result = GetSubscriptionOffsetResult.parse_content(content)
        return result

    def update_subscription_offset(self, project_name, topic_name, sub_id, offsets):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')
        if not check_type(offsets, dict):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE % ('offsets', dict.__name__))

        for (k, v) in offsets.items():
            if isinstance(v, dict):
                offsets[k] = OffsetWithSession.from_dict(v)

        url = Path.OFFSETS % (project_name, topic_name, sub_id)
        request_param = UpdateSubscriptionOffsetParams(offsets)

        self._rest_client.put(url, data=request_param.content())

    def join_group(self, project_name, topic_name, consumer_group, session_timeout):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if consumer_group is None or len(consumer_group) == 0:
            raise InvalidParameterException("Consumer group format is invalid")

        url = Path.SUBSCRIPTION % (project_name, topic_name, consumer_group)
        request_param = JoinGroupParams(session_timeout)

        content = self._rest_client.post(url, data=request_param.content())
        result = JoinGroupResult.parse_content(content)
        return result

    def heart_beat(self, project_name, topic_name, consumer_group, consumer_id, version_id, hold_shard_list, read_end_shard_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if consumer_group is None or len(consumer_group) == 0:
            raise InvalidParameterException("Consumer group format is invalid")

        if hold_shard_list is None:
            raise InvalidParameterException("Hold shard list is none")

        url = Path.SUBSCRIPTION % (project_name, topic_name, consumer_group)
        request_param = HeartBeatParams(consumer_id, version_id, hold_shard_list, read_end_shard_list)
        content = self._rest_client.post(url, data=request_param.content())
        result = HeartBeatResult.parse_content(content)
        return result

    def sync_group(self, project_name, topic_name, consumer_group, consumer_id, version_id, release_shard_list, read_end_shard_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if consumer_group is None or len(consumer_group) == 0:
            raise InvalidParameterException("Consumer group format is invalid")

        url = Path.SUBSCRIPTION % (project_name, topic_name, consumer_group)
        request_param = SyncGroupParams(consumer_id, version_id, release_shard_list, read_end_shard_list)
        content = self._rest_client.post(url, data=request_param.content())
        result = SyncGroupResult.parse_content(content)
        return result

    def leave_group(self, project_name, topic_name, consumer_group, consumer_id, version_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if consumer_group is None or len(consumer_group) == 0:
            raise InvalidParameterException("Consumer group format is invalid")

        url = Path.SUBSCRIPTION % (project_name, topic_name, consumer_group)
        request_param = LeaveGroupParams(consumer_id, version_id)
        content = self._rest_client.post(url, data=request_param.content())
        result = LeaveGroupResult.parse_content(content)
        return result

    def list_topic_schema(self, project_name, topic_name, page_number=-1, page_size=-1):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.TOPIC % (project_name, topic_name)
        request_param = ListTopicSchemaParams(page_number, page_size)
        content = self._rest_client.post(url, data=request_param.content())
        result = ListTopicSchemaResult.parse_content(content)
        return result

    def get_topic_schema(self, project_name, topic_name, schema=None, version_id=-1):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        if schema is None and version_id == -1:
            raise InvalidParameterException("Error. Schema and version_id are both empty.")

        url = Path.TOPIC % (project_name, topic_name)
        request_param = GetTopicSchemaParams(version_id, schema)
        content = self._rest_client.post(url, data=request_param.content())
        result = GetTopicSchemaResult.parse_content(content)
        return result

    def register_topic_schema(self, project_name, topic_name, schema):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.TOPIC % (project_name, topic_name)
        request_param = RegisterTopicSchemaParams(schema)
        content = self._rest_client.post(url, data=request_param.content())
        result = RegisterTopicSchemaResult.parse_content(content)
        return result

    def delete_topic_schema(self, project_name, topic_name, version_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.TOPIC % (project_name, topic_name)
        request_param = DeleteTopicSchemaParams(version_id)
        content = self._rest_client.post(url, data=request_param.content())
        result = DeleteTopicSchemaResult.parse_content(content)
        return result

    # =======================================================
    # internal api
    # =======================================================

    def create_subscription(self, project_name, topic_name, comment):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        url = Path.SUBSCRIPTIONS % (project_name, topic_name)
        request_param = CreateSubscriptionParams(comment)

        content = self._rest_client.post(url, data=request_param.content())

        result = CreateSubscriptionResult.parse_content(content)
        return result

    def delete_subscription(self, project_name, topic_name, sub_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')

        url = Path.SUBSCRIPTION % (project_name, topic_name, sub_id)

        self._rest_client.delete(url)

    def get_subscription(self, project_name, topic_name, sub_id):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')

        url = Path.SUBSCRIPTION % (project_name, topic_name, sub_id)

        content = self._rest_client.get(url)

        result = GetSubscriptionResult.parse_content(content)
        return result

    def update_subscription(self, project_name, topic_name, sub_id, comment):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')

        url = Path.SUBSCRIPTION % (project_name, topic_name, sub_id)
        request_param = UpdateSubscriptionParams(comment)

        self._rest_client.put(url, data=request_param.content())

    def update_subscription_state(self, project_name, topic_name, sub_id, state):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')
        if not check_type(state, SubscriptionState):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE %
                                            ('state', SubscriptionState.__name__))

        url = Path.SUBSCRIPTION % (project_name, topic_name, sub_id)
        request_param = UpdateSubscriptionStateParams(state)

        self._rest_client.put(url, data=request_param.content())

    def list_subscription(self, project_name, topic_name, query_key, page_index, page_size):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if not check_positive(page_index):
            raise InvalidParameterException(ErrorMessage.PARAMETER_NOT_POSITIVE % 'page_index')
        if check_negative(page_size):
            raise InvalidParameterException(ErrorMessage.PARAMETER_NEGATIVE % 'page_size')

        url = Path.SUBSCRIPTIONS % (project_name, topic_name)
        request_param = ListSubscriptionParams(query_key, page_index, page_size)

        content = self._rest_client.post(url, data=request_param.content())
        result = ListSubscriptionResult.parse_content(content)
        return result

    def reset_subscription_offset(self, project_name, topic_name, sub_id, offsets):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(sub_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'sub_id')
        if check_empty(offsets):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'offsets')
        if not check_type(offsets, dict):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE % ('offsets', dict.__name__))

        for (k, v) in offsets.items():
            if isinstance(v, dict):
                offsets[k] = OffsetBase.from_dict(v)

        url = Path.OFFSETS % (project_name, topic_name, sub_id)
        request_param = ResetSubscriptionOffsetParams(offsets)

        self._rest_client.put(url, data=request_param.content())

    # =======================================================
    # private function
    # =======================================================

    def __is_shard_load_completed(self, project_name, topic_name):
        shards = self.list_shard(project_name, topic_name)
        for shard in shards.shards:
            if shard.state not in (ShardState.ACTIVE, ShardState.CLOSED):
                return False
        return True

    def __get_shard(self, project_name, topic_name, shard_id):
        shards = self.list_shard(project_name, topic_name)
        for shard in shards.shards:
            if shard.shard_id == shard_id:
                return shard

    def __create_topic(self, project_name, topic_name, shard_count, life_cycle, record_type, comment,
                       extend_mode=None, record_schema=None):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if not check_topic_name_valid(topic_name):
            raise InvalidParameterException(ErrorMessage.INVALID_TOPIC_NAME)
        if not check_positive(life_cycle):
            raise InvalidParameterException(ErrorMessage.PARAMETER_NOT_POSITIVE % 'life_cycle')
        if not check_type(record_type, RecordType):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE % ('record_type', RecordType.__name__))
        if record_type == RecordType.TUPLE:
            if not check_type(record_schema, RecordSchema):
                raise InvalidParameterException(ErrorMessage.INVALID_RECORD_SCHEMA_TYPE)
        if extend_mode is not None and not isinstance(extend_mode, bool):
            raise InvalidParameterException('type of extend mode should be \'bool\'')

        url = Path.TOPIC % (project_name, topic_name)
        request_param = CreateTopicRequestParams(shard_count, life_cycle, record_type, record_schema, extend_mode, comment)

        self._rest_client.post(url, data=request_param.content())

    def __get_records(self, project_name, topic_name, sub_id, shard_id, cursor, limit_num, record_schema=None):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')
        if check_empty(cursor):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'cursor')
        if check_type(cursor, GetCursorResult):
            raise InvalidParameterException(ErrorMessage.INVALID_TYPE % ('cursor', 'str'))
        if sub_id is not None:
            raise InvalidOperationException("Json protocol not support this method")

        url = Path.SHARD % (project_name, topic_name, shard_id)

        request_param = GetRecordsRequestParams(cursor, limit_num)

        content = self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers(),
                                         compress_format=self._compress_format)

        result = GetRecordsResult.parse_content(content, record_schema=record_schema)

        return result


class DataHubPB(DataHubJson):
    """
    DataHub protobuf client
    """

    def __init__(self, access_id, access_key, endpoint=None, compress_format=None, enable_schema_register=False, **kwargs):
        super().__init__(access_id, access_key, endpoint, compress_format, enable_schema_register, **kwargs)

    def put_records(self, project_name, topic_name, record_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')

        if record_list is None or len(record_list) == 0:
            raise InvalidParameterException("Record list is null or empty")

        url = Path.SHARDS % (project_name, topic_name)

        request_param = PutPBRecordsRequestParams(record_list)

        content = self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers(),
                                         compress_format=self._compress_format)

        result = PutPBRecordsResult.parse_content(content)

        return result

    def put_records_by_shard(self, project_name, topic_name, shard_id, record_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')

        if record_list is None or len(record_list) == 0:
            raise InvalidParameterException("Record list is null or empty")

        url = Path.SHARD % (project_name, topic_name, shard_id)

        request_param = PutPBRecordsRequestParams(record_list)

        self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers(),
                               compress_format=self._compress_format)

    def get_blob_records(self, project_name, topic_name, sub_id, shard_id, cursor, limit_num):
        return self.__get_records(project_name, topic_name, sub_id, shard_id, cursor, limit_num)

    def get_tuple_records(self, project_name, topic_name, sub_id, shard_id, record_schema, cursor, limit_num):
        return self.__get_records(project_name, topic_name, sub_id, shard_id, cursor, limit_num, record_schema)

    def __get_records(self, project_name, topic_name, sub_id, shard_id, cursor, limit_num, record_schema=None):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')
        if check_empty(cursor):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'cursor')

        url = Path.SHARD % (project_name, topic_name, shard_id)
        request_param = GetPBRecordsRequestParams(cursor, limit_num)

        content = self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers(sub_id),
                                         compress_format=self._compress_format)

        result = GetPBRecordsResult.parse_content(content, record_schema=record_schema)

        return result


class DataHubBatch(DataHubJson):
    """
    DataHub batch client
    """

    def __init__(self, access_id, access_key, endpoint=None, compress_format=None, enable_schema_register=True, **kwargs):
        super().__init__(access_id, access_key, endpoint, compress_format, enable_schema_register, **kwargs)
        self._schema_register = SchemaRegistryClient(self) if enable_schema_register else None

    def put_records(self, project_name, topic_name, record_list):
        raise DatahubException("This method is not supported for batch client, please use put_records_by_shard")

    def put_records_by_shard(self, project_name, topic_name, shard_id, record_list):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')

        if record_list is None or len(record_list) == 0:
            raise InvalidParameterException("Record list is null or empty")

        url = Path.SHARD % (project_name, topic_name, shard_id)
        request_param = PutBatchRecordsRequestParams(record_list, project_name, topic_name, self._compress_format,
                                                     self._schema_register)
        self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers())

    def get_blob_records(self, project_name, topic_name, sub_id, shard_id, cursor, limit_num):
        return self.__get_records(project_name, topic_name, sub_id, shard_id, cursor, limit_num)

    def get_tuple_records(self, project_name, topic_name, sub_id, shard_id, record_schema, cursor, limit_num):
        return self.__get_records(project_name, topic_name, sub_id, shard_id, cursor, limit_num, record_schema)

    def __get_records(self, project_name, topic_name, sub_id, shard_id, cursor, limit_num, record_schema=None):
        if check_empty(project_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'project_name')
        if check_empty(topic_name):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'topic_name')
        if check_empty(shard_id):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'shard_id')
        if check_empty(cursor):
            raise InvalidParameterException(ErrorMessage.PARAMETER_EMPTY % 'cursor')

        url = Path.SHARD % (project_name, topic_name, shard_id)
        request_param = GetBatchRecordsRequestParams(cursor, limit_num)

        content = self._rest_client.post(url, data=request_param.content(), headers=request_param.extra_headers(sub_id),
                                         compress_format=self._compress_format)
        result = GetBatchRecordsResult.parse_content(content, record_schema=record_schema, project_name=project_name,
                                                     topic_name=topic_name, init_schema=record_schema,
                                                     schema_register=self._schema_register if record_schema else None)
        return result