本文整理汇总了Python中tornado.httputil.url_concat函数的典型用法代码示例。如果您正苦于以下问题:Python url_concat函数的具体用法?Python url_concat怎么用?Python url_concat使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了url_concat函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: get
def get(self):
code = self.get_argument('code')
client = AsyncHTTPClient()
# https://api.github.com/users/no13bus
params = {
'client_id': githubapi['CLIENT_ID'],
'redirect_uri': githubapi['REDIRECT_URL'],
'client_secret': githubapi['CLIENT_SECRET'],
'code': code
}
url = url_concat("https://github.com/login/oauth/access_token", params)
req = HTTPRequest(url=url, method="POST",headers={"Accept": "application/json"}, body='')
res = yield client.fetch(req)
resp_json = json.loads(res.body)
token = resp_json['access_token']
req_url = url_concat("https://api.github.com/user", {'access_token': token})
req_user = HTTPRequest(url=req_url, method="GET")
res_user = yield client.fetch(req_user)
user_info = json.loads(res_user.body)
username = user_info['login']
avatar_url = user_info['avatar_url']
user = yield self.db.user.find_one({'username': username})
if not user:
user_id = yield self.db.user.insert({'username': username, 'token': token, 'avatar_url': avatar_url})
print user_id
self.set_secure_cookie("token",token)
self.set_secure_cookie("username",username)
self.set_secure_cookie("avatar_url",avatar_url)
self.redirect('/')
开发者ID:frouty,项目名称:ohmyrepo,代码行数:31,代码来源:handlers.py
示例2: fetch
def fetch(url, method='GET', params=None, headers=None, cookies=None,
data=None, streaming_callback=None, header_callback=None,
connect_timeout=None, request_timeout=None):
"""Make an HTTP request and return a Future.
This is mostly just a wrapper for tornado.httpclient, but adds support for
sending cookies.
Raises HTTPError and IOError.
"""
if headers is None:
headers = {}
if params is not None:
url = httputil.url_concat(url, params)
if cookies is not None:
# abuse SimpleCookie to escape our cookies for us
simple_cookies = http.cookies.SimpleCookie(cookies)
headers['cookie'] = '; '.join(val.output(header='')[1:]
for val in simple_cookies.values())
http_client = httpclient.AsyncHTTPClient()
res = yield http_client.fetch(httpclient.HTTPRequest(
httputil.url_concat(url, params), method=method,
headers=httputil.HTTPHeaders(headers), body=data,
streaming_callback=streaming_callback, header_callback=header_callback,
connect_timeout=connect_timeout, request_timeout=request_timeout
))
return res
开发者ID:Equidamoid,项目名称:hangups,代码行数:27,代码来源:http_utils.py
示例3: post
def post(self):
self.set_secure_cookie("user", '', expires_days = -1)
if self.get_argument("code", False):
#print "Got code: %s url: %s" % (self.get_argument("code", False), self._url_)
self.get_authenticated_user(
redirect_uri=settings.canvas_page,
client_id=settings.fb_appid,
client_secret=settings.fb_secret,
code=self.get_argument("code"),
callback=self.async_callback(
self.onAuth))
return
# self.authorize_redirect(redirect_uri=self._url_,
# client_id=settings.fb_appid,
# extra_params={"scope": "publish_stream,offline_access"})
args = {
"redirect_uri": settings.canvas_page,
"client_id": settings.fb_appid,
"scope": "publish_stream,offline_access"
}
print url_concat(self._OAUTH_AUTHORIZE_URL, args)
self.write('<script> top.location.href = "' + url_concat(self._OAUTH_AUTHORIZE_URL, args) + '";</script>');
self.finish()
开发者ID:dvirsky,项目名称:badger,代码行数:27,代码来源:MainHandler.py
示例4: get
def get(self, name, user_path):
current_user = self.get_current_user()
if current_user and current_user.name == name:
# If people visit /user/:name directly on the Hub,
# the redirects will just loop, because the proxy is bypassed.
# Try to check for that and warn,
# though the user-facing behavior is unchainged
host_info = urlparse(self.request.full_url())
port = host_info.port
if not port:
port = 443 if host_info.scheme == 'https' else 80
if port != Server.from_url(self.proxy.public_url).port and port == self.hub.port:
self.log.warning("""
Detected possible direct connection to Hub's private ip: %s, bypassing proxy.
This will result in a redirect loop.
Make sure to connect to the proxied public URL %s
""", self.request.full_url(), self.proxy.public_url)
# logged in as correct user, spawn the server
if current_user.spawner:
if current_user.spawn_pending or current_user.proxy_pending:
# spawn has started, but not finished
self.statsd.incr('redirects.user_spawn_pending', 1)
html = self.render_template("spawn_pending.html", user=current_user)
self.finish(html)
return
# spawn has supposedly finished, check on the status
status = yield current_user.spawner.poll()
if status is not None:
if current_user.spawner.options_form:
self.redirect(url_concat(url_path_join(self.hub.base_url, 'spawn'),
{'next': self.request.uri}))
return
else:
yield self.spawn_single_user(current_user)
# set login cookie anew
self.set_login_cookie(current_user)
without_prefix = self.request.uri[len(self.hub.base_url):]
target = url_path_join(self.base_url, without_prefix)
if self.subdomain_host:
target = current_user.host + target
self.redirect(target)
self.statsd.incr('redirects.user_after_login')
elif current_user:
# logged in as a different user, redirect
self.statsd.incr('redirects.user_to_user', 1)
target = url_path_join(current_user.url, user_path or '')
self.redirect(target)
else:
# not logged in, clear any cookies and reload
self.statsd.incr('redirects.user_to_login', 1)
self.clear_login_cookie()
self.redirect(url_concat(
self.settings['login_url'],
{'next': self.request.uri},
))
开发者ID:Carreau,项目名称:jupyterhub,代码行数:57,代码来源:base.py
示例5: authenticate
def authenticate(self, handler, data=None):
code = handler.get_argument("code", False)
if not code:
raise web.HTTPError(400, "OAUTH_CALLBACK_URL has been called without a token")
http_client = AsyncHTTPClient()
# Exchange the OAuth code for a Custom OAuth Access Token
# API specifies a GET request yet requires URL parameters
token_req_param = dict(
client_id=self.client_id,
client_secret=self.client_secret,
grant_type=self.grant_type,
redirect_uri=self.oauth_callback_url,
code=code
)
if not self.oauth_access_token_url:
raise web.HTTPError(400, "OAUTH_ACCESS_TOKEN_URL is not defined")
token_url = url_concat(self.oauth_access_token_url, token_req_param)
token_req = HTTPRequest(url=token_url,
method="GET",
headers={"Accept": "application/json"},
validate_cert=True,
ca_certs=self.client_cert_path
)
resp = yield http_client.fetch(token_req)
resp_json = self.parse_response(resp)
profile_req_params = dict(
access_token=resp_json['access_token']
)
# Retrieve user information with a valid token obtained from the previous
# request
if not self.oauth_profile_url:
raise web.HTTPError(400, "OAUTH_PROFILE_URL is not defined")
profile_url = url_concat(self.oauth_profile_url, profile_req_params)
profile_req = HTTPRequest(
url=profile_url,
validate_cert=True,
ca_certs=self.client_cert_path)
# This request returns a JSON string
resp = yield http_client.fetch(profile_req)
resp_json = json.loads(resp.body.decode('utf8', 'replace'))
user_id = resp_json['id']
self.log.info("OAuth user id: %s" % user_id)
# Check is user is authorized to access to Stratio Intelligence
if not self.is_allowed(resp_json):
raise web.HTTPError(403, "User " + user_id + " is not allowed to access to Stratio Intelligence")
# User id is returned to be registered into app data base
return user_id
开发者ID:Stratio,项目名称:oauthenticator,代码行数:55,代码来源:singlesignon.py
示例6: get_ips_by_array_id
def get_ips_by_array_id(self,array_id):
urlfilter={"filter":'storagesystem eq "%s"'%(array_id),"per_page":100}
url = url_concat(SWARM_API_URL_IP,urlfilter)
response_body= self.send_swarm_request(url,method="GET")
ip_entries=json.loads(response_body)["entries"]
ipids=map(lambda x: str(x["ipid"]),ip_entries)
urlfilter={"filter":'id in (%s)'%(",".join(ipids))}
url = url_concat(SWARM_API_URL_SINGLE_IP,urlfilter)
response_body= self.send_swarm_request(url,method="GET")
ips=json.loads(response_body)["entries"]
return ips
开发者ID:cyril851122,项目名称:Twister,代码行数:11,代码来源:Swarm.py
示例7: get_reserved_array_by_user_id
def get_reserved_array_by_user_id(self,user_id):
reservations=self.get_active_reservation_by_user_id(user_id)
reservation_ids=map(lambda x: str(x["id"]),reservations)
urlfilter={"filter":'reservation in (%s)'%(",".join(reservation_ids)),"per_page":100}
url = url_concat(SWARM_API_URL_ARRAY_RESERVATION,urlfilter)
response_body= self.send_swarm_request(url,method="GET")
reservation_arrays=json.loads(response_body)["entries"]
array_ids=map(lambda x:str(x["storagesystem"]),reservation_arrays)
urlfilter={"filter":'id in (%s)'%(",".join(array_ids)),"per_page":100}
url=url_concat(SWARM_API_URL_ARRAY,urlfilter)
response_body= self.send_swarm_request(url,method="GET")
reserved_arrays=json.loads(response_body)["entries"]
return reserved_arrays
开发者ID:cyril851122,项目名称:Twister,代码行数:13,代码来源:Swarm.py
示例8: request
def request(self, url_path, method="GET", **kwargs):
params = self.build_params(url_path, **kwargs)
if not urlparse.urlparse(url_path).netloc:
url = url_concat(self.build_url(url_path, **kwargs), params)
else:
url = url_concat(url_path, params)
client = AsyncHTTPClient(force_instance=True)
try:
result = yield client.fetch(url, method=method, headers=self.build_headers())
raise Return(result)
finally:
client.close()
开发者ID:ElasticBox,项目名称:elastickube,代码行数:14,代码来源:client.py
示例9: _get_certivox_server_secret_share_dta
def _get_certivox_server_secret_share_dta(self, expires):
path = 'serverSecret'
url_params = url_concat('{0}{1}'.format(Keys.certivoxServer(), path), {
'app_id': self.app_id,
'expires': expires,
'signature': signMessage('{0}{1}{2}'.format(path, self.app_id, expires), self.app_key)
})
log.debug('MIRACL server secret request: {0}'.format(url_params))
httpclient = tornado.httpclient.HTTPClient()
try:
response = httpclient.fetch(url_params, **fetchConfig(url_params))
except tornado.httpclient.HTTPError as e:
log.error(e)
raise SecretsError('Unable to get Server Secret from the MIRACL TA server')
httpclient.close()
try:
data = json.loads(response.body)
except ValueError as e:
log.error(e)
raise SecretsError('Invalid response from TA server')
if 'serverSecret' not in data:
raise SecretsError('serverSecret not in response from TA server')
return data["serverSecret"]
开发者ID:CertiVox-i3-NTT,项目名称:milagro-mfa-server,代码行数:26,代码来源:secrets.py
示例10: authorize_redirect
def authorize_redirect(self, redirect_uri=None, client_id=None,
client_secret=None, extra_params=None,
callback=None, scope=None, response_type="code"):
"""Redirects the user to obtain OAuth authorization for this service.
Some providers require that you register a redirect URL with
your application instead of passing one via this method. You
should call this method to log the user in, and then call
``get_authenticated_user`` in the handler for your
redirect URL to complete the authorization process.
.. versionchanged:: 3.1
Returns a `.Future` and takes an optional callback. These are
not strictly necessary as this method is synchronous,
but they are supplied for consistency with
`OAuthMixin.authorize_redirect`.
.. deprecated:: 5.1
The ``callback`` argument and returned awaitable will be removed
in Tornado 6.0; this will be an ordinary synchronous function.
"""
args = {
"redirect_uri": redirect_uri,
"client_id": client_id,
"response_type": response_type
}
if extra_params:
args.update(extra_params)
if scope:
args['scope'] = ' '.join(scope)
self.redirect(
url_concat(self._OAUTH_AUTHORIZE_URL, args))
callback()
开发者ID:JackDandy,项目名称:SickGear,代码行数:34,代码来源:auth.py
示例11: get
def get(self, name):
current_user = self.get_current_user()
if current_user and current_user.name == name:
# logged in, spawn the server
if current_user.spawner:
if current_user.spawn_pending:
# spawn has started, but not finished
html = self.render_template("spawn_pending.html", user=current_user)
self.finish(html)
return
# spawn has supposedly finished, check on the status
status = yield current_user.spawner.poll()
if status is not None:
if current_user.spawner.options_form:
self.redirect(url_path_join(self.hub.server.base_url, 'spawn'))
return
else:
yield self.spawn_single_user(current_user)
# set login cookie anew
self.set_login_cookie(current_user)
without_prefix = self.request.uri[len(self.hub.server.base_url):]
target = url_path_join(self.base_url, without_prefix)
self.redirect(target)
else:
# not logged in to the right user,
# clear any cookies and reload (will redirect to login)
self.clear_login_cookie()
self.redirect(url_concat(
self.settings['login_url'],
{'next': self.request.uri,
}))
开发者ID:paul918,项目名称:jupyterhub,代码行数:32,代码来源:base.py
示例12: query
def query(self, callback, query, *args, **kw):
method = kw.get("method", "POST")
result_format = kw.get("result_format", DEFAULT_FORMAT)
content_type = DEFAULT_CONTENT_TYPE
headers = {
"Content-Type": content_type,
}
params = {
"query": query,
"format": result_format
}
url = self.endpoint_url
if method == "GET":
url = url_concat(url, params)
body = None
elif method == "POST":
body = urllib.urlencode(params)
request = HTTPRequest(url=url,
method=method,
headers=headers,
body=body,
auth_username=self.user,
auth_password=self.password,
auth_mode=self.auth_mode
)
response = yield gen.Task(self.client.fetch, request)
callback(response, *args, **kw)
开发者ID:icaromedeiros,项目名称:tornado-virtuoso-client,代码行数:32,代码来源:client.py
示例13: post
def post(self):
contract_id = self.get_argument('contract_id')
agent_id = self.get_argument('agent_id')
self.db.execute('update contract set deleted = 1 where id = %s', contract_id)
self.redirect(url_concat(self.reverse_url('agent.contract'), {'agent_id': agent_id}))
开发者ID:xutaoding,项目名称:osp_autumn,代码行数:7,代码来源:contract.py
示例14: getData
def getData(self,url,method,data,cookie):
try:
client = HTTPClient()
request = HTTPRequest(
url,
method=method,
headers={
'Cookie':cookie
}
)
if data and method=="GET":
data = json.loads(data)
url = url_concat(url,data)
request.url = url
elif data and method=="POST":
data = json.loads(data)
print data
data = urllib.urlencode(data)
request.body = data
# print request.url
response = client.fetch(request)
return response.body
except Exception,e:
# print str(e)
return None
开发者ID:HeraldStudio,项目名称:HeraldAppApi,代码行数:25,代码来源:yuyueHandler.py
示例15: post
def post(self):
form = Form(self.request.arguments, shop_schema)
if not form.validate():
return self.render("distributor/shop.html", form=form, error="error")
if form.url.value:
form.url.value = form.url.value.lower()
if not re.match(r"https?", form.url.value):
form.url.value = "http://" + form.url.value
distributor_id = self.get_argument("distributor-id")
distributor = self.db.get("select name from distributor where id = %s", distributor_id)
# 新建分销商铺
shop_id = self.db.execute_lastrowid(
"insert into distributor_shop (distributor_id,name,taobao_nick,money_manager,"
"url,distributor_name,created_at,created_by) values (%s,%s,%s,%s,%s,%s,now(),%s)",
distributor_id,
form.name.value.strip(),
form.taobao_nick.value.strip(),
form.money_manager.value,
form.url.value,
distributor.name,
self.current_user.name,
)
self.redirect(url_concat(self.reverse_url("distributor.show_shop_list"), {"distributor_id": distributor_id}))
开发者ID:xutaoding,项目名称:osp_autumn,代码行数:28,代码来源:shop.py
示例16: authorize_redirect
def authorize_redirect(
self,
redirect_uri: str = None,
client_id: str = None,
client_secret: str = None,
extra_params: Dict[str, Any] = None,
scope: str = None,
response_type: str = "code",
) -> None:
"""Redirects the user to obtain OAuth authorization for this service.
Some providers require that you register a redirect URL with
your application instead of passing one via this method. You
should call this method to log the user in, and then call
``get_authenticated_user`` in the handler for your
redirect URL to complete the authorization process.
.. versionchanged:: 6.0
The ``callback`` argument and returned awaitable were removed;
this is now an ordinary synchronous function.
"""
handler = cast(RequestHandler, self)
args = {"response_type": response_type}
if redirect_uri is not None:
args["redirect_uri"] = redirect_uri
if client_id is not None:
args["client_id"] = client_id
if extra_params:
args.update(extra_params)
if scope:
args["scope"] = " ".join(scope)
url = self._OAUTH_AUTHORIZE_URL # type: ignore
handler.redirect(url_concat(url, args))
开发者ID:bdarnell,项目名称:tornado,代码行数:34,代码来源:auth.py
示例17: toProfilesAPI
def toProfilesAPI(self, mid):
u"""PROFILES APIでLINEServerにプロフィールを問い合わせ
※Contact Response ObjectのJSONデータが返却されてくる
※TODO:複数一括リクエストは未対応
"""
# リクエストはGetでユーザーの識別子 (複数時はカンマ区切り)を渡す
http_client = tornado.httpclient.HTTPClient()
url = url_concat(self.PROFILES_API, {"mids": mid})
print(url)
result = None
try:
response = http_client.fetch(
url,
method='GET',
headers=self.REQUEST_HEADER,
body=None
)
json_dic = tornado.escape.json_decode(response.body)
result = json_dic["contacts"]
except http_client.HTTPError as e:
# HTTPError is raised for non-200 responses
print("Error: " + str(e))
except Exception as e:
# Other errors
print("Error: " + str(e))
http_client.close()
return result
开发者ID:Toshihiko-Tabata,项目名称:botw4m,代码行数:27,代码来源:testserver.py
示例18: watch
def watch(self, url_path, on_data, **kwargs):
local_data = dict(buffer="")
class WatchFuture(Future):
def cancel(self):
client.close()
logging.debug("AsyncHTTPClient closed")
def data_callback(data):
split_data = data.split("\n")
for index, fragment in enumerate(split_data):
if index + 1 < len(split_data):
on_data(json.loads(local_data["buffer"] + fragment))
local_data["buffer"] = ""
else:
local_data["buffer"] += fragment
params = self.build_params(url_path, **kwargs)
url = url_concat(self.build_url(url_path, **kwargs), params)
request = HTTPRequest(
url=url,
method="GET",
headers=self.build_headers(),
request_timeout=3600,
streaming_callback=data_callback)
client = AsyncHTTPClient(force_instance=True)
future = WatchFuture()
chain_future(client.fetch(request), future)
return future
开发者ID:ElasticBox,项目名称:elastickube,代码行数:33,代码来源:client.py
示例19: authorize_redirect
def authorize_redirect(self, redirect_uri=None, client_id=None,
client_secret=None, extra_params=None,
callback=None):
"""Redirects the user to obtain OAuth authorization for this service.
Some providers require that you register a redirect URL with
your application instead of passing one via this method. You
should call this method to log the user in, and then call
``get_authenticated_user`` in the handler for your
redirect URL to complete the authorization process.
.. versionchanged:: 3.1
Returns a `.Future` and takes an optional callback. These are
not strictly necessary as this method is synchronous,
but they are supplied for consistency with
`OAuthMixin.authorize_redirect`.
"""
args = {
"redirect_uri": redirect_uri,
"client_id": client_id
}
if extra_params:
args.update(extra_params)
self.redirect(
url_concat(self._OAUTH_AUTHORIZE_URL, args))
callback()
开发者ID:auscompgeek,项目名称:perfectgift,代码行数:26,代码来源:auth.py
示例20: get
async def get(self):
self.statsd.incr('login.request')
user = self.current_user
if user:
# set new login cookie
# because single-user cookie may have been cleared or incorrect
self.set_login_cookie(user)
self.redirect(self.get_next_url(user), permanent=False)
else:
if self.authenticator.auto_login:
auto_login_url = self.authenticator.login_url(self.hub.base_url)
if auto_login_url == self.settings['login_url']:
# auto_login without a custom login handler
# means that auth info is already in the request
# (e.g. REMOTE_USER header)
user = await self.login_user()
if user is None:
# auto_login failed, just 403
raise web.HTTPError(403)
else:
self.redirect(self.get_next_url(user))
else:
if self.get_argument('next', default=False):
auto_login_url = url_concat(
auto_login_url, {'next': self.get_next_url()}
)
self.redirect(auto_login_url)
return
username = self.get_argument('username', default='')
self.finish(self._render(username=username))
开发者ID:jupyterhub,项目名称:jupyterhub,代码行数:30,代码来源:login.py
注:本文中的tornado.httputil.url_concat函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论