Source code for django.http.response

importdatetimeimportjsonimportmimetypesimportosimportreimportsysimporttimefromemail.headerimportHeaderfromhttp.clientimportresponsesfromurllib.parseimportquote,urlparsefromdjango.confimportsettingsfromdjango.coreimportsignals,signingfromdjango.core.exceptionsimportDisallowedRedirectfromdjango.core.serializers.jsonimportDjangoJSONEncoderfromdjango.http.cookieimportSimpleCookiefromdjango.utilsimporttimezonefromdjango.utils.encodingimportiri_to_urifromdjango.utils.httpimporthttp_date_charset_from_content_type_re=re.compile(r';\s*charset=(?P<charset>[^\s;]+)',re.I)classBadHeaderError(ValueError):passclassHttpResponseBase:""" An HTTP response base class with dictionary-accessed headers. This class doesn't handle content. It should not be used directly. Use the HttpResponse and StreamingHttpResponse subclasses instead. """status_code=200def__init__(self,content_type=None,status=None,reason=None,charset=None):# _headers is a mapping of the lowercase name to the original case of# the header (required for working with legacy systems) and the header# value. Both the name of the header and its value are ASCII strings.self._headers={}self._closable_objects=[]# This parameter is set by the handler. It's necessary to preserve the# historical behavior of request_finished.self._handler_class=Noneself.cookies=SimpleCookie()self.closed=FalseifstatusisnotNone:try:self.status_code=int(status)except(ValueError,TypeError):raiseTypeError('HTTP status code must be an integer.')ifnot100<=self.status_code<=599:raiseValueError('HTTP status code must be an integer from 100 to 599.')self._reason_phrase=reasonself._charset=charsetifcontent_typeisNone:content_type='%s; charset=%s'%(settings.DEFAULT_CONTENT_TYPE,self.charset)self['Content-Type']=content_type@propertydefreason_phrase(self):ifself._reason_phraseisnotNone:returnself._reason_phrase# Leave self._reason_phrase unset in order to use the default# reason phrase for status code.returnresponses.get(self.status_code,'Unknown Status Code')@reason_phrase.setterdefreason_phrase(self,value):self._reason_phrase=value@propertydefcharset(self):ifself._charsetisnotNone:returnself._charsetcontent_type=self.get('Content-Type','')matched=_charset_from_content_type_re.search(content_type)ifmatched:# Extract the charset and strip its double quotesreturnmatched.group('charset').replace('"','')returnsettings.DEFAULT_CHARSET@charset.setterdefcharset(self,value):self._charset=valuedefserialize_headers(self):"""HTTP headers as a bytestring."""defto_bytes(val,encoding):returnvalifisinstance(val,bytes)elseval.encode(encoding)headers=[(to_bytes(key,'ascii')+b': '+to_bytes(value,'latin-1'))forkey,valueinself._headers.values()]returnb'\r\n'.join(headers)__bytes__=serialize_headers@propertydef_content_type_for_repr(self):return', "%s"'%self['Content-Type']if'Content-Type'inselfelse''def_convert_to_charset(self,value,charset,mime_encode=False):""" Convert headers key/value to ascii/latin-1 native strings. `charset` must be 'ascii' or 'latin-1'. If `mime_encode` is True and `value` can't be represented in the given charset, apply MIME-encoding. """ifnotisinstance(value,(bytes,str)):value=str(value)if((isinstance(value,bytes)and(b'\n'invalueorb'\r'invalue))orisinstance(value,str)and('\n'invalueor'\r'invalue)):raiseBadHeaderError("Header values can't contain newlines (got %r)"%value)try:ifisinstance(value,str):# Ensure string is valid in given charsetvalue.encode(charset)else:# Convert bytestring using given charsetvalue=value.decode(charset)exceptUnicodeErrorase:ifmime_encode:value=Header(value,'utf-8',maxlinelen=sys.maxsize).encode()else:e.reason+=', HTTP response headers must be in %s format'%charsetraisereturnvaluedef__setitem__(self,header,value):header=self._convert_to_charset(header,'ascii')value=self._convert_to_charset(value,'latin-1',mime_encode=True)self._headers[header.lower()]=(header,value)def__delitem__(self,header):self._headers.pop(header.lower(),False)def__getitem__(self,header):returnself._headers[header.lower()][1]defhas_header(self,header):"""Case-insensitive check for a header."""returnheader.lower()inself._headers__contains__=has_headerdefitems(self):returnself._headers.values()defget(self,header,alternate=None):returnself._headers.get(header.lower(),(None,alternate))[1]defset_cookie(self,key,value='',max_age=None,expires=None,path='/',domain=None,secure=False,httponly=False,samesite=None):""" Set a cookie. ``expires`` can be: - a string in the correct format, - a naive ``datetime.datetime`` object in UTC, - an aware ``datetime.datetime`` object in any time zone. If it is a ``datetime.datetime`` object then calculate ``max_age``. """self.cookies[key]=valueifexpiresisnotNone:ifisinstance(expires,datetime.datetime):iftimezone.is_aware(expires):expires=timezone.make_naive(expires,timezone.utc)delta=expires-expires.utcnow()# Add one second so the date matches exactly (a fraction of# time gets lost between converting to a timedelta and# then the date string).delta=delta+datetime.timedelta(seconds=1)# Just set max_age - the max_age logic will set expires.expires=Nonemax_age=max(0,delta.days*86400+delta.seconds)else:self.cookies[key]['expires']=expireselse:self.cookies[key]['expires']=''ifmax_ageisnotNone:self.cookies[key]['max-age']=max_age# IE requires expires, so set it if hasn't been already.ifnotexpires:self.cookies[key]['expires']=http_date(time.time()+max_age)ifpathisnotNone:self.cookies[key]['path']=pathifdomainisnotNone:self.cookies[key]['domain']=domainifsecure:self.cookies[key]['secure']=Trueifhttponly:self.cookies[key]['httponly']=Trueifsamesite:ifsamesite.lower()notin('lax','strict'):raiseValueError('samesite must be "lax" or "strict".')self.cookies[key]['samesite']=samesitedefsetdefault(self,key,value):"""Set a header unless it has already been set."""ifkeynotinself:self[key]=valuedefset_signed_cookie(self,key,value,salt='',**kwargs):value=signing.get_cookie_signer(salt=key+salt).sign(value)returnself.set_cookie(key,value,**kwargs)defdelete_cookie(self,key,path='/',domain=None):# Most browsers ignore the Set-Cookie header if the cookie name starts# with __Host- or __Secure- and the cookie doesn't use the secure flag.secure=key.startswith(('__Secure-','__Host-'))self.set_cookie(key,max_age=0,path=path,domain=domain,secure=secure,expires='Thu, 01 Jan 1970 00:00:00 GMT',)# Common methods used by subclassesdefmake_bytes(self,value):"""Turn a value into a bytestring encoded in the output charset."""# Per PEP 3333, this response body must be bytes. To avoid returning# an instance of a subclass, this function returns `bytes(value)`.# This doesn't make a copy when `value` already contains bytes.# Handle string types -- we can't rely on force_bytes here because:# - Python attempts str conversion first# - when self._charset != 'utf-8' it re-encodes the contentifisinstance(value,bytes):returnbytes(value)ifisinstance(value,str):returnbytes(value.encode(self.charset))# Handle non-string types.returnstr(value).encode(self.charset)# These methods partially implement the file-like object interface.# See https://docs.python.org/library/io.html#io.IOBase# The WSGI server must call this method upon completion of the request.# See http://blog.dscpl.com.au/2012/10/obligations-for-calling-close-on.htmldefclose(self):forclosableinself._closable_objects:try:closable.close()exceptException:passself.closed=Truesignals.request_finished.send(sender=self._handler_class)defwrite(self,content):raiseIOError("This %s instance is not writable"%self.__class__.__name__)defflush(self):passdeftell(self):raiseIOError("This %s instance cannot tell its position"%self.__class__.__name__)# These methods partially implement a stream-like object interface.# See https://docs.python.org/library/io.html#io.IOBasedefreadable(self):returnFalsedefseekable(self):returnFalsedefwritable(self):returnFalsedefwritelines(self,lines):raiseIOError("This %s instance is not writable"%self.__class__.__name__)

