Python bottle.request 模块，POST 实例源码

def _request(request, request_fallback=None):
        ''' Extract request fields wherever they may come from: GET, POST, forms, fallback '''
        # Use lambdas to avoid evaluating bottle.request.* which may throw an Error
        all_dicts = [
            lambda: request.json,
            lambda: request.forms,
            lambda: request.query,
            lambda: request.files,
            #lambda: request.POST,
            lambda: request_fallback
        ]
        request_dict = dict()
        for req_dict_ in all_dicts:
            try:
                req_dict = req_dict_()
            except KeyError:
                continue
            if req_dict is not None and hasattr(req_dict, 'items'):
                for req_key, req_val in req_dict.items():
                    request_dict[req_key] = req_val
        return request_dict

def test_post(self):
        """ Environ: POST data """
        sq = tob('a=a&a=1&b=b&c=&d&cn=%e7%93%b6')
        e = {}
        wsgiref.util.setup_testing_defaults(e)
        e['wsgi.input'].write(sq)
        e['wsgi.input'].seek(0)
        e['CONTENT_LENGTH'] = str(len(sq))
        e['REQUEST_METHOD'] = "POST"
        request = BaseRequest(e)
        self.assertTrue('a' in request.POST)
        self.assertTrue('b' in request.POST)
        self.assertEqual(['a','1'], request.POST.getall('a'))
        self.assertEqual(['b'], request.POST.getall('b'))
        self.assertEqual('1', request.POST['a'])
        self.assertEqual('b', request.POST['b'])
        self.assertEqual('', request.POST['c'])
        self.assertEqual('', request.POST['d'])
        self.assertEqual(tonat(tob('?'), 'latin1'), request.POST['cn'])
        self.assertEqual(touni('?'), request.POST.cn)

def build_preview():
    if request.content_type == 'application/json':
        rst = request.json['rst']
    else:
        rst = request.POST['rst']

    key = md5(rst)
    if key in _CACHE:
        res, warnings = _CACHE[key]
    else:
        res, warnings = _CACHE[key] = rst2html(rst, theme='acr')

    return {'result': res, 'warnings': warnings,
            'valid': len(warnings) == 0}

def call_api(func, args=""):
    response.headers.replace("Content-type", "application/json")
    response.headers.append("Cache-Control", "no-cache, must-revalidate")

    s = request.environ.get('beaker.session')
    if 'session' in request.POST:
        s = s.get_by_id(request.POST['session'])

    if not s or not s.get("authenticated", False):
        return HTTPError(403, json.dumps("Forbidden"))

    if not PYLOAD.isAuthorized(func, {"role": s["role"], "permission": s["perms"]}):
        return HTTPError(401, json.dumps("Unauthorized"))

    args = args.split("/")[1:]
    kwargs = {}

    for x, y in chain(request.GET.iteritems(), request.POST.iteritems()):
        if x == "session": continue
        kwargs[x] = unquote(y)

    try:
        return callApi(func, *args, **kwargs)
    except Exception, e:
        print_exc()
        return HTTPError(500, json.dumps({"error": e.message, "traceback": format_exc()}))

def add(request):
    package = request.POST.get('referer', None)
    urls = filter(lambda x: x != "", request.POST['urls'].split("\n"))

    if package:
        PYLOAD.addPackage(package, urls, 0)
    else:
        PYLOAD.generateAndAddPackages(urls, 0)

    return ""

def set_captcha():
    if request.environ.get('REQUEST_METHOD', "GET") == "POST":
        try:
            PYLOAD.setCaptchaResult(request.forms["cap_id"], request.forms["cap_result"])
        except:
            pass

    task = PYLOAD.getCaptchaTask()

    if task.tid >= 0:
        src = "data:image/%s;base64,%s" % (task.type, task.data)

        return {'captcha': True, 'id': task.tid, 'src': src, 'result_type' : task.resultType}
    else:
        return {'captcha': False}

def save_config(category):
    for key, value in request.POST.iteritems():
        try:
            section, option = key.split("|")
        except:
            continue

        if category == "general": category = "core"

        PYLOAD.setConfigValue(section, option, decode(value), category)

def add_account():
    login = request.POST["account_login"]
    password = request.POST["account_password"]
    type = request.POST["account_type"]

    PYLOAD.updateAccount(type, login, password)

def change_password():

    user = request.POST["user_login"]
    oldpw = request.POST["login_current_password"]
    newpw = request.POST["login_new_password"]

    if not PYLOAD.changePassword(user, oldpw, newpw):
        print "Wrong password"
        return HTTPError()

def test_method(self):
        self.assertEqual(BaseRequest({}).method, 'GET')
        self.assertEqual(BaseRequest({'REQUEST_METHOD':'GET'}).method, 'GET')
        self.assertEqual(BaseRequest({'REQUEST_METHOD':'GeT'}).method, 'GET')
        self.assertEqual(BaseRequest({'REQUEST_METHOD':'get'}).method, 'GET')
        self.assertEqual(BaseRequest({'REQUEST_METHOD':'POst'}).method, 'POST')
        self.assertEqual(BaseRequest({'REQUEST_METHOD':'FanTASY'}).method, 'FANTASY')

