Python parse.parse_qsl方法代码示例

本文整理汇总了Python中urllib.parse.parse_qsl方法的典型用法代码示例。如果您正苦于以下问题:Python parse.parse_qsl方法的具体用法?Python parse.parse_qsl怎么用?Python parse.parse_qsl使用的例子?那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块urllib.parse的用法示例。

在下文中一共展示了parse.parse_qsl方法的29个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: _authenticate

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def _authenticate(self, client, interactive=True):
        if not client.is_registered():
            self._register_client(client)
        flask.session['destination'] = flask.request.url
        flask.session['state'] = rndstr()
        flask.session['nonce'] = rndstr()
        # Use silent authentication for session refresh
        # This will not show login prompt to the user
        extra_auth_params = {}
        if not interactive:
            extra_auth_params['prompt'] = 'none'
        login_url = client.authentication_request(flask.session['state'],
                                                  flask.session['nonce'],
                                                  extra_auth_params)
        auth_params = dict(parse_qsl(login_url.split('?')[1]))
        flask.session['fragment_encoded_response'] = AuthResponseHandler.expect_fragment_encoded_response(auth_params)
        return redirect(login_url) 
开发者ID:zamzterz,项目名称:Flask-pyoidc,代码行数:23,代码来源:flask_pyoidc.py


示例2: read

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def read(self):
        try:
            from PIL import Image
            from pyzbar.pyzbar import decode
            decoded_data = decode(Image.open(self.filename))
            if path.isfile(self.filename):
                remove(self.filename)
            try:
                url = urlparse(decoded_data[0].data.decode())
                query_params = parse_qsl(url.query)
                self._codes = dict(query_params)
                return self._codes.get("secret")
            except (KeyError, IndexError):
                Logger.error("Invalid QR image")
                return None
        except ImportError:
            from ..application import Application
            Application.USE_QRSCANNER = False
            QRReader.ZBAR_FOUND = False 
开发者ID:bilelmoussaoui,项目名称:Authenticator,代码行数:21,代码来源:qr_reader.py


示例3: GET

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def GET(self):
        """ list all rucio accounts.
        HTTP Success:
            200 OK
        HTTP Error:
            401 Unauthorized
            406 Not Acceptable
            500 InternalError
        :param Rucio-Account: Account identifier.
        :param Rucio-Auth-Token: as an 32 character hex string.
        :returns: A list containing all account names as dict.
        """
        header('Content-Type', 'application/x-json-stream')
        filter = {}
        if ctx.query:
            filter = dict(parse_qsl(ctx.query[1:]))
        for account in list_accounts(filter=filter):
            yield render_json(**account) + "\n" 
开发者ID:rucio,项目名称:rucio,代码行数:24,代码来源:account.py


示例4: urlparams

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def urlparams(url_, hash=None, **query):
    """Add a fragment and/or query paramaters to a URL.
    New query params will be appended to exising parameters, except duplicate
    names, which will be replaced.
    """
    url = urlparse.urlparse(url_)
    fragment = hash if hash is not None else url.fragment
    # Use dict(parse_qsl) so we don't get lists of values.
    query_dict = dict(urlparse.parse_qsl(url.query))
    query_dict.update(query)
    query_string = urlencode(
        [(k, v) for k, v in query_dict.items() if v is not None])
    new = urlparse.ParseResult(url.scheme, url.netloc, url.path, url.params,
                               query_string, fragment)
    return new.geturl() 
开发者ID:mozilla,项目名称:sugardough,代码行数:20,代码来源:helpers.py


示例5: transform_url

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def transform_url(url, qparams=None, **kwargs):
    """ Modify url
    :param url: url to transform (can be relative)
    :param qparams: additional query params to add to end of url
    :param kwargs: pieces of URL to modify - e.g. netloc=localhost:8000
    :return: Modified URL
    .. versionadded:: 3.2.0
    """
    if not url:
        return url
    link_parse = urlsplit(url)
    if qparams:
        current_query = dict(parse_qsl(link_parse.query))
        current_query.update(qparams)
        link_parse = link_parse._replace(query=urlencode(current_query))
    return urlunsplit(link_parse._replace(**kwargs)) 
开发者ID:Flask-Middleware,项目名称:flask-security,代码行数:20,代码来源:utils.py


