def __create_topic()

in datahub/implement.py [0:0]

• 18 lines of code
• 9 McCabe index (conditional complexity)

    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())