[docs]classHttpResponse(HttpResponseBase):""" An HTTP response class with a string as content. This content that can be read, appended to, or replaced. """streaming=False

[docs]def__init__(self,content=b'',*args,**kwargs):super().__init__(*args,**kwargs)# Content is a bytestring. See the `content` property methods.self.content=content

def__repr__(self):return'<%(cls)s status_code=%(status_code)d%(content_type)s>'%{'cls':self.__class__.__name__,'status_code':self.status_code,'content_type':self._content_type_for_repr,}defserialize(self):"""Full HTTP message, including headers, as a bytestring."""returnself.serialize_headers()+b'\r\n\r\n'+self.content__bytes__=serialize@propertydefcontent(self):returnb''.join(self._container)@content.setterdefcontent(self,value):# Consume iterators upon assignment to allow repeated iteration.ifhasattr(value,'__iter__')andnotisinstance(value,(bytes,str)):content=b''.join(self.make_bytes(chunk)forchunkinvalue)ifhasattr(value,'close'):try:value.close()exceptException:passelse:content=self.make_bytes(value)# Create a list of properly encoded bytestrings to support write().self._container=[content]def__iter__(self):returniter(self._container)

[docs]classStreamingHttpResponse(HttpResponseBase):""" A streaming HTTP response class with an iterator as content. This should only be iterated once, when the response is streamed to the client. However, it can be appended to or replaced with a new iterator that wraps the original content (or yields entirely new content). """streaming=Truedef__init__(self,streaming_content=(),*args,**kwargs):super().__init__(*args,**kwargs)# `streaming_content` should be an iterable of bytestrings.# See the `streaming_content` property methods.self.streaming_content=streaming_content@propertydefcontent(self):raiseAttributeError("This %s instance has no `content` attribute. Use ""`streaming_content` instead."%self.__class__.__name__)@propertydefstreaming_content(self):returnmap(self.make_bytes,self._iterator)@streaming_content.setterdefstreaming_content(self,value):self._set_streaming_content(value)def_set_streaming_content(self,value):# Ensure we can never iterate on "value" more than once.self._iterator=iter(value)ifhasattr(value,'close'):self._closable_objects.append(value)def__iter__(self):returnself.streaming_contentdefgetvalue(self):returnb''.join(self.streaming_content)

