本文整理汇总了Python中requests.post函数的典型用法代码示例。如果您正苦于以下问题:Python post函数的具体用法?Python post怎么用?Python post使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了post函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: worker
def worker(self):
if not ENDPOINT:
return
while True:
datapack = {'sensors': {}}
for sensor in Sensor.get_all():
data = {}
for variant in sensor.get_variants():
data[variant] = sensor.value(variant)
datapack['sensors'][sensor.id] = data
gevent.sleep(3)
url = ENDPOINT % ajenti.config.tree.installation_id
if not self.get_key():
continue
try:
requests.post(url, data={
'data': json.dumps(datapack),
'key': self.get_key()
})
self.last_report = datetime.now()
self.last_error = None
except Exception, e:
self.last_error = e
开发者ID:balchjd,项目名称:ajenti,代码行数:27,代码来源:main.py
示例2: test_Anrufer_sperren_in_web
def test_Anrufer_sperren_in_web():
u = inte_testutils.TestUtil()
u.unblock_callerid(TelephoneNumber('0790000003'))
assert not u.is_blocked_callerid(TelephoneNumber('0790000003'))
requests.post('http://localhost/block_caller', data={'telephone_number': '0790000003'})
assert u.is_blocked_callerid(TelephoneNumber('0790000003'))
开发者ID:synox,项目名称:telewall,代码行数:7,代码来源:test_InT03.py
示例3: test_api_fire_event_with_invalid_json
def test_api_fire_event_with_invalid_json(self):
"""Test if the API allows us to fire an event."""
test_value = []
def listener(event):
"""Helper method that will verify our event got called."""
test_value.append(1)
hass.bus.listen_once("test_event_bad_data", listener)
req = requests.post(
_url(const.URL_API_EVENTS_EVENT.format("test_event_bad_data")),
data=json.dumps('not an object'),
headers=HA_HEADERS)
hass.pool.block_till_done()
self.assertEqual(400, req.status_code)
self.assertEqual(0, len(test_value))
# Try now with valid but unusable JSON
req = requests.post(
_url(const.URL_API_EVENTS_EVENT.format("test_event_bad_data")),
data=json.dumps([1, 2, 3]),
headers=HA_HEADERS)
hass.pool.block_till_done()
self.assertEqual(400, req.status_code)
self.assertEqual(0, len(test_value))
开发者ID:bdfoster,项目名称:blumate,代码行数:30,代码来源:test_api.py
示例4: csv_upload
def csv_upload(self, filename, operation):
# create the request url for init.
rest_url_pre = 'https://' + self.iba_host + '/wapi/v' + self.iba_wapi_version
try:
init_url = rest_url_pre + '/fileop?_function=uploadinit'
# send request to the server to establish the object, with token and url.
r = requests.post(url=init_url, auth=(self.iba_user, self.iba_password), verify=self.iba_verify_ssl)
r_json = r.json()
# need to do escaping or we will have issues when it's sent to the requests.post
token = r_json['token'].encode('unicode-escape')
req_url = r_json['url']
files = {'file': (filename, open(filename, 'rb')) }
file_r = requests.post(url=req_url, auth=(self.iba_user, self.iba_password), verify=self.iba_verify_ssl, files=files)
# Use the token to tell the server to process the file as a csv_import
csv_url = rest_url_pre + '/fileop?_function=csv_import'
payload = '{ "token": "' + token + '", "operation": "' + operation + '" }'
requests.post(url=csv_url, auth=(self.iba_user, self.iba_password), verify=self.iba_verify_ssl, data=payload)
except ValueError:
raise Exception(r)
except Exception:
raise
开发者ID:stscott,项目名称:Infoblox-API-Python,代码行数:25,代码来源:infoblox.py
示例5: merge_kitchens_improved
def merge_kitchens_improved(self, from_kitchen, to_kitchen, resolved_conflicts=None):
"""
merges kitchens
'/v2/kitchen/merge/<string:kitchenname>/<string:kitchenname>', methods=['POST']
:param resolved_conflicts:
:param self: DKCloudAPI
:param from_kitchen: string
:param to_kitchen: string
:rtype: dict
"""
rc = DKReturnCode()
if from_kitchen is None or isinstance(from_kitchen, basestring) is False:
rc.set(rc.DK_FAIL, 'issue with from kitchen')
return rc
if to_kitchen is None or isinstance(to_kitchen, basestring) is False:
rc.set(rc.DK_FAIL, 'issue with to kitchen')
return rc
url = '%s/v2/kitchen/merge/%s/%s' % (self.get_url_for_direct_rest_call(), from_kitchen, to_kitchen)
try:
if resolved_conflicts is not None and len(resolved_conflicts) > 0:
data = dict()
data['resolved_conflicts'] = resolved_conflicts
response = requests.post(url, data=json.dumps(data), headers=self._get_common_headers())
else:
response = requests.post(url, headers=self._get_common_headers())
rdict = self._get_json(response)
except (RequestException, ValueError, TypeError), c:
rc.set("merge_kitchens: exception: %s" % str(c))
return rc
开发者ID:DataKitchen,项目名称:DKCloudCommand,代码行数:29,代码来源:DKCloudAPI.py
示例6: upload
def upload(self, path):
"""
Uploads a file to the current folder. Takes a file location as an
argument, either relative or absolute. If a file of the same name is
already in the current folder, a new version is uploaded.
Returns True if file upload is successful, False otherwise.
"""
parent_id = self.current_folder.get('id')
item_id = None
filename = path.split('/')[-1]
for entry in self.current_folder.get('item_collection').get('entries'):
if (entry.get('name') == filename and entry.get('type') == 'file'):
item_id = entry.get('id')
break
if (item_id == None):
f = requests.post('https://upload.box.com/api/2.0/files/content',
headers = { 'Authorization': 'Bearer ' + self.token },
data = { 'filename': filename, 'parent_id':parent_id },
files = { filename: open(path, 'rb')})
else:
f = requests.post('https://upload.box.com/api/2.0/files/' +
item_id + '/content', headers = { 'Authorization': 'Bearer ' +
self.token }, data = { 'filename': filename }, files = {
filename: open(path, 'rb')})
if (f.status_code == 409):
print "File upload caused a conflict"
return False
elif (f.status_code != 201):
print "Problem uploading the file"
return False
else:
print "File uploaded"
return True
开发者ID:michaelmc,项目名称:boxnix,代码行数:34,代码来源:boxfolders.py
示例7: post_to_slack
def post_to_slack(self, file_name, file_content, bot_callback=None, title=None):
slack = getSecrets('slack')
data = {
'token' : slack['api_token'],
'title' : "i deepdreamed..." if title is None else title,
'channels' : slack['channel_id']
}
try:
r = requests.post("https://slack.com/api/files.upload", data=data, files={ 'file' : file_content})
res = json.loads(r.content)
if not res['ok']:
return False
if bot_callback is not None:
r = requests.post(slack['webhook_url'], data={'payload' : json.dumps({ 'text' : bot_callback })})
print r.content
return True
except Exception as e:
print e, type(e)
return False
开发者ID:DeepLab,项目名称:deeplab_x_deepdream_annex,代码行数:26,代码来源:deepdream.py
示例8: test_diff
def test_diff(self):
r = requests.post(self.url("/test_diff?edit"), data={
"body": "# test diff\n\n"
})
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
r = requests.post(self.url("/test_diff?edit"), data={
"body": "# test diff\n\nadd an newline here \n"
})
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
r = requests.get(self.url("/test_diff?history"))
self.assertRegexpMatches(r.text, r'<a href="\?version=[0-9a-f]{40}">')
version_re = re.compile(r'<a href="\?version=([0-9a-f]{40})">')
versions = version_re.findall(r.text)
self.assertEqual(len(versions), 2)
r = requests.get(self.url("/test_diff?diff=%s,%s" % (versions[0], versions[1])))
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
self.assertIn(r'Diff for file from %s to %s' % (versions[0], versions[1]), r.text)
r = requests.get(self.url("/test_diff?diff=%s,%s" % (versions[0][:13], versions[1][:13])))
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
self.assertIn(r'Diff for file from %s to %s' % (versions[0][:13], versions[1][:13]), r.text)
r = requests.get(self.url("/test_diff?diff=%syyyy,xxxx%s" % (versions[0][:13], versions[1][:13])))
self.assertGreaterEqual(r.status_code, 400)
self.assertLess(r.status_code, 500)
开发者ID:Ibsenhasdream,项目名称:strapdown-zeta,代码行数:34,代码来源:test.py
示例9: test_file_dir_with_same_name
def test_file_dir_with_same_name(self):
text = u"file a content"
r = requests.post(self.url("a?edit"), data={
"body": text
})
self.assertIn(text, r.text)
self.assertIn("strapdown.min.js", r.text)
text2 = "does not matter"
r = requests.post(self.url("a/b?edit"), data={
"body": text2
})
self.assertIn(text2, r.text)
self.assertIn("strapdown.min.js", r.text)
# should be the file
r = requests.get(self.url("a"))
self.assertIn(text, r.text)
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
# should be the directory
r = requests.get(self.url("a/"))
self.assertNotIn(text, r.text)
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
self.assertIn('Directory Listing', r.text)
self.assertIn('b.md', r.text)
# should be file b
r = requests.get(self.url("a/b"))
self.assertIn(text2, r.text)
self.assertGreaterEqual(r.status_code, 200)
self.assertLess(r.status_code, 300)
开发者ID:Ibsenhasdream,项目名称:strapdown-zeta,代码行数:34,代码来源:test.py
示例10: send_mail
def send_mail(self, request):
fn = self.cleaned_data['firstname']
ln = self.cleaned_data['lastname']
subj = self.cleaned_data['subject']
fr = self.cleaned_data['email']
msg = self.cleaned_data['message']
try:
requests.post(
"https://api.mailgun.net/v2/neumerance.mailgun.org/messages",
auth= ("api", "key-85jxyi84br32xu0cizfihmmbbxv576j5"),
data={
"from":fr,
"to": ["[email protected]",],
"subject": subj,
"text": msg
}
)
messages.info(request,
"Thank you for sending me your feedback, I will get back to you as soon as possible.",
extra_tags="success"
)
except:
messages.info(request,
"Unable to send your message at this time, Please try again later.",
extra_tags="error"
)
开发者ID:neumerance,项目名称:webservices,代码行数:26,代码来源:forms.py
示例11: test_save_option
def test_save_option(self):
data = {"Title": "title", "Toc": "false", "HeadingNumber": "i.a.a.i"}
# dir does not exists
r = requests.post(self.url("test") + "/test_dir_not_exists?option", data=json.dumps(data))
self.assertEqual(r.status_code, 400)
self.assertEqual(json.loads(r.content), {"code": 1})
# missing Title field
r = requests.post(self.url("test.md") + "?option", data=json.dumps({"Toc": "false", "HeadingNumber": "i.a.a.i"}))
self.assertEqual(r.status_code, 400)
self.assertEqual(json.loads(r.content), {"code": 1})
# Toc show be either `true` or `false`
r = requests.post(self.url("test.md") + "?option", data=json.dumps({"Title": "test", "Toc": "falsexxx", "HeadingNumber": "i.a.a.i"}))
self.assertEqual(r.status_code, 400)
self.assertEqual(json.loads(r.content), {"code": 1})
# HeadingNumber format shoule be `i.a.a.i`
r = requests.post(self.url("test.md") + "?option", data=json.dumps({"Title": "test", "Toc": "false", "HeadingNumber": "i.a.a.iiii"}))
self.assertEqual(r.status_code, 400)
self.assertEqual(json.loads(r.content), {"code": 1})
# succeeded
r = requests.post(self.url("test1xxx") + "?option", data=json.dumps({"Title": "test", "Toc": "false", "HeadingNumber": "i.a.a.i"}))
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers["Content-Type"], "application/json")
self.assertEqual(json.loads(r.content), {"code": 0})
self.assertEqual(json.loads(open(self.cwd + "/test1xxx.md.option.json", "r").read()), {"Title": "test", "Toc": "false", "HeadingNumber": "i.a.a.i", "Theme": "", "Host": ""})
开发者ID:Ibsenhasdream,项目名称:strapdown-zeta,代码行数:28,代码来源:test.py
示例12: analyze
def analyze(result_directory=None, upload=False, compute_cheapest=0, compute_average_per_day=False, compute_average_per_trip_length=False, compute_average_per_days_in_advance=False):
data, meta_data = read_results(result_directory)
result = []
result.append("Leaving from %s going to %s" % (meta_data['from'], meta_data['to']))
result.append("Size of date range: %s (days)" % meta_data['window_size'])
result.append("Minimum trip length: %s (days)" % meta_data['min_trip'])
result.append("Maximum trip length: %s (days)" % meta_data['max_trip'])
if compute_cheapest:
result.extend(get_cheapest(data, compute_cheapest))
if compute_average_per_day:
result.extend(get_average_per_day(data))
if compute_average_per_trip_length:
result.extend(get_average_per_trip_length(data))
if compute_average_per_days_in_advance:
result.extend(get_average_per_days_in_advance(data))
if upload:
upload_url = settings.DEBUG_UPLOAD_URL if settings.DEBUG else settings.UPLOAD_URL
requests.post(upload_url, data={'content': json.dumps(result)})
print "\n".join(result)
开发者ID:jpmunz,项目名称:flightscraper,代码行数:27,代码来源:analyze.py
示例13: verify_signature
def verify_signature(engine, signature, remote, payload, concurrency):
if concurrency > 1:
busy_process = subprocess.Popen([engine], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
busy_process.stdin.write('setoption name Threads value %d\n' % (concurrency-1))
busy_process.stdin.write('go infinite\n')
try:
bench_sig = ''
print 'Verifying signature of %s ...' % (os.path.basename(engine))
with open(os.devnull, 'wb') as f:
p = subprocess.Popen([engine, 'bench'], stderr=subprocess.PIPE, stdout=f, universal_newlines=True)
for line in iter(p.stderr.readline,''):
if 'Nodes searched' in line:
bench_sig = line.split(': ')[1].strip()
if 'Nodes/second' in line:
bench_nps = float(line.split(': ')[1].strip())
p.wait()
if p.returncode != 0:
raise Exception('Bench exited with non-zero code %d' % (p.returncode))
if int(bench_sig) != int(signature):
requests.post(remote + '/api/stop_run', data=json.dumps(payload))
raise Exception('Wrong bench in %s Expected: %s Got: %s' % (engine, signature, bench_sig))
finally:
if concurrency > 1:
busy_process.stdin.write('quit\n')
busy_process.kill()
return bench_nps
开发者ID:4616,项目名称:fishtest,代码行数:31,代码来源:games.py
示例14: setUp
def setUp(self):
"""
# Create a db
"""
headers = {'Content-Type': 'application/json; charset=utf-8'}
db_data = {'name': 'testDB'}
response = requests.post(__db_url__, json=db_data, headers=headers)
if response.status_code == 201:
self.assertEqual(response.status_code, 201)
else:
self.assertEqual(response.status_code, 404)
# Create a server
response = requests.get(__db_url__)
value = response.json()
if value:
db_length = len(value['databases'])
last_db_id = value['databases'][db_length-1]['id']
url = 'http://%s:8000/api/1.0/databases/%u/servers/' % \
(__host_or_ip__,last_db_id)
data = {'description': 'test', 'hostname': __host_or_ip__, 'name': 'test'}
response = requests.post(url, json=data, headers=headers)
if response.status_code == 201:
self.assertEqual(response.status_code, 201)
else:
self.assertEqual(response.status_code, 404)
else:
print "The database list is empty"
开发者ID:jango2015,项目名称:voltdb,代码行数:29,代码来源:start_stop_database_test.py
示例15: getTrackIDs
def getTrackIDs(performer):
r = requests.post('http://muzis.ru/api/search.api', data = {'q_performer':performer})
artId = json.loads(r.text)["performers"][0]["id"]
r = requests.post('http://muzis.ru/api/get_songs_by_performer.api', data = {'performer_id':artId,'type':'3'})
listOfSongs = json.loads(r.text)['songs']
#a = [x["file_mp3"] for x in listOfSongs]
return(listOfSongs[:5])
开发者ID:howardthefcuk,项目名称:muzisFiveLakes,代码行数:7,代码来源:bot.py
示例16: send_data
def send_data(self, eventdata, spooledevent=False):
# only send if there is data to be sent
success = None
if eventdata:
tries = 0
r = None
# If at first you don't succeed, try again. And again, and again, and again, and again.
# Server must return 200 OK or we will assume the delivery failed.
# while tries <= 5:
try:
headers = {'User-Agent': self.useragent,
'Content-Type': 'application/json'}
url = self.send_to
if self.key:
r = requests.post(url, headers=headers, data=json.dumps(eventdata),
verify=self.verify, cert=(self.cert, self.key))
elif url.startswith('https'):
r = requests.post(url, headers=headers, data=json.dumps(eventdata), verify=self.verify)
else:
r = requests.post(url, headers=headers, data=json.dumps(eventdata))
success = r.status_code
if not r.status_code == 200:
if not spooledevent:
spooler.spool_data(json.dumps(eventdata))
errorstr = "Warning: server unable to accept events. Spooling events to file."
print >> sys.stderr, errorstr
else:
self.write_last_event(eventdata)
except ConnectionError as e:
if not spooledevent:
spooler.spool_data(json.dumps(eventdata))
errorstr = "Warning: could not reach {}.".format(self.send_to)
print >> sys.stderr, errorstr
return success
开发者ID:scherma,项目名称:sneeze,代码行数:34,代码来源:hayfever.py
示例17: bale_CRITs_indicator
def bale_CRITs_indicator(base_url, data, indicator_que):
""" One thread of adding indicators to CRITs"""
while not indicator_que.empty():
indicator = indicator_que.get()
if indicator[1] == 'IPv4':
# using the IP API
url = base_url + 'ips/'
data['add_indicator'] = "true"
data['ip'] = indicator[0]
data['ip_type'] = 'Address - ipv4-addr'
data['reference'] = indicator[3]
# getting the source automatically:
source = re.findall(r'\/\/(.*?)\/', data['reference'])
if source:
data['source'] = source[0]
res = requests.post(url, data=data, verify=False)
if not res.status_code in [201, 200, 400]:
logger.info("Issues with adding: %s" % data['ip'])
elif indicator[1] == "FQDN":
# using the Domain API
url = base_url + 'domains/'
data['add_indicator'] = "true"
data['domain'] = indicator[0]
data['reference'] = indicator[3]
# getting the source automatically:
source = re.findall(r'\/\/(.*?)\/', data['reference'])
if source:
data['source'] = source[0]
res = requests.post(url, data=data, verify=False)
if not res.status_code in [201, 200, 400]:
logger.info("Issues with adding: %s" % data['domain'])
else:
logger.info("don't yet know what to do with: %s[%s]" % (indicator[1], indicator[0]))
开发者ID:Darkan,项目名称:combine,代码行数:33,代码来源:baler.py
示例18: postfile
def postfile(host, port, user, auth_token, db, table, filename, lines, version):
"""
post data reading from file to server
"""
if not os.path.isfile(filename):
click.echo('[ERROR] Filename not exist')
return
headers = {'X-USERNAME':user, 'X-AUTH-TOKEN':auth_token}
url = "http://%s:%s/%s/%s/%s" % (host, port, version, db, table)
total_lines = wc['-l', filename]().split()[0]
over_lines = 0
post_data = []
for data in open(filename):
js = json.loads(data)
js['tags']= js['tags'][0] if len(js['tags']) else ''
post_data.append(js)
if len(post_data) == lines:
over_lines = over_lines + lines
r = requests.post(url, data=json.dumps(post_data), headers=headers)
if r.status_code != 200:
click.echo("Invalid response from server")
return
click.echo('Post %s/%s lines data to ingestion Server' % (over_lines, total_lines))
post_data = []
if len(post_data) > 0:
over_lines = over_lines + len(post_data)
requests.post(url, data=json.dumps(post_data), headers=headers)
click.echo('Post %s/%s lines data to ingestion Server' % (over_lines,total_lines))
click.echo("All data post to ingestion server")
开发者ID:pmars,项目名称:tools,代码行数:33,代码来源:ingestion_client.py
示例19: _authenticate
def _authenticate(self, response=None, headers=None):
if self._authn_type is None:
return headers
url = "%s://%s:%s%s" % (self._authn_protocol, self._authn_server, self._authn_port,
self._authn_url)
new_headers = headers or {}
try:
if self._ksinsecure:
response = requests.post(url, data=self._authn_body,
headers=self._DEFAULT_AUTHN_HEADERS, verify=False)
elif not self._ksinsecure and self._use_ks_certs:
response = requests.post(url, data=self._authn_body,
headers=self._DEFAULT_AUTHN_HEADERS, verify=self._kscertbundle)
else:
response = requests.post(url, data=self._authn_body,
headers=self._DEFAULT_AUTHN_HEADERS)
except Exception as e:
errmsg = 'Unable to connect to keystone for authentication. '
errmsg += 'Exception %s' %(e)
raise RuntimeError(errmsg)
if (response.status_code == 200) or (response.status_code == 201):
# plan is to re-issue original request with new token
if 'v2' in self._authn_url:
authn_content = json.loads(response.text)
self._auth_token = authn_content['access']['token']['id']
else:
self._auth_token = response.headers['x-subject-token']
new_headers['X-AUTH-TOKEN'] = self._auth_token
return new_headers
else:
raise RuntimeError('Authentication Failure')
开发者ID:JSpaceTeam,项目名称:contrail-controller,代码行数:32,代码来源:vnc_api.py
示例20: send_message
def send_message(data=None):
"""function for sending sms.
validates if user is valid > sends message > redirects to source path
Assumes URL of the form /smsapi?number=12341234&message=message&source=/url
"""
number = request.args.get('number')
message = request.args.get('message')
source = request.args.get('source', '/')
anonymous = request.args.get('anonymous')
if message is not None:
if len(message) > 141:
abort(400)
if number is None:
abort(400)
if(anonymous==None):
logMessage(number,message)
if Production:
requests.post(os.environ['BLOWERIO_URL'] + '/messages', data={'to': '+' + number, 'message': message})
else:
app.logger.info(str({'to': '+' + number, 'message': message}))
return redirect("."+source)
开发者ID:mkowoods,项目名称:TextThem,代码行数:29,代码来源:TextThem.py
注:本文中的requests.post函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论