def pushtxn(raw_tx):
'''Insight send raw tx API'''
url = PUSH_TX_URL
payload = urllib.urlencode({
"rawtx": raw_tx
})
result = urlfetch.fetch(url,
method=urlfetch.POST,
payload=payload
)
if result.status_code == 200:
j = json.loads(result.content)
txid = j.get('txid')
return txid, raw_tx
else:
msg = 'Error accessing insight API:'+str(result.status_code)+" "+str(result.content)
ErrorNotification.new(msg)
return None, msg
python类POST的实例源码
def exportItems( module, target, importKey, startCursor, endCursor):
Skel = skeletonByKind( module )
query = Skel().all().cursor( startCursor, endCursor )
for item in query.run(250):
flatItem = DbTransfer.genDict( item )
formFields = {
"e": pickle.dumps(flatItem).encode("HEX"),
"key": importKey
}
result = urlfetch.fetch( url=target,
payload=urllib.urlencode(formFields),
method=urlfetch.POST,
headers={'Content-Type': 'application/x-www-form-urlencoded'})
if startCursor == endCursor:
try:
utils.sendEMailToAdmins("Export of kind %s finished" % module,
"ViUR finished to export kind %s to %s.\n" % (module, target))
except: #OverQuota, whatever
pass
# --- import ---
def check_language(text):
if len(text) > MAX_LENGTH:
logging.info("trimming text from %s to %s", len(text), MAX_LENGTH)
text = text[:MAX_LENGTH]
base_url = 'https://www.googleapis.com/language/translate/v2/detect'
params = {'key': API_KEY, 'q': text}
form_data = urls.urlencode(params)
if urllib2_fallback:
request = urllib2.Request(base_url, form_data, {'X-HTTP-Method-Override': 'GET'})
response_content = urllib2.urlopen(request).read()
else:
result = urlfetch.fetch(url=base_url, payload=form_data, method=urlfetch.POST, headers={'X-HTTP-Method-Override': 'GET'})
if result.status_code != 200:
error = "result status code is %s for content %s" % (result.status_code, result.content)
logging.error(error)
raise Exception("Error in translation: %s" % error)
response_content = result.content
json_content = json.loads(response_content)
real_results = json_content['data']['detections'][0][0]
logging.info("text classification returned %s", real_results)
if real_results['confidence'] > 0.10:
return real_results['language']
else:
return None
def auth_callback_provider( self ):
params = { 'code': self.request.get( 'code' ),
'client_id': self.get_auth_request_client_id(),
'client_secret': self.get_client_secret(),
'redirect_uri': self.domain_name[ :-1 ] + self.get_auth_callback(),
'grant_type': 'authorization_code',
}
urlParams = enki.libutil.urlencode( params )
url = self.token_endpoint()
result = self.urlfetch_safe( url = url,
payload = urlParams,
method = urlfetch.POST,
headers = { 'Content-Type': 'application/x-www-form-urlencoded' }
)
self.process_token_result( result )
def get_access_token(force=False):
"""Tries to obtain access token from memcache and, if it fails,
obtains a new set and stores in memcache.
See https://dev.twitter.com/oauth/application-only.
Deleting the memcache key `access_token` will trigger a token refresh.
"""
token = memcache.get('access_token')
if force or token is None:
logging.warning('Needed to fetch access_token')
encoded_key = urllib.quote_plus(CUSTOMER_KEY)
encoded_secret = urllib.quote_plus(CUSTOMER_SECRET)
encoded_credentials = base64.b64encode(
"{}:{}".format(encoded_key, encoded_secret))
response = urlfetch.fetch(
'https://api.twitter.com/oauth2/token',
payload='grant_type=client_credentials',
method=urlfetch.POST,
headers={'Authorization': 'Basic ' + encoded_credentials})
if response.status_code == urlfetch.httplib.OK:
response_data = json.loads(response.content)
token = response_data['access_token']
memcache.set('access_token', token, 2592000) # 30 days
return token
def urlread(url, data=None, headers=None):
if data is not None:
if headers is None:
headers = {"Content-type": "application/x-www-form-urlencoded"}
method = urlfetch.POST
else:
if headers is None:
headers = {}
method = urlfetch.GET
result = urlfetch.fetch(url, method=method,
payload=data, headers=headers)
if result.status_code == 200:
return result.content
else:
raise urllib2.URLError("fetch error url=%s, code=%d" % (url, result.status_code))
def get_request_token(base):
'''
Get request token
'''
data = urllib.urlencode({
'consumer_key': POCKET_CONSUMER_KEY,
'redirect_uri': base + POCKET_FINISH_REDIRECT
})
logging.debug(data)
res = urlfetch.fetch(
url=POCKET_OAUTH_REQUEST,
method=urlfetch.POST,
payload=data,
validate_certificate=True)
code = redirect = None
logging.debug(res.status_code)
if res.status_code == 200:
result = res.content
if 'code=' in result:
code = result.replace('code=','')
redirect = POCKET_AUTHORIZE_REDIR + '?request_token=%s&redirect_uri=%s' % (code, base + POCKET_FINISH_REDIRECT)
return (code, redirect)
def get_access_token(code):
'''
Get request token
'''
data = urllib.urlencode({
'consumer_key': POCKET_CONSUMER_KEY,
'code': code
})
logging.debug(data)
res = urlfetch.fetch(
url=POCKET_OAUTH_AUTHORIZE,
method=urlfetch.POST,
payload=data,
validate_certificate=True)
code = redirect = None
logging.debug(res.status_code)
if res.status_code == 200:
result = res.content
data = urlparse.parse_qs(result)
access_token = data.get('access_token', [None])[0]
return access_token
def fetch(url, data=None, headers=None,
cookie=Cookie.SimpleCookie(),
user_agent='Mozilla/5.0'):
headers = headers or {}
if data is not None:
data = urllib.urlencode(data)
if user_agent:
headers['User-agent'] = user_agent
headers['Cookie'] = ' '.join(
['%s=%s;' % (c.key, c.value) for c in cookie.values()])
try:
from google.appengine.api import urlfetch
except ImportError:
req = urllib2.Request(url, data, headers)
html = urllib2.urlopen(req).read()
else:
method = ((data is None) and urlfetch.GET) or urlfetch.POST
while url is not None:
response = urlfetch.fetch(url=url, payload=data,
method=method, headers=headers,
allow_truncated=False, follow_redirects=False,
deadline=10)
# next request will be a get, so no need to send the data again
data = None
method = urlfetch.GET
# load cookies from the response
cookie.load(response.headers.get('set-cookie', ''))
url = response.headers.get('location')
html = response.content
return html
def obtain_bearer_token(host, path):
"""Given a bearer token, send a GET request to the API.
Args:
host (str): The domain host of the API.
path (str): The path of the API after the domain.
params (dict): An optional set of query parameters in the request.
Returns:
str: OAuth bearer token, obtained using client_id and client_secret.
Raises:
HTTPError: An error occurs from the HTTP request.
"""
url = '{0}{1}'.format(host, quote(path.encode('utf8')))
data = urlencode({
'client_id': CLIENT_ID,
'client_secret': CLIENT_SECRET,
'grant_type': GRANT_TYPE,
})
print('@@@@@@@@@' + CLIENT_ID)
headers = {
'content-type': 'application/x-www-form-urlencoded',
}
result = urlfetch.fetch(
url=url,
payload=data,
method=urlfetch.POST,
headers=headers)
print('@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@' + result.content)
return "BIO6_LpbIcFkeKDB9SsSAONt3lE2IwrdiTxUeq-Ag1MKOzSc4m-8QyPjdV6WmI27ySuLEKv7czHoJmJjFHrCyjfgxucTvKPpJG9JCsg_08KCz4J-WrEfeaiACoJ2WXYx"
def fetch(url, data=None, headers=None,
cookie=Cookie.SimpleCookie(),
user_agent='Mozilla/5.0'):
headers = headers or {}
if not data is None:
data = urllib.urlencode(data)
if user_agent:
headers['User-agent'] = user_agent
headers['Cookie'] = ' '.join(
['%s=%s;' % (c.key, c.value) for c in cookie.values()])
try:
from google.appengine.api import urlfetch
except ImportError:
req = urllib2.Request(url, data, headers)
html = urllib2.urlopen(req).read()
else:
method = ((data is None) and urlfetch.GET) or urlfetch.POST
while url is not None:
response = urlfetch.fetch(url=url, payload=data,
method=method, headers=headers,
allow_truncated=False, follow_redirects=False,
deadline=10)
# next request will be a get, so no need to send the data again
data = None
method = urlfetch.GET
# load cookies from the response
cookie.load(response.headers.get('set-cookie', ''))
url = response.headers.get('location')
html = response.content
return html
def fetch(url, data=None, headers=None,
cookie=Cookie.SimpleCookie(),
user_agent='Mozilla/5.0'):
headers = headers or {}
if data is not None:
data = urllib.urlencode(data)
if user_agent:
headers['User-agent'] = user_agent
headers['Cookie'] = ' '.join(
['%s=%s;' % (c.key, c.value) for c in cookie.values()])
try:
from google.appengine.api import urlfetch
except ImportError:
req = urllib2.Request(url, data, headers)
html = urllib2.urlopen(req).read()
else:
method = ((data is None) and urlfetch.GET) or urlfetch.POST
while url is not None:
response = urlfetch.fetch(url=url, payload=data,
method=method, headers=headers,
allow_truncated=False, follow_redirects=False,
deadline=10)
# next request will be a get, so no need to send the data again
data = None
method = urlfetch.GET
# load cookies from the response
cookie.load(response.headers.get('set-cookie', ''))
url = response.headers.get('location')
html = response.content
return html
def sendData(form_field):
url = "https://api.line.me/v2/bot/message/push"
result = urlfetch.fetch(
url=url,
payload=json.dumps(form_field, ensure_ascii=False),
method=urlfetch.POST,
headers={
'Content-type': 'application/json',
'Authorization': 'Bearer ' + const.ChannelAccessToken
}
)
if result.status_code == 200:
logging.debug(result.content)
else:
logging.debug(result.content)
def sendData(ifttt_event, ifttt_key, senderID, from_id, event_type, msg_data):
if ifttt_key == None:
send2Line.sendText(str(senderID), const.MSG_NONREGISTRATION)
return
if msg_data.count('\n'):
msg_data = u"<pre>" + msg_data + u"</pre>"
form_fields = {
"value1": from_id,
"value2": event_type,
"value3": msg_data,
}
logging.debug(form_fields)
logging.debug(ifttt_key)
requestStr = json.dumps(form_fields, ensure_ascii=False)
# requestStr=requestStr.replace('\n', '\\n').replace('\r', '')
url = "https://maker.ifttt.com/trigger/" + ifttt_event + "/with/key/" + ifttt_key
result = urlfetch.fetch(
url=url,
payload=requestStr,
method=urlfetch.POST,
headers={
'Content-Type': ' application/json'
}
)
logging.debug(result.status_code)
if result.status_code == 200:
logging.debug(result.content)
elif result.status_code == 401:
send2Line.sendText(str(senderID), const.MSG_MAKERKEY_FAILED)
logging.debug(result.content)
else:
logging.debug(result.content)
def telegram_post(data, deadline=3):
return urlfetch.fetch(url=TELEGRAM_URL_SEND, payload=data, method=urlfetch.POST,
headers=JSON_HEADER, deadline=deadline)
def telegram_query(uid, deadline=3):
data = json.dumps({'chat_id': uid, 'action': 'typing'})
return urlfetch.fetch(url=TELEGRAM_URL_CHAT_ACTION, payload=data, method=urlfetch.POST,
headers=JSON_HEADER, deadline=deadline)
def telegram_photo(data, deadline=3):
return urlfetch.fetch(url=TELEGRAM_URL_SEND_PHOTO, payload=data, method=urlfetch.POST,
headers=JSON_HEADER, deadline=deadline)
def send_typing(uid):
data = json.dumps({'chat_id': uid, 'action': 'typing'})
try:
rpc = urlfetch.create_rpc()
urlfetch.make_fetch_call(rpc, url=TELEGRAM_URL_CHAT_ACTION, payload=data,
method=urlfetch.POST, headers=JSON_HEADER)
except:
return
def get(self):
taskqueue.add(url='/promo', method="POST")
def fetch(url, data=None, headers=None,
cookie=Cookie.SimpleCookie(),
user_agent='Mozilla/5.0'):
headers = headers or {}
if data is not None:
data = urllib.urlencode(data)
if user_agent:
headers['User-agent'] = user_agent
headers['Cookie'] = ' '.join(
['%s=%s;' % (c.key, c.value) for c in cookie.values()])
try:
from google.appengine.api import urlfetch
except ImportError:
req = urllib2.Request(url, data, headers)
html = urllib2.urlopen(req).read()
else:
method = ((data is None) and urlfetch.GET) or urlfetch.POST
while url is not None:
response = urlfetch.fetch(url=url, payload=data,
method=method, headers=headers,
allow_truncated=False, follow_redirects=False,
deadline=10)
# next request will be a get, so no need to send the data again
data = None
method = urlfetch.GET
# load cookies from the response
cookie.load(response.headers.get('set-cookie', ''))
url = response.headers.get('location')
html = response.content
return html
def fromClient(self, valuesCache, name, data):
"""
Reads a value from the client.
If this value is valid for this bone,
store this value and return None.
Otherwise our previous value is
left unchanged and an error-message
is returned.
:param name: Our name in the skeleton
:type name: String
:param data: *User-supplied* request-data
:type data: Dict
:returns: None or String
"""
if request.current.get().isDevServer: #We dont enforce captchas on dev server
return( None )
user = utils.getCurrentUser()
if user and "root" in user["access"]: # Don't bother trusted users with this (not supported by admin/vi anyways)
return( None )
if not "recaptcha_challenge_field" in data.keys() or not "recaptcha_response_field" in data.keys():
return( u"No Captcha given!" )
data = { "privatekey": self.privateKey,
"remoteip": request.current.get().request.remote_addr,
"challenge": data["recaptcha_challenge_field"],
"response": data["recaptcha_response_field"]
}
response = urlfetch.fetch( url="http://www.google.com/recaptcha/api/verify",
payload=urllib.urlencode( data ),
method=urlfetch.POST,
headers={"Content-Type": "application/x-www-form-urlencoded"} )
if str(response.content).strip().lower().startswith("true"):
return( None )
return( u"Invalid Captcha" )
def get(self):
# [START urlfetch-post]
try:
form_data = urllib.urlencode(UrlPostHandler.form_fields)
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
result = urlfetch.fetch(
url='http://localhost:8080/submit_form',
payload=form_data,
method=urlfetch.POST,
headers=headers)
self.response.write(result.content)
except urlfetch.Error:
logging.exception('Caught exception fetching url')
# [END urlfetch-post]
def fetch(url, data=None, headers=None,
cookie=Cookie.SimpleCookie(),
user_agent='Mozilla/5.0'):
headers = headers or {}
if not data is None:
data = urllib.urlencode(data)
if user_agent:
headers['User-agent'] = user_agent
headers['Cookie'] = ' '.join(
['%s=%s;' % (c.key, c.value) for c in cookie.values()])
try:
from google.appengine.api import urlfetch
except ImportError:
req = urllib2.Request(url, data, headers)
html = urllib2.urlopen(req).read()
else:
method = ((data is None) and urlfetch.GET) or urlfetch.POST
while url is not None:
response = urlfetch.fetch(url=url, payload=data,
method=method, headers=headers,
allow_truncated=False, follow_redirects=False,
deadline=10)
# next request will be a get, so no need to send the data again
data = None
method = urlfetch.GET
# load cookies from the response
cookie.load(response.headers.get('set-cookie', ''))
url = response.headers.get('location')
html = response.content
return html
def __init__(self, host, port=None, strict=False, timeout=None):
from google.appengine.api import urlfetch
self._fetch = urlfetch.fetch
self._method_map = {
'GET': urlfetch.GET,
'POST': urlfetch.POST,
'HEAD': urlfetch.HEAD,
'PUT': urlfetch.PUT,
'DELETE': urlfetch.DELETE,
'PATCH': urlfetch.PATCH,
}
self.host = host
self.port = port
self._method = self._url = None
self._body = ''
self.headers = []
if not isinstance(timeout, (float, int, long)):
timeout = None
self.timeout = timeout
def _MakeRemoteSyncCall(self, service, call, request, response):
"""Send an RPC to a remote_api endpoint."""
request_pb = remote_api_pb.Request()
request_pb.set_service_name(service)
request_pb.set_method(call)
request_pb.set_request(request.Encode())
response_pb = remote_api_pb.Response()
encoded_request = request_pb.Encode()
try:
urlfetch_response = urlfetch.fetch(self.remote_url, encoded_request,
urlfetch.POST, self.extra_headers,
follow_redirects=False,
deadline=10)
except Exception, e:
logging.exception('Fetch failed to %s', self.remote_url)
raise FetchFailed(e)
if urlfetch_response.status_code != 200:
logging.error('Fetch failed to %s; Status %s; body %s',
self.remote_url,
urlfetch_response.status_code,
urlfetch_response.content)
raise FetchFailed(urlfetch_response.status_code)
response_pb.ParseFromString(urlfetch_response.content)
if response_pb.has_application_error():
error_pb = response_pb.application_error()
raise apiproxy_errors.ApplicationError(error_pb.code(),
error_pb.detail())
elif response_pb.has_exception():
raise pickle.loads(response_pb.exception())
elif response_pb.has_java_exception():
raise UnknownJavaServerError('An unknown error has occured in the '
'Java remote_api handler for this call.')
else:
response.ParseFromString(response_pb.response())
def __init__(self, host, port=None, strict=None,
timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None,
context=None):
# net.proto.ProcotolBuffer relies on httplib so importing urlfetch at the
# module level causes a failure on prod. That means the import needs to be
# lazy.
from google.appengine.api import urlfetch
self._fetch = urlfetch.fetch
self._method_map = {
'GET': urlfetch.GET,
'POST': urlfetch.POST,
'HEAD': urlfetch.HEAD,
'PUT': urlfetch.PUT,
'DELETE': urlfetch.DELETE,
'PATCH': urlfetch.PATCH,
}
self.host = host
self.port = port
# With urllib2 in Python 2.6, an object can be passed here.
# The default is set to socket.GLOBAL_DEFAULT_TIMEOUT which is an object.
# We only accept float, int or long values, otherwise it can be
# silently ignored.
if not isinstance(timeout, (float, int, long)):
timeout = None
self.timeout = timeout
# Both 'strict' and 'source_address' are ignored.
self._method = self._url = None
self._body = ''
self.headers = []
def auth_callback_provider( self ):
params = { 'openid.ns': '',
'openid.op_endpoint': '',
'openid.claimed_id': '',
'openid.identity': '',
'openid.return_to': '',
'openid.response_nonce': '',
'openid.assoc_handle': '',
'openid.signed': '',
'openid.sig': '',
}
for key in params:
params[ key ] = self.request.get( key )
params[ 'openid.mode' ] = 'check_authentication'
# param {'openid.claimed_id': u'http://steamcommunity.com/openid/id/7****************'}
claimedId = str( params[ 'openid.claimed_id' ])[ len( 'http://steamcommunity.com/openid/id/' ): ]
loginInfo = { 'provider_name': self.get_provider_name( ),
'provider_uid': claimedId,
'email': '',
'email_verified': '' }
urlParams = enki.libutil.urlencode( params )
fullURL = 'https://steamcommunity.com/openid/login'
result = self.urlfetch_safe( url = fullURL, payload = urlParams, method = urlfetch.POST )
if 'ns:http://specs.openid.net/auth/2.0\nis_valid:true\n' in result.content: # only if is_valid do we trust the loginInfo
self.provider_authenticated_callback( loginInfo )
#===== TWITTER =========================================================================================================
def auth_sign( self, normalised_url, ordered_params, token_secret = '', method_get = False ):
# note: create signature see https://dev.twitter.com/oauth/overview/creating-signatures
params_to_sign = enki.libutil.urlencode( ordered_params )
oauth_signature_string = ''
if method_get:
oauth_signature_string = 'GET&' + percent_encode( normalised_url ) + '&' + percent_encode( params_to_sign )
else:
oauth_signature_string = 'POST&' + percent_encode( normalised_url ) + '&' + percent_encode( params_to_sign )
key = percent_encode( settings.secrets.CLIENT_SECRET_TWITTER ) + '&' + token_secret
hmac_hash = hmac.new( key, oauth_signature_string, hashlib.sha1 )
oauth_signature = base64.b64encode( hmac_hash.digest())
return oauth_signature
def auth_request( self ):
# STEP 1
# note: these parameters need to be sorted alphabetically by key. They are therefore a list of tuples and not a dictionary.
params = [( 'oauth_callback' , self.domain_name[ :-1 ] + self.get_auth_callback()),
( 'oauth_consumer_key' , settings.secrets.CLIENT_ID_TWITTER ),
( 'oauth_nonce' , webapp2_extras.security.generate_random_string( length = 42, pool = webapp2_extras.security.ALPHANUMERIC ).encode( 'utf-8' )),
( 'oauth_signature_method' , "HMAC-SHA1" ),
( 'oauth_timestamp' , str( int( time.time()))),
( 'oauth_version' , "1.0" )]
normalised_url = 'https://api.twitter.com/oauth/request_token/'
oauth_signature = self.auth_sign( normalised_url, params )
params.append(( 'oauth_signature', oauth_signature ))
url_params = enki.libutil.urlencode( params )
result = self.urlfetch_safe( url = normalised_url, payload = url_params, method = urlfetch.POST )
response = self.process_result_as_query_string( result )
# STEP 2
if response.get( 'oauth_callback_confirmed' ) != 'true' :
self.abort( 401 )
return
else:
oauth_token = response.get( 'oauth_token' )
self.session[ 'twitter_oauth_token' ] = oauth_token
self.session[ 'twitter_oauth_token_secret' ] = response.get( 'oauth_token_secret' )
url_redirect_params = enki.libutil.urlencode([( 'oauth_token', oauth_token )])
url_redirect = 'https://api.twitter.com/oauth/authenticate?' + url_redirect_params
self.redirect( url_redirect )
return
def get_download_URL( self, enkiDL_URL, secret, item_to_download, ip_addr ):
form_fields = { 'item' : item_to_download, 'secret' : secret, 'ip_addr' : ip_addr }
form_data = enki.libutil.urlencode( form_fields )
try:
result = urlfetch.fetch( url = enkiDL_URL, payload = form_data, method = urlfetch.POST )
if result.status_code == 200:
token = result.content
self.download_url = enkiDL_URL + 'download?token=' + str( token ) + '&item=' + str( item_to_download )
else:
self.error = 1
return
except urlfetch.DownloadError:
self.error = 2
return
