NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

wrappers.py (4618B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\werkzeug\contrib\wrappers.py
      2 import codecs
      3 from werkzeug.exceptions import BadRequest
      4 from werkzeug.utils import cached_property
      5 from werkzeug.http import dump_options_header, parse_options_header
      6 from werkzeug._internal import _decode_unicode
      7 try:
      8     from simplejson import loads
      9 except ImportError:
     10     from json import loads
     11 
     12 def is_known_charset(charset):
     13     try:
     14         codecs.lookup(charset)
     15     except LookupError:
     16         return False
     17 
     18     return True
     19 
     20 
     21 class JSONRequestMixin(object):
     22 
     23     @cached_property
     24     def json(self):
     25         if 'json' not in self.environ.get('CONTENT_TYPE', ''):
     26             raise BadRequest('Not a JSON request')
     27         try:
     28             return loads(self.data)
     29         except Exception:
     30             raise BadRequest('Unable to read JSON request')
     31 
     32 
     33 class ProtobufRequestMixin(object):
     34     protobuf_check_initialization = True
     35 
     36     def parse_protobuf(self, proto_type):
     37         if 'protobuf' not in self.environ.get('CONTENT_TYPE', ''):
     38             raise BadRequest('Not a Protobuf request')
     39         obj = proto_type()
     40         try:
     41             obj.ParseFromString(self.data)
     42         except Exception:
     43             raise BadRequest('Unable to parse Protobuf request')
     44 
     45         if self.protobuf_check_initialization and not obj.IsInitialized():
     46             raise BadRequest('Partial Protobuf request')
     47         return obj
     48 
     49 
     50 class RoutingArgsRequestMixin(object):
     51 
     52     def _get_routing_args(self):
     53         return self.environ.get('wsgiorg.routing_args', ())[0]
     54 
     55     def _set_routing_args(self, value):
     56         if self.shallow:
     57             raise RuntimeError('A shallow request tried to modify the WSGI environment.  If you really want to do that, set `shallow` to False.')
     58         self.environ['wsgiorg.routing_args'] = (value, self.routing_vars)
     59 
     60     routing_args = property(_get_routing_args, _set_routing_args, doc='\n        The positional URL arguments as `tuple`.')
     61     del _get_routing_args
     62     del _set_routing_args
     63 
     64     def _get_routing_vars(self):
     65         rv = self.environ.get('wsgiorg.routing_args')
     66         if rv is not None:
     67             return rv[1]
     68         rv = {}
     69         if not self.shallow:
     70             self.routing_vars = rv
     71         return rv
     72 
     73     def _set_routing_vars(self, value):
     74         if self.shallow:
     75             raise RuntimeError('A shallow request tried to modify the WSGI environment.  If you really want to do that, set `shallow` to False.')
     76         self.environ['wsgiorg.routing_args'] = (self.routing_args, value)
     77 
     78     routing_vars = property(_get_routing_vars, _set_routing_vars, doc='\n        The keyword URL arguments as `dict`.')
     79     del _get_routing_vars
     80     del _set_routing_vars
     81 
     82 
     83 class ReverseSlashBehaviorRequestMixin(object):
     84 
     85     @cached_property
     86     def path(self):
     87         path = (self.environ.get('PATH_INFO') or '').lstrip('/')
     88         return _decode_unicode(path, self.charset, self.encoding_errors)
     89 
     90     @cached_property
     91     def script_root(self):
     92         path = (self.environ.get('SCRIPT_NAME') or '').rstrip('/') + '/'
     93         return _decode_unicode(path, self.charset, self.encoding_errors)
     94 
     95 
     96 class DynamicCharsetRequestMixin(object):
     97     default_charset = 'latin1'
     98 
     99     def unknown_charset(self, charset):
    100         return 'latin1'
    101 
    102     @cached_property
    103     def charset(self):
    104         header = self.environ.get('CONTENT_TYPE')
    105         if header:
    106             ct, options = parse_options_header(header)
    107             charset = options.get('charset')
    108             if charset:
    109                 if is_known_charset(charset):
    110                     return charset
    111                 return self.unknown_charset(charset)
    112         return self.default_charset
    113 
    114 
    115 class DynamicCharsetResponseMixin(object):
    116     default_charset = 'utf-8'
    117 
    118     def _get_charset(self):
    119         header = self.headers.get('content-type')
    120         if header:
    121             charset = parse_options_header(header)[1].get('charset')
    122             if charset:
    123                 return charset
    124         return self.default_charset
    125 
    126     def _set_charset(self, charset):
    127         header = self.headers.get('content-type')
    128         ct, options = parse_options_header(header)
    129         if not ct:
    130             raise TypeError('Cannot set charset if Content-Type header is missing.')
    131         options['charset'] = charset
    132         self.headers['Content-Type'] = dump_options_header(ct, options)
    133 
    134     charset = property(_get_charset, _set_charset, doc="\n        The charset for the response.  It's stored inside the\n        Content-Type header as a parameter.")
    135     del _get_charset
    136     del _set_charset