[docs]defset_headers(self,filelike):""" Set some common response headers (Content-Length, Content-Type, and Content-Disposition) based on the `filelike` response content. """encoding_map={'bzip2':'application/x-bzip','gzip':'application/gzip','xz':'application/x-xz',}filename=getattr(filelike,'name',None)filename=filenameif(isinstance(filename,str)andfilename)elseself.filenameifos.path.isabs(filename):self['Content-Length']=os.path.getsize(filelike.name)elifhasattr(filelike,'getbuffer'):self['Content-Length']=filelike.getbuffer().nbytesifself.get('Content-Type','').startswith(settings.DEFAULT_CONTENT_TYPE):iffilename:content_type,encoding=mimetypes.guess_type(filename)# Encoding isn't set to prevent browsers from automatically# uncompressing files.content_type=encoding_map.get(encoding,content_type)self['Content-Type']=content_typeor'application/octet-stream'else:self['Content-Type']='application/octet-stream'ifself.as_attachment:filename=self.filenameoros.path.basename(filename)iffilename:try:filename.encode('ascii')file_expr='filename="{}"'.format(filename)exceptUnicodeEncodeError:file_expr="filename*=utf-8''{}".format(quote(filename))self['Content-Disposition']='attachment; {}'.format(file_expr)

[docs]classJsonResponse(HttpResponse):""" An HTTP response class that consumes data to be serialized to JSON. :param data: Data to be dumped into json. By default only ``dict`` objects are allowed to be passed due to a security flaw before EcmaScript 5. See the ``safe`` parameter for more information. :param encoder: Should be a json encoder class. Defaults to ``django.core.serializers.json.DjangoJSONEncoder``. :param safe: Controls if only ``dict`` objects may be serialized. Defaults to ``True``. :param json_dumps_params: A dictionary of kwargs passed to json.dumps(). """def__init__(self,data,encoder=DjangoJSONEncoder,safe=True,json_dumps_params=None,**kwargs):ifsafeandnotisinstance(data,dict):raiseTypeError('In order to allow non-dict objects to be serialized set the ''safe parameter to False.')ifjson_dumps_paramsisNone:json_dumps_params={}kwargs.setdefault('content_type','application/json')data=json.dumps(data,cls=encoder,**json_dumps_params)super().__init__(content=data,**kwargs)