def update_subscription_offset()

in datahub/implement.py [0:0]

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

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