示例6: test_spa_get

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def test_spa_get(app, client):
    """
    Test 'single-page-application' style redirects
    This uses json only.
    """
    with capture_flashes() as flashes:
        with capture_passwordless_login_requests() as requests:
            response = client.post(
                "/login",
                json=dict(email="matt@lp.com"),
                headers={"Content-Type": "application/json"},
            )
            assert response.headers["Content-Type"] == "application/json"
        token = requests[0]["login_token"]
        response = client.get("/login/" + token)
        assert response.status_code == 302
        split = urlsplit(response.headers["Location"])
        assert "localhost:8081" == split.netloc
        assert "/login-redirect" == split.path
        qparams = dict(parse_qsl(split.query))
        assert qparams["email"] == "matt@lp.com"
    assert len(flashes) == 0 
开发者ID:Flask-Middleware,项目名称:flask-security,代码行数:25,代码来源:test_passwordless.py


示例7: test_spa_get

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def test_spa_get(app, client):
    """
    Test 'single-page-application' style redirects
    This uses json only.
    """
    with capture_reset_password_requests() as requests:
        response = client.post(
            "/reset",
            json=dict(email="joe@lp.com"),
            headers={"Content-Type": "application/json"},
        )
        assert response.headers["Content-Type"] == "application/json"
        assert "user" not in response.json["response"]
    token = requests[0]["token"]
    response = client.get("/reset/" + token)
    assert response.status_code == 302
    split = urlsplit(response.headers["Location"])
    assert "localhost:8081" == split.netloc
    assert "/reset-redirect" == split.path
    qparams = dict(parse_qsl(split.query))
    assert qparams["email"] == "joe@lp.com"
    assert qparams["token"] == token 
开发者ID:Flask-Middleware,项目名称:flask-security,代码行数:25,代码来源:test_recoverable.py


示例8: test_tf_link_spa

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def test_tf_link_spa(app, client, get_message):
    # Verify two-factor required when using magic link and SPA
    # This currently isn't supported and should redirect to an error.
    with app.mail.record_messages() as outbox:
        response = client.post(
            "/us-signin/send-code",
            data=dict(identity="matt@lp.com", chosen_method="email"),
            follow_redirects=True,
        )
        assert response.status_code == 200
        assert b"Sign In" in response.data
    matcher = re.match(
        r".*(http://[^\s*]*).*", outbox[0].body, re.IGNORECASE | re.DOTALL
    )
    magic_link = matcher.group(1)
    response = client.get(magic_link, follow_redirects=False)
    split = urlsplit(response.location)
    assert "localhost:8081" == split.netloc
    assert "/login-error" == split.path
    qparams = dict(parse_qsl(split.query))
    assert qparams["tf_required"] == "1"
    assert qparams["email"] == "matt@lp.com" 
开发者ID:Flask-Middleware,项目名称:flask-security,代码行数:25,代码来源:test_unified_signin.py


示例9: test_spa_get

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def test_spa_get(app, client):
    """
    Test 'single-page-application' style redirects
    This uses json only.
    """
    with capture_flashes() as flashes:
        with capture_registrations() as registrations:
            response = client.post(
                "/register",
                json=dict(email="dude@lp.com", password="awesome sunset"),
                headers={"Content-Type": "application/json"},
            )
            assert response.headers["Content-Type"] == "application/json"
        token = registrations[0]["confirm_token"]
        response = client.get("/confirm/" + token)
        assert response.status_code == 302
        split = urlsplit(response.headers["Location"])
        assert "localhost:8081" == split.netloc
        assert "/confirm-redirect" == split.path
        qparams = dict(parse_qsl(split.query))
        assert qparams["email"] == "dude@lp.com"
    # Arguably for json we shouldn't have any - this is buried in register_user
    # but really shouldn't be.
    assert len(flashes) == 1 
开发者ID:Flask-Middleware,项目名称:flask-security,代码行数:27,代码来源:test_confirmable.py


