def init_and_get_subscription_offset()

in datahub/implement.py [0:0]

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

    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