def get_cursor()

in datahub/implement.py [0:0]

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

    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