def code(self, var=None, **params):
""" The code page
"""
client_id = Config.get_config(Config.FIELD_CLIENT_ID)
client_secret = Config.get_config(Config.FIELD_CLIENT_SECRET)
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {
"client_id" : client_id,
"client_secret" : client_secret,
"code" : code,
"grant_type" : "authorization_code",
"redirect_uri" : callback
}
result = requests.post(AlexaService.AMAZON_TOKEN_URL, data=payload)
result = result.json()
# Save the refresh token and reset access token
Config.save_config(
Config.FIELD_REFRESH_TOKEN,
format(result['refresh_token'])
)
Config.save_config(Config.FIELD_ACCESS_TOKEN, "")
html = "<b>Success!</b><br/>"
html += "Refresh token has been added to your credentials file.<br/>"
html += "You may now reboot the Pi or restart the service.<br/>"
html += "Your token: %s" % result['refresh_token']
return html
python类url()的实例源码
def code(self, var=None, **params):
code = quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID,
"client_secret" : Client_Secret,
"code" : code,
"grant_type" : "authorization_code",
"redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
# write refresh_token line from AMZN response
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
# sent to provisioning browser
success_msg = "<H1>Success!</H1>"
success_msg +="<p>{} has been provisioned</p>".format(ProductID)
success_msg +="<p>The refresh token has been added to your creds file</p>"
if debug is True:
success_msg +="<H2>Debug</H2>"
success_msg +="<p>refresh_token:<br>"
success_msg +="{}</p>".format(resp['refresh_token'])
return success_msg
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id": Client_ID, "client_secret": Client_Secret,
"code": code, "grant_type": "authorization_code", "redirect_uri": callback}
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data=payload)
resp = r.json()
line = 'refresh_token = "' + resp['refresh_token'] + '"'
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def put(self, variant, size):
"""Store the current variant in the cache."""
request = cherrypy.serving.request
response = cherrypy.serving.response
uri = cherrypy.url(qs=request.query_string)
uricache = self.store.get(uri)
if uricache is None:
uricache = AntiStampedeCache()
uricache.selecting_headers = [
e.value for e in response.headers.elements('Vary')]
self.store[uri] = uricache
if len(self.store) < self.maxobjects:
total_size = self.cursize + size
# checks if there's space for the object
if (size < self.maxobj_size and total_size < self.maxsize):
# add to the expirations list
expiration_time = response.time + self.delay
bucket = self.expirations.setdefault(expiration_time, [])
bucket.append((size, uri, uricache.selecting_headers))
# add to the cache
header_values = [request.headers.get(h, '')
for h in uricache.selecting_headers]
uricache[tuple(sorted(header_values))] = variant
self.tot_puts += 1
self.cursize = total_size
def delete(self):
"""Remove ALL cached variants of the current resource."""
uri = cherrypy.url(qs=cherrypy.serving.request.query_string)
self.store.pop(uri, None)
def __init__(self, urls, status=None, encoding=None):
import cherrypy
request = cherrypy.serving.request
if isinstance(urls, text_or_bytes):
urls = [urls]
abs_urls = []
for url in urls:
url = tonative(url, encoding or self.encoding)
# Note that urljoin will "do the right thing" whether url is:
# 1. a complete URL with host (e.g. "http://www.example.com/test")
# 2. a URL relative to root (e.g. "/dummy")
# 3. a URL relative to the current path
# Note that any query string in cherrypy.request is discarded.
url = _urljoin(cherrypy.url(), url)
abs_urls.append(url)
self.urls = abs_urls
# RFC 2616 indicates a 301 response code fits our goal; however,
# browser support for 301 is quite messy. Do 302/303 instead. See
# http://www.alanflavell.org.uk/www/post-redirect.html
if status is None:
if request.protocol >= (1, 1):
status = 303
else:
status = 302
else:
status = int(status)
if status < 300 or status > 399:
raise ValueError('status must be between 300 and 399.')
self.status = status
CherryPyException.__init__(self, abs_urls, status)
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def authresponse(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id": Client_ID, "client_secret": Client_Secret, "code": code,
"grant_type": "authorization_code", "redirect_uri": callback}
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data=payload)
resp = r.json()
line = '\nrefresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now run <code>python alexa.py</code> <br>{}".format(
resp['refresh_token'])
def logged_in(f: callable, *args, **kwargs):
if cmd_args.no_anonymous_access:
if not cherrypy.session.get('logged_in'):
url = cherrypy.url() # http://0.0.0.0:8080/dbs
splits = url.split('/') # ['https:', '', '0.0.0.0:8080', 'dbs']
if len(splits) > 3 and splits[3] in ['dbs', 'metrics', 'logs']:
raise cherrypy.HTTPRedirect('/login' + ('?returl=/' + '/'.join(splits[3:])))
else:
raise cherrypy.HTTPRedirect('/login')
return f(*args, **kwargs)
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
return "Done, add the following line to your creds file:<br><br>refresh_token ='%s'" % resp['refresh_token']
def authresponse(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"code": code,
"grant_type": "authorization_code",
"redirect_uri": callback
}
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data=payload)
resp = r.json()
return "Success! Here is your refresh token:<br>{}".format(
resp['refresh_token'])
def put(self, variant, size):
"""Store the current variant in the cache."""
request = cherrypy.serving.request
response = cherrypy.serving.response
uri = cherrypy.url(qs=request.query_string)
uricache = self.store.get(uri)
if uricache is None:
uricache = AntiStampedeCache()
uricache.selecting_headers = [
e.value for e in response.headers.elements('Vary')]
self.store[uri] = uricache
if len(self.store) < self.maxobjects:
total_size = self.cursize + size
# checks if there's space for the object
if (size < self.maxobj_size and total_size < self.maxsize):
# add to the expirations list
expiration_time = response.time + self.delay
bucket = self.expirations.setdefault(expiration_time, [])
bucket.append((size, uri, uricache.selecting_headers))
# add to the cache
header_values = [request.headers.get(h, '')
for h in uricache.selecting_headers]
uricache[tuple(sorted(header_values))] = variant
self.tot_puts += 1
self.cursize = total_size
def delete(self):
"""Remove ALL cached variants of the current resource."""
uri = cherrypy.url(qs=cherrypy.serving.request.query_string)
self.store.pop(uri, None)
def index(self):
return """<html>
<body>Try some <a href='%s?a=7'>other</a> path,
or a <a href='%s?n=14'>default</a> path.<br />
Or, just look at the pretty picture:<br />
<img src='%s' />
</body></html>""" % (url("other"), url("else"),
url("files/made_with_cherrypy_small.png"))
def testSlashes(self):
# Test that requests for index methods without a trailing slash
# get redirected to the same URI path with a trailing slash.
# Make sure GET params are preserved.
self.getPage("/redirect?id=3")
self.assertStatus(301)
self.assertMatchesBody('<a href=([\'"])%s/redirect/[?]id=3\\1>'
"%s/redirect/[?]id=3</a>" % (self.base(), self.base()))
if self.prefix():
# Corner case: the "trailing slash" redirect could be tricky if
# we're using a virtual root and the URI is "/vroot" (no slash).
self.getPage("")
self.assertStatus(301)
self.assertMatchesBody("<a href=(['\"])%s/\\1>%s/</a>" %
(self.base(), self.base()))
# Test that requests for NON-index methods WITH a trailing slash
# get redirected to the same URI path WITHOUT a trailing slash.
# Make sure GET params are preserved.
self.getPage("/redirect/by_code/?code=307")
self.assertStatus(301)
self.assertMatchesBody("<a href=(['\"])%s/redirect/by_code[?]code=307\\1>"
"%s/redirect/by_code[?]code=307</a>"
% (self.base(), self.base()))
# If the trailing_slash tool is off, CP should just continue
# as if the slashes were correct. But it needs some help
# inside cherrypy.url to form correct output.
self.getPage('/url?path_info=page1')
self.assertBody('%s/url/page1' % self.base())
self.getPage('/url/leaf/?path_info=page1')
self.assertBody('%s/url/page1' % self.base())
def testFlatten(self):
for url in ["/flatten/as_string", "/flatten/as_list",
"/flatten/as_yield", "/flatten/as_dblyield",
"/flatten/as_refyield"]:
self.getPage(url)
self.assertBody('content')
def __init__(self, urls, status=None, encoding=None):
import cherrypy
request = cherrypy.serving.request
if isinstance(urls, text_or_bytes):
urls = [urls]
abs_urls = []
for url in urls:
url = tonative(url, encoding or self.encoding)
# Note that urljoin will "do the right thing" whether url is:
# 1. a complete URL with host (e.g. "http://www.example.com/test")
# 2. a URL relative to root (e.g. "/dummy")
# 3. a URL relative to the current path
# Note that any query string in cherrypy.request is discarded.
url = _urljoin(cherrypy.url(), url)
abs_urls.append(url)
self.urls = abs_urls
# RFC 2616 indicates a 301 response code fits our goal; however,
# browser support for 301 is quite messy. Do 302/303 instead. See
# http://www.alanflavell.org.uk/www/post-redirect.html
if status is None:
if request.protocol >= (1, 1):
status = 303
else:
status = 302
else:
status = int(status)
if status < 300 or status > 399:
raise ValueError("status must be between 300 and 399.")
self.status = status
CherryPyException.__init__(self, abs_urls, status)
def url(self, path="", qs="", script_name=None, relative=None):
"""Create an absolute URL for the given path.
If 'path' starts with a slash ('/'), this will return (base +
script_name + path + qs). If it does not start with a slash,
this returns (base url + script_name [+ request.path_info] +
path + qs).
If script_name is None, an appropriate value will be
automatically determined from the current request path.
If no parameters are specified, an absolute URL for the current
request path (minus the querystring) by passing no args. If
url(qs=request.query_string), is called, the original client URL
(assuming no internal redirections) should be returned.
If relative is None or not provided, an appropriate value will
be automatically determined. If False, the output will be an
absolute URL (including the scheme, host, vhost, and
script_name). If True, the output will instead be a URL that
is relative to the current request path, perhaps including '..'
atoms. If relative is the string 'server', the output will
instead be a URL that is relative to the server root; i.e., it
will start with a slash.
"""
return cherrypy.url(path=path, qs=qs, script_name=script_name,
relative=relative)