def test_bodypost(self):
        sq = tob('foobar')
        e = {}
        wsgiref.util.setup_testing_defaults(e)
        e['wsgi.input'].write(sq)
        e['wsgi.input'].seek(0)
        e['CONTENT_LENGTH'] = str(len(sq))
        e['REQUEST_METHOD'] = "POST"
        request = BaseRequest(e)
        self.assertEqual('', request.POST['foobar'])

def test_body_noclose(self):
        """ Test that the body file handler is not closed after request.POST """
        sq = tob('a=a&a=1&b=b&c=&d')
        e = {}
        wsgiref.util.setup_testing_defaults(e)
        e['wsgi.input'].write(sq)
        e['wsgi.input'].seek(0)
        e['CONTENT_LENGTH'] = str(len(sq))
        e['REQUEST_METHOD'] = "POST"
        request = BaseRequest(e)
        self.assertEqual(sq, request.body.read())
        request.POST # This caused a body.close() with Python 3.x
        self.assertEqual(sq, request.body.read())

def test_params(self):
        """ Environ: GET and POST are combined in request.param """
        e = {}
        wsgiref.util.setup_testing_defaults(e)
        e['wsgi.input'].write(tob('b=b&c=p'))
        e['wsgi.input'].seek(0)
        e['CONTENT_LENGTH'] = '7'
        e['QUERY_STRING'] = 'a=a&c=g'
        e['REQUEST_METHOD'] = "POST"
        request = BaseRequest(e)
        self.assertEqual(['a','b','c'], sorted(request.params.keys()))
        self.assertEqual('p', request.params['c'])

def test_multipart(self):
        """ Environ: POST (multipart files and multible values per key) """
        fields = [('field1','value1'), ('field2','value2'), ('field2','value3')]
        files = [('file1','filename1.txt','content1'), ('??','??foo.py', 'ä\nö\rü')]
        e = tools.multipart_environ(fields=fields, files=files)
        request = BaseRequest(e)
        # File content
        self.assertTrue('file1' in request.POST)
        self.assertTrue('file1' in request.files)
        self.assertTrue('file1' not in request.forms)
        cmp = tob('content1') if sys.version_info >= (3,2,0) else 'content1'
        self.assertEqual(cmp, request.POST['file1'].file.read())
        # File name and meta data
        self.assertTrue('??' in request.POST)
        self.assertTrue('??' in request.files)
        self.assertTrue('??' not in request.forms)
        self.assertEqual('foo.py', request.POST['??'].filename)
        self.assertTrue(request.files['??'])
        self.assertFalse(request.files.file77)
        # UTF-8 files
        x = request.POST['??'].file.read()
        if (3,2,0) > sys.version_info >= (3,0,0):
            x = x.encode('utf8')
        self.assertEqual(tob('ä\nö\rü'), x)
        # No file
        self.assertTrue('file3' not in request.POST)
        self.assertTrue('file3' not in request.files)
        self.assertTrue('file3' not in request.forms)
        # Field (single)
        self.assertEqual('value1', request.POST['field1'])
        self.assertTrue('field1' not in request.files)
        self.assertEqual('value1', request.forms['field1'])
        # Field (multi)
        self.assertEqual(2, len(request.POST.getall('field2')))
        self.assertEqual(['value2', 'value3'], request.POST.getall('field2'))
        self.assertEqual(['value2', 'value3'], request.forms.getall('field2'))
        self.assertTrue('field2' not in request.files)

def contact(db):
    """
    Our contact-us form, basically, present a form if it's a GET request,
    validate and process the form if it's a POST request. Filthy but works.
    """
    form = ContactForm(request.POST, nocaptcha={'ip_address': '127.0.0.1'})
    if request.method == 'POST' and form.validate():
        # process the form, captcha is valid.

        message_text = "Contact Email: {email}\n\n {contact_text}".format(
            email=form.email.data,
            contact_text=form.contact_text.data
        )

        # put together a gmail client for sending messages
        gmail_client = GMail(settings.ADMIN_EMAIL,
                             settings.ADMIN_EMAIL_PASSWORD)
        message = Message('[xqzes] Contact Form',
                          to=settings.ADMIN_EMAIL,
                          text=message_text)
        gmail_client.send(message)

        return redirect("/contact/?contacted=True")

    return template(
        'contact',
        form=form,
        contacted=strtobool(request.GET.get('contacted', 'false'))
    )

def submit(db):

    # TODO: break this out along with others in to an excuses package.
    class SubmissionForm(Form):
        attribution_name = StringField(
            'Your Name (for future attribution purposes)',
            [
                validators.InputRequired(),
                validators.Length(
                    min=3,
                    max=50,
                    message="Srsly, give us a decent username "
                            "(%(min)d - %(max)d chars),"
                            " doesn't even have to be real."
                )
            ]
        )
        excuse = TextAreaField(
            'What\'s YOUR excuse !?!?',
            [
                validators.Length(
                    min=5,
                    max=140,
                    message="Please provide %(min)d - %(max)d "
                            "characters"),
            ]
        )
        nocaptcha = NoCaptchaField(
            public_key=settings.RECAPTCHA_SITE_KEY,
            private_key=settings.RECAPTCHA_SECRET_KEY,
            secure=True,
        )

    form = SubmissionForm(request.POST, nocaptcha={'ip_address': '127.0.0.1'})

    submitted = False
    if request.method == 'POST' and form.validate():
        excuse_record = Excuse(form.attribution_name.data,
                               form.excuse.data)
        db.add(excuse_record)
        submitted = True

    return template('submit', form=form, submitted=submitted)