示例10: _strip_signing_parameters

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def _strip_signing_parameters(self, url):
        """ Duplicated Unsiged URLs from Django-Stroage
        Method from: https://github.com/jschneier/django-storages/blob/master/storages/backends/s3boto3.py
        Boto3 does not currently support generating URLs that are unsigned. Instead we
        take the signed URLs and strip any querystring params related to signing and expiration.
        Note that this may end up with URLs that are still invalid, especially if params are
        passed in that only work with signed URLs, e.g. response header params.
        The code attempts to strip all query parameters that match names of known parameters
        from v2 and v4 signatures, regardless of the actual signature version used.
        """
        split_url = urlsplit(url)
        qs = parse_qsl(split_url.query, keep_blank_values=True)
        blacklist = {
            'x-amz-algorithm', 'x-amz-credential', 'x-amz-date',
            'x-amz-expires', 'x-amz-signedheaders', 'x-amz-signature',
            'x-amz-security-token', 'awsaccesskeyid', 'expires', 'signature',
        }
        filtered_qs = ((key, val) for key, val in qs if key.lower() not in blacklist)
        # Note: Parameters that did not have a value in the original query string will have
        # an '=' sign appended to it, e.g ?foo&bar becomes ?foo=&bar=
        joined_qs = ('='.join(keyval) for keyval in filtered_qs)
        split_url = split_url._replace(query="&".join(joined_qs))
        return split_url.geturl() 
开发者ID:OasisLMF,项目名称:OasisPlatform,代码行数:27,代码来源:storage_manager.py


示例11: parse_vcs_requirement

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def parse_vcs_requirement(req):
    """Parses VCS line to egg name, version etc.
    """
    if '+' not in req:
        return None
    vcs, url = req.split('+', 1)
    if vcs not in ('git', 'svn', 'hg'):
        return None
    parsed_url = urlparse(url)
    parsed = dict(parse_qsl(parsed_url.fragment))
    if 'egg' not in parsed:
        return None
    egg = parsed['egg'].rsplit('-', 1)
    if len(egg) > 1:
        try:
            pkg_resources_parse_version(egg[1])
        except pkg_resources._vendor.packaging.version.InvalidVersion:
            return parsed['egg'].lower(), req, None
        return egg[0].lower(), req, egg[1]
    else:
        return parsed['egg'].lower(), req, None 
开发者ID:Deepwalker,项目名称:pundler,代码行数:23,代码来源:pundle.py


示例12: get_video_id_from_url

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def get_video_id_from_url(self, url, info):
        """
        Get YouTube video ID from URL
        """
        try:
            path = info.path
            domain = info.netloc
            video_id = ""
            if domain == "youtu.be":
                video_id = path.split("/")[1]
            else:
                parsed = parse_qsl(info.query)
                params = dict(parsed)
                if "v" in params:
                    video_id = params["v"]
            if video_id:
                return video_id
            else:
                log.error("SpiffyTitles: error getting video id from %s" % (url))
        except IndexError as e:
            log.error(
                "SpiffyTitles: error getting video id from %s (%s)" % (url, str(e))
            ) 
开发者ID:oddluck,项目名称:limnoria-plugins,代码行数:25,代码来源:plugin.py


示例13: _parse_from_uri

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def _parse_from_uri(self, uri: str):
        uri_parts = urlparse(uri)
        conn_type = uri_parts.scheme
        if conn_type == 'postgresql':
            conn_type = 'postgres'
        elif '-' in conn_type:
            conn_type = conn_type.replace('-', '_')
        self.conn_type = conn_type
        self.host = _parse_netloc_to_hostname(uri_parts)
        quoted_schema = uri_parts.path[1:]
        self.schema = unquote(quoted_schema) if quoted_schema else quoted_schema
        self.login = unquote(uri_parts.username) \
            if uri_parts.username else uri_parts.username
        self.password = unquote(uri_parts.password) \
            if uri_parts.password else uri_parts.password
        self.port = uri_parts.port
        if uri_parts.query:
            self.extra = json.dumps(dict(parse_qsl(uri_parts.query, keep_blank_values=True))) 
开发者ID:apache,项目名称:airflow,代码行数:20,代码来源:connection.py


示例14: add_params

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def add_params(url, extra=None, remove=None):
    """Given a URL, add new query parameters by merging in the contents of the
    `extra` dictionary.
    :param url: (str)
    :param extra: (dict)
    :param remove: (list or set)
    :returns: (str) URL including new parameters
    """
    if not (extra or remove):
        return url
    parsed = parse.urlparse(url)._asdict()
    params = parse.parse_qsl(parsed["query"])
    if extra:
        params += list(extra.items())
    if remove:
        params = [pair for pair in params if pair[0] not in remove]
    parsed["query"] = parse.urlencode(params, doseq=True)
    return parse.urlunparse(parse.ParseResult(**parsed)) 
开发者ID:codeforboston,项目名称:cornerwise,代码行数:26,代码来源:utils.py


