python类unquote_plus()的实例源码

def process_key_event(event, context):
    processor = EncryptExtantKeys(config)
    for record in event.get('Records', []):
        bucket = record['s3']['bucket']['name']
        key = {'Key': unquote_plus(record['s3']['object']['key']),
               'Size': record['s3']['object']['size']}
        version = record['s3']['object'].get('versionId')
        if version is not None:
            key['VersionId'] = version
            # lambda event is always latest version, but IsLatest
            # is not in record
            key['IsLatest'] = True
            method = processor.process_version
        else:
            method = processor.process_key
        try:
            result = retry(method, s3, key, bucket)
        except ClientError as e:
            # Ensure we know which key caused an issue
            print("error %s:%s code:%s" % (
                bucket, key['Key'], e.response['Error']))
            raise
        if not result:
            return
        print("remediated %s:%s" % (bucket, key['Key']))

def __init__(self, raw, errors='replace'):
        """Takes a string of type application/x-www-form-urlencoded.
        """
        # urllib needs bytestrings in py2 and unicode strings in py3
        raw_str = raw.encode('ascii') if PY2 else raw

        self.decoded = _decode(unquote_plus(raw_str), errors=errors)
        self.raw = raw

        common_kw = dict(keep_blank_values=True, strict_parsing=False)
        if PY2:
            # in python 2 parse_qs does its own unquote_plus'ing ...
            as_dict = parse_qs(raw_str, **common_kw)
            # ... but doesn't decode to unicode.
            for k, vals in list(as_dict.items()):
                as_dict[_decode(k, errors=errors)] = [
                    _decode(v, errors=errors) for v in vals
                ]
        else:
            # in python 3 parse_qs does the decoding
            as_dict = parse_qs(raw_str, errors=errors, **common_kw)

        Mapping.__init__(self, as_dict)

def get(self, name):
        self.response.out.write(unquote_plus(name))

def __init__(self, url):
        parts = parse.urlparse(url)
        _query = frozenset(parse.parse_qsl(parts.query))
        _path = parse.unquote_plus(parts.path)
        parts = parts._replace(query=_query, path=_path)
        self.parts = parts

def __init__(self, url):
        parts = parse.urlparse(url)
        _query = frozenset(parse.parse_qsl(parts.query))
        _path = parse.unquote_plus(parts.path)
        parts = parts._replace(query=_query, path=_path)
        self.parts = parts

def post_query_extract(mime, length, stream):
    """
    Extract a url-encoded form POST/PUT from stream
    content length, return None
    Attempt to decode application/x-www-form-urlencoded or multipart/*,
    otherwise read whole block and b64encode
    """
    post_query = b''

    try:
        length = int(length)
    except (ValueError, TypeError):
        return

    if length <= 0:
        return

    while length > 0:
        buff = stream.read(length)
        length -= len(buff)

        if not buff:
            break

        post_query += buff

    if not mime:
        mime = ''

    if mime.startswith('application/x-www-form-urlencoded'):
        post_query = to_native_str(post_query)
        post_query = unquote_plus(post_query)

    elif mime.startswith('multipart/'):
        env = {'REQUEST_METHOD': 'POST',
               'CONTENT_TYPE': mime,
               'CONTENT_LENGTH': len(post_query)}

        args = dict(fp=BytesIO(post_query),
                    environ=env,
                    keep_blank_values=True)

        if six.PY3:
            args['encoding'] = 'utf-8'

        data = cgi.FieldStorage(**args)

        values = []
        for item in data.list:
            values.append((item.name, item.value))

        post_query = urlencode(values, True)

    else:
        post_query = base64.b64encode(post_query)
        post_query = to_native_str(post_query)
        post_query = '__warc_post_data=' + post_query

    return post_query