class Range(object): """ Represents the Range header. This only represents ``bytes`` ranges, which are the only kind specified in HTTP. This can represent multiple sets of ranges, but no place else is this multi-range facility supported. """ def __init__(self, ranges): # expect non-inclusive for begin, end in ranges: assert end is None or end >= 0, "Bad ranges: %r" % ranges self.ranges = ranges def satisfiable(self, length): """ Returns true if this range can be satisfied by the resource with the given byte length. """ return self.range_for_length(length) is not None def range_for_length(self, length): """ *If* there is only one range, and *if* it is satisfiable by the given length, then return a (begin, end) non-inclusive range of bytes to serve. Otherwise return None """ if length is None or len(self.ranges) != 1: return None start, end = self.ranges[0] if end is None: end = length if start < 0: start += length if _is_content_range_valid(start, end, length): stop = min(end, length) return (start, stop) else: return None def content_range(self, length): """ Works like range_for_length; returns None or a ContentRange object You can use it like:: response.content_range = req.range.content_range(response.content_length) Though it's still up to you to actually serve that content range! """ range = self.range_for_length(length) if range is None: return None return ContentRange(range[0], range[1], length) def __str__(self): parts = [] for begin, end in self.ranges: if end is None: if begin >= 0: parts.append('%s-' % begin) else: parts.append(str(begin)) else: if begin < 0: raise ValueError("(%r, %r) should have a non-negative first value" % (begin, end)) if end <= 0: raise ValueError("(%r, %r) should have a positive second value" % (begin, end)) parts.append('%s-%s' % (begin, end-1)) return 'bytes=%s' % ','.join(parts) def __repr__(self): return '<%s ranges=%s>' % ( self.__class__.__name__, ', '.join(map(repr, self.ranges))) @classmethod def parse(cls, header): """ Parse the header; may return None if header is invalid """ bytes = cls.parse_bytes(header) if bytes is None: return None units, ranges = bytes if units != 'bytes' or ranges is None: return None return cls(ranges) @staticmethod def parse_bytes(header): """ Parse a Range header into (bytes, list_of_ranges). ranges in list_of_ranges are non-inclusive (unlike the HTTP header). Will return None if the header is invalid """ if not header: raise TypeError("The header must not be empty") ranges = [] last_end = 0 try: (units, range) = header.split("=", 1) units = units.strip().lower() for item in range.split(","): if '-' not in item: raise ValueError() if item.startswith('-'): # This is a range asking for a trailing chunk. if last_end < 0: raise ValueError('too many end ranges') begin = int(item) end = None last_end = -1 else: (begin, end) = item.split("-", 1) begin = int(begin) if begin < last_end or last_end < 0: raise ValueError('begin= end: raise ValueError('begin>end') else: end = None last_end = end ranges.append((begin, end)) except ValueError, e: # In this case where the Range header is malformed, # section 14.16 says to treat the request as if the # Range header was not present. How do I log this? return None return (units, ranges) class ContentRange(object): """ Represents the Content-Range header This header is ``start-stop/length``, where start-stop and length can be ``*`` (represented as None in the attributes). """ def __init__(self, start, stop, length): if not _is_content_range_valid(start, stop, length): raise ValueError("Bad start:stop/length: %r-%r/%r" % (start, stop, length)) self.start = start self.stop = stop # this is python-style range end (non-inclusive) self.length = length def __repr__(self): return '<%s %s>' % (self.__class__.__name__, self) def __str__(self): if self.length is None: length = '*' else: length = self.length if self.start is None: assert self.stop is None return 'bytes */%s' % length stop = self.stop - 1 # from non-inclusive to HTTP-style return 'bytes %s-%s/%s' % (self.start, stop, length) def __iter__(self): """ Mostly so you can unpack this, like: start, stop, length = res.content_range """ return iter([self.start, self.stop, self.length]) @classmethod def parse(cls, value): """ Parse the header. May return None if it cannot parse. """ if value is None: return None value = value.strip() if not value.startswith('bytes '): # Unparseable return None value = value[len('bytes '):].strip() if '/' not in value: # Invalid, no length given return None range, length = value.split('/', 1) if length == '*': length = None elif length.isdigit(): length = int(length) else: return None # invalid length if range == '*': return cls(None, None, length) elif '-' not in range: # Invalid, no range return None else: start, stop = range.split('-', 1) try: start = int(start) stop = int(stop) stop += 1 # convert to non-inclusive except ValueError: # Parse problem return None if _is_content_range_valid(start, stop, length, response=True): return cls(start, stop, length) return None def _is_content_range_valid(start, stop, length, response=False): if (start is None) != (stop is None): return False elif start is None: return length is None or length >= 0 elif length is None: return 0 <= start < stop elif start >= stop: return False elif response and stop > length: # "content-range: bytes 0-50/10" is invalid for a response # "range: bytes 0-50" is valid for a request to a 10-bytes entity return False else: return 0 <= start < length