示例15: attack

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def attack(self, payload, url):
        try:
            # Current request parameters
            params = dict(parse_qsl(urlsplit(url).query))
            # Change the value of the parameters with the payload
            tainted_params = {x: payload for x in params}
            if len(tainted_params) > 0:
                # Prepare the attack URL
                attack_url = urlsplit(url).geturl() + urlencode(tainted_params)
                self.output.debug("Testing: %s" % attack_url)
                resp = self.request.send(
                    url=attack_url, method="GET", payload=None, headers=None
                )
                if self.errors(resp.text):
                    self.output.finding(
                        "That site is may be vulnerable to LDAP Injection at %s\nInjection: %s"
                        % (url, payload)
                    )
        except Exception as e:
            self.logger.error(e)
            self.output.error("Error occured\nAborting this attack...\n")
            self.output.debug("Traceback: %s" % e)
            return 
开发者ID:shenril,项目名称:Sitadel,代码行数:26,代码来源:ldap.py


示例16: attack

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def attack(self, payload, url):
        flag = r"root:/root:/bin/bash|default=multi([0])disk([0])rdisk([0])partition([1])\\WINDOWS"
        try:
            # Current request parameters
            params = dict(parse_qsl(urlsplit(url).query))
            # Change the value of the parameters with the payload
            tainted_params = {x: payload for x in params}
            if len(tainted_params) > 0:
                # Prepare the attack URL
                attack_url = urlsplit(url).geturl() + urlencode(tainted_params)
                self.output.debug("Testing: %s" % attack_url)
                resp = self.request.send(
                    url=attack_url, method="GET", payload=None, headers=None
                )
                if re.search(flag, resp.text):
                    self.output.finding(
                        "That site is may be vulnerable to Remote File Inclusion (RFI) at %s\nInjection: %s"
                        % (url, payload)
                    )
        except Exception as e:
            self.logger.error(e)
            self.output.error("Error occured\nAborting this attack...\n")
            self.output.debug("Traceback: %s" % e)
            return 
开发者ID:shenril,项目名称:Sitadel,代码行数:27,代码来源:rfi.py


示例17: attack

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def attack(self, payload, url):
        try:
            # Current request parameters
            params = dict(parse_qsl(urlsplit(url).query))
            # Change the value of the parameters with the payload
            tainted_params = {x: payload for x in params}
            if len(tainted_params) > 0:
                # Prepare the attack URL
                attack_url = urlsplit(url).geturl() + urlencode(tainted_params)
                self.output.debug("Testing: %s" % attack_url)
                resp = self.request.send(
                    url=attack_url, method="GET", payload=None, headers=None
                )
                if re.search(r"XPATH syntax error:|XPathException", resp.text, re.I):
                    self.output.finding(
                        "That site may be vulnerable to XPath Injection at %s\nInjection: %s"
                        % (url, payload)
                    )
        except Exception as e:
            self.logger.error(e)
            self.output.error("Error occured\nAborting this attack...\n")
            self.output.debug("Traceback: %s" % e)
            return 
开发者ID:shenril,项目名称:Sitadel,代码行数:26,代码来源:xpath.py


示例18: attack

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def attack(self, payload, url):
        try:
            # Current request parameters
            params = dict(parse_qsl(urlsplit(url).query))
            # Change the value of the parameters with the payload
            tainted_params = {x: payload for x in params}
            if len(tainted_params) > 0:
                # Prepare the attack URL
                attack_url = urlsplit(url).geturl() + urlencode(tainted_params)
                self.output.debug("Testing: %s" % attack_url)
                resp = self.request.send(
                    url=attack_url, method="GET", payload=None, headers=None
                )
                erro = self.dberror(resp.text)
                if erro is not None:
                    self.output.finding(
                        "That site may be vulnerable to SQL Injection at %s\nInjection: %s"
                        % (url, payload)
                    )
        except Exception as e:
            self.logger.error(e)
            self.output.error("Error occured\nAborting this attack...\n")
            self.output.debug("Traceback: %s" % e)
            return 
开发者ID:shenril,项目名称:Sitadel,代码行数:27,代码来源:sql.py


示例19: attack

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def attack(self, payload, url):
        try:
            # Current request parameters
            params = dict(parse_qsl(urlsplit(url).query))
            # Change the value of the parameters with the payload
            tainted_params = {x: payload for x in params}
            if len(tainted_params) > 0:
                # Prepare the attack URL
                attack_url = urlsplit(url).geturl() + urlencode(tainted_params)
                self.output.debug("Testing: %s" % attack_url)
                resp = self.request.send(
                    url=attack_url, method="GET", payload=None, headers=None
                )
                if resp.status_code == 200:
                    if re.search(payload, resp.text):
                        self.output.finding(
                            "That site is may be vulnerable to HTML Code Injection at %s\nInjection: %s"
                            % (url, payload)
                        )
        except Exception as e:
            self.logger.error(e)
            self.output.error("Error occured\nAborting this attack...\n")
            self.output.debug("Traceback: %s" % e)
            return 
开发者ID:shenril,项目名称:Sitadel,代码行数:27,代码来源:html.py


示例20: attack

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def attack(self, payload, url):
        try:
            # Current request parameters
            params = dict(parse_qsl(urlsplit(url).query))
            # Change the value of the parameters with the payload
            tainted_params = {x: payload for x in params}
            if len(tainted_params) > 0:
                # Prepare the attack URL
                attack_url = urlsplit(url).geturl() + urlencode(tainted_params)
                self.output.debug("Testing: %s" % attack_url)
                resp = self.request.send(
                    url=attack_url, method="GET", payload=None, headers=None
                )
                if resp.status_code == 200:
                    if re.search(payload[0], resp.text, re.I):
                        self.output.finding(
                            "That site may be vulnerable to Cross Site Scripting (XSS) at %s \nInjection: %s"
                            % (url, payload[0])
                        )
        except Exception as e:
            self.logger.error(e)
            self.output.error("Error occured\nAborting this attack...\n")
            self.output.debug("Traceback: %s" % e)
            return 
开发者ID:shenril,项目名称:Sitadel,代码行数:27,代码来源:xss.py


示例21: do_GET

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def do_GET(self):
        """Get requests."""
        q = urlparse(self.path)
        request = {
            "headers": dict(self.headers),
            "path": q.path,
            "queryStringParameters": dict(parse_qsl(q.query)),
            "httpMethod": self.command,
        }
        response = app(request, None)
        self.send_response(int(response["statusCode"]))
        for r in response["headers"]:
            self.send_header(r, response["headers"][r])
        self.end_headers()
        if isinstance(response["body"], str):
            self.wfile.write(bytes(response["body"], "utf-8"))
        else:
            self.wfile.write(response["body"]) 
开发者ID:vincentsarago,项目名称:lambda-proxy,代码行数:22,代码来源:cli.py


示例22: do_POST

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def do_POST(self):
        """POST requests."""
        q = urlparse(self.path)
        body = self.rfile.read(int(dict(self.headers).get("Content-Length")))
        body = base64.b64encode(body).decode()
        request = {
            "headers": dict(self.headers),
            "path": q.path,
            "queryStringParameters": dict(parse_qsl(q.query)),
            "body": body,
            "httpMethod": self.command,
            "isBase64Encoded": True,
        }
        response = app(request, None)
        self.send_response(int(response["statusCode"]))
        for r in response["headers"]:
            self.send_header(r, response["headers"][r])
        self.end_headers()
        if isinstance(response["body"], str):
            self.wfile.write(bytes(response["body"], "utf-8"))
        else:
            self.wfile.write(response["body"]) 
开发者ID:vincentsarago,项目名称:lambda-proxy,代码行数:26,代码来源:cli.py


示例23: get_parse_callable

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def get_parse_callable():
    class Result(dict):
        def get(self, k):
            return 200
    def parse(url):
        for k, v in parse_qsl(url.split("?")[1]):
            if k == "max_results":
                max_result = int(v)
        result = Result()
        result['entries'] = [
            get_random_arxiv_entry() for _ in range(max_result)]
        return result
    return parse 
开发者ID:lukasschwab,项目名称:arxiv.py,代码行数:21,代码来源:test_search.py


示例24: get_oauth_url

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def get_oauth_url(self):
        """ Returns the URL with OAuth params """
        params = OrderedDict()
        if "?" in self.url:
            url = self.url[:self.url.find("?")]
            for key, value in parse_qsl(urlparse(self.url).query):
                params[key] = value
        else:
            url = self.url
        params["oauth_consumer_key"] = self.consumer_key
        params["oauth_timestamp"] = self.timestamp
        params["oauth_nonce"] = self.generate_nonce()
        params["oauth_signature_method"] = "HMAC-SHA256"
        params["oauth_signature"] = self.generate_oauth_signature(params, url)
        query_string = urlencode(params)
        return "%s?%s" % (url, query_string) 
开发者ID:woocommerce,项目名称:wc-api-python,代码行数:22,代码来源:oauth.py


示例25: get_oauth_token

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def get_oauth_token(config):
    global consumer
    global client
    global req_token
    consumer = oauth.Consumer(config.consumer_key, config.consumer_secret)
    client = oauth.Client(consumer)
    resp, content = client.request(config.request_token_url, 'GET')
    if resp['status'] != '200':
        raise Exception("Invalid response {}".format(resp['status']))
    request_token = dict(parse_qsl(content.decode('utf-8')))
    req_token = RequestToken(**request_token) 
开发者ID:PacktPublishing,项目名称:Python-Programming-Blueprints,代码行数:19,代码来源:twitter_auth.py


示例26: callback

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def callback():
    global req_token
    global consumer
    config = read_config()
    oauth_verifier = request.args.get('oauth_verifier', '')
    token = oauth.Token(req_token.oauth_token,
                        req_token.oauth_token_secret)
    token.set_verifier(oauth_verifier)
    client = oauth.Client(consumer, token)
    resp, content = client.request(config.access_token_url, 'POST')
    access_token = dict(parse_qsl(content.decode('utf-8')))
    with open('.twitterauth', 'w') as req_auth:
        file_content = yaml.dump(access_token, default_flow_style=False)
        req_auth.write(file_content)
    return 'All set! You can close the browser window and stop the server.' 
开发者ID:PacktPublishing,项目名称:Python-Programming-Blueprints,代码行数:26,代码来源:twitter_auth.py


示例27: execute_request

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def execute_request(hashtag):
    config = read_config()
    if hashtag.refresh_url:
        refresh_url = hashtag.refresh_url[1:]
        url_params = dict(parse_qsl(refresh_url))
    else:
        url_params = {
            'q': f'#{hashtag.name}',
            'result_type': 'mixed'
        }
    url = prepare_request(config.search_endpoint, url_params)
    data = requests.get(url)
    results = json.loads(data.text)
    return (hashtag, results, ) 
开发者ID:PacktPublishing,项目名称:Python-Programming-Blueprints,代码行数:21,代码来源:request.py


示例28: process_access_token

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def process_access_token(request, service, response):
    if response.status_code != requests.codes.ok:
        raise Exception('Response status code not equal ok.')
    try:
        access_token = json.loads(response.text)
    except Exception:
        access_token = dict(parse_qsl(response.text))
    if service.data_header:
        args = model_to_dict(service)
        args.update(access_token)
        headers = json.loads(service.data_header % args)
    else:
        headers = None
    response = requests.get(service.data_uri % access_token, headers=headers)
    return process_data(request, service, access_token, response) 
开发者ID:aropan,项目名称:clist,代码行数:19,代码来源:views.py


示例29: test_authentication_request

# 需要导入模块: from urllib import parse [as 别名]
# 或者: from urllib.parse import parse_qsl [as 别名]
def test_authentication_request(self):
        extra_user_auth_params = {'foo': 'bar', 'abc': 'xyz'}
        config = ProviderConfiguration(provider_metadata=self.PROVIDER_METADATA,
                                       client_metadata=self.CLIENT_METADATA,
                                       auth_request_params=extra_user_auth_params)
        state = 'test_state'
        nonce = 'test_nonce'
        facade = PyoidcFacade(config, self.REDIRECT_URI)
        extra_lib_auth_params = {'foo': 'baz', 'qwe': 'rty'}
        auth_request = facade.authentication_request(state, nonce, extra_lib_auth_params)
        assert auth_request.startswith(self.PROVIDER_METADATA['authorization_endpoint'])
        auth_request_params = dict(parse_qsl(urlparse(auth_request).query))
        expected_auth_params = {
            'scope': 'openid',
            'response_type': 'code',
            'client_id': self.CLIENT_METADATA['client_id'],
            'redirect_uri': self.REDIRECT_URI,
            'state': state,
            'nonce': nonce
        }
        expected_auth_params.update(extra_user_auth_params)
        expected_auth_params.update(extra_lib_auth_params)
        assert auth_request_params == expected_auth_params 
开发者ID:zamzterz,项目名称:Flask-pyoidc,代码行数:28,代码来源:test_pyoidc_facade.py



注:本文中的urllib.parse.parse_qsl方法示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。