本文整理汇总了Python中requests.patch函数的典型用法代码示例。如果您正苦于以下问题:Python patch函数的具体用法?Python patch怎么用?Python patch使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了patch函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: mark_test_status
def mark_test_status(self, test_case_name,status='skipped',test_run_name=None,test_log=None):
"""
This method will mark the executed step in a test run as Passed / Failed / Skipped in Test Lodge
provided test run name and the executed step title.
:param test_case_name: The title of the test case for which the status need to be marked in Test Lodge.
test_run_name(optional): The run name of the test run which contains the executed step.
if None, pylodge will assume the created test run as the test run that has the executed test.
status(optional): The execution status of the test. Passed / Failed / Skipped. The default is Skipped.
test_log(optional): It is possible to pass a runtime log in to this method.
If this argument is appropriately set, then pylodge will also insert the runtime log
as a comment for the executed step
"""
project_id =self.project_id
if test_run_name==None:
run_id = self.run_id
else:
run_id = self.fetch_test_run_id(test_run_name)
test_case_id = self.fetch_and_save_test_case_id_from_test_name(test_case_name)
if status.lower() == 'passed':
status_flag = 1
issue_tracker_flag=0
elif status.lower() == 'failed':
status_flag = 0
issue_tracker_flag=1
elif status.lower() == 'skipped':
status_flag = 2
issue_tracker_flag=0
requests.patch(
self._api_url + '/v1/projects/%s/runs/%s/executed_steps/%s.json' % (project_id, run_id, test_case_id),
json={
'executed_step': {'actual_result': 'Test Case %s and the log is \n %s'%(status,test_log),
'passed': status_flag, 'create_issue_tracker_ticket': issue_tracker_flag}},
auth=self._auth_tuple)
开发者ID:gettalent,项目名称:pylodge,代码行数:35,代码来源:pylodge.py
示例2: test_connection
def test_connection(cluster, log_write_url=None, girder_token=None):
cluster_id = cluster['_id']
cluster_url = '%s/clusters/%s' % (cumulus.config.girder.baseUrl, cluster_id)
log = get_cluster_logger(cluster, girder_token)
headers = {'Girder-Token': girder_token}
try:
# First fetch the cluster with this 'admin' token so we get the
# passphrase filled out.
r = requests.get(cluster_url, headers=headers)
check_status(r)
cluster = r.json()
with get_connection(girder_token, cluster) as conn:
status = 'running'
# Test can we can connect to cluster
output = conn.execute('pwd')
if len(output) < 1:
log.error('Unable connect to cluster')
status = 'error'
r = requests.patch(
cluster_url, headers=headers, json={'status': status})
check_status(r)
except Exception as ex:
r = requests.patch(cluster_url, headers=headers,
json={'status': 'error'})
# Log the error message
log.exception(ex)
开发者ID:mgrauer,项目名称:cumulus,代码行数:29,代码来源:cluster.py
示例3: add_reminder
def add_reminder(issue, headers, config, dryrun):
"""
Adds a reminder to the given issue.
:param issue: the issue to add a reminder to
:param headers: headers to use for requests against API
:param config: config to use
:param dryrun: whether to only simulate the writing API calls
"""
until = datetime.datetime.now() + datetime.timedelta(config["grace_period"])
personalized_reminder = config["reminder"].format(author=issue["author"], until=until.strftime("%Y-%m-%d %H:%M"))
# post a comment
logger.debug("-> Adding a reminder comment via POST %s" % issue["comments_url"])
if not dryrun:
requests.post(issue["comments_url"], headers=headers, data=json.dumps({"body": personalized_reminder}))
# label the issue if configured
if "label" in config and config["label"]:
current_labels = list(issue["labels"])
current_labels.append(config["label"])
logger.debug("-> Marking issues as invalid via PATCH %s, labels=%r" % (issue["url"], current_labels))
if not dryrun:
requests.patch(issue["url"], headers=headers, data=json.dumps({"labels": current_labels}))
开发者ID:buntobot,项目名称:GitIssueBot,代码行数:26,代码来源:approve.py
示例4: mark_issue_valid
def mark_issue_valid(issue, headers, config, dryrun):
"""
Marks a (formerly invalidated) issue as valid.
:param issue: the issue to mark as valid
:param headers: headers to use for requests against API
:param config: config to use
:param dryrun: whether to only simulate the writing API calls
"""
label = config.get("label", None)
oklabel = config.get("oklabel", None)
if not label and not oklabel:
return
current_labels = list(issue["labels"])
# apply the "incomplete ticket" label if configured
if label and label in current_labels:
current_labels.remove(label)
# apply the "ok ticket" label if configured and issue wouldn't be ignored otherwise
if oklabel and not oklabel in current_labels and not (has_ignored_labels(issue, config) or has_ignored_title(issue, config)):
current_labels.append(oklabel)
logger.debug("-> Marking issue valid via PATCH %s, labels=%r" % (issue["url"], current_labels))
if not dryrun:
requests.patch(issue["url"], headers=headers, data=json.dumps({"labels": current_labels}))
开发者ID:buntobot,项目名称:GitIssueBot,代码行数:29,代码来源:approve.py
示例5: update_prices
def update_prices(request):
l_limit = 0.02
treshhold = 0.04
for c in Company.objects.all():
incomplete_tasks = Task.objects.filter(company = c).exclude(id__in=[a.task.id for a in Answer.objects.all()])
timestamp = now() - timedelta(hours=1)
answers = Answer.objects.filter(timestamp__gte = timestamp)
answers = answers.filter(task__in = [tc for tc in Task.objects.filter(company=c)])
answers = answers.values('task').annotate(dcount=Count('task'))
answer_count = len(answers)
x = len(incomplete_tasks)/float(answer_count+1)
addition = treshhold * (1/(1+math.pow(math.e,(-x+5))))
new_price = addition+l_limit
price_json = '{"price":"%f"}' % (new_price)
header = {'Content-type': 'application/json'}
user = "aic_c1"
password = "aic"
count = 0
if incomplete_tasks:
print("Sending new price %.2f for tasks of company %s" % (new_price, c.name))
for t in incomplete_tasks:
url = "http://localhost:8000/api/tasks/" + str(t.crowdsourcing_id) + "/"
count += 1
sys.stdout.write("\r%d/%d" % (count, len(incomplete_tasks)))
sys.stdout.flush()
requests.patch(url,data=price_json,headers=header,auth=(user,password))
if incomplete_tasks:
print("")
return HttpResponse('{"text":"Prices successfully updated","success":true}', content_type="application/json")
开发者ID:Horrendus,项目名称:aic13,代码行数:35,代码来源:views.py
示例6: test_change_status
def test_change_status():
object_id = _create_object()
object_url = objects_url + '/' + object_id
requests.patch(object_url, json={'status': 'committed'})
r = requests.get(object_url)
assert r.status_code == 200
assert r.json()['status'] == 'committed'
开发者ID:akrause2014,项目名称:eudat,代码行数:7,代码来源:tests.py
示例7: run
def run(self):
while True:
req = self.outbound_queue.get()
if not req:
break
if req.request_type == 'GET':
params = json.dumps(req.params)
requests.get(req.url, params=params)
elif req.request_type == 'POST':
to_post = json.dumps(req.data)
params = json.dumps(req.params)
requests.post(req.url, params=params, data=to_post)
elif req.request_type == 'PUT':
to_put = json.dumps(req.data)
params = json.dumps(req.params)
requests.put(req.url, params=params, data=to_put)
elif req.request_type == 'PATCH':
to_patch = json.dumps(req.data)
params = json.dumps(req.params)
requests.patch(req.url, params=params, data=to_patch)
elif req.request_type == 'DELETE':
params = json.dumps(req.params)
requests.delete(req.url, params=params)
else:
print 'Invalid request type: %s' % req.request_type
开发者ID:arw180,项目名称:firebase-example,代码行数:25,代码来源:command_thread.py
示例8: apply_releasability
def apply_releasability(config, dest, endpoint, crits_id, json):
'''apply releasability markings to crits objects via api, return bool to indicate success'''
# Skip Crits objects that don't support releasability
endpoints_to_ignore = ['relationships']
if endpoint in endpoints_to_ignore:
return
url = crits_url(config, dest)
attempt_certificate_validation = \
config['crits']['sites'][dest]['api']['attempt_certificate_validation']
if not attempt_certificate_validation:
requests.packages.urllib3.disable_warnings()
data = {'source': config['crits']['sites'][dest]['api']['source']}
params = {'api_key': config['crits']['sites'][dest]['api']['key'],
'username': config['crits']['sites'][dest]['api']['user']}
data.update(json)
if config['crits']['sites'][dest]['api']['ssl']:
r = requests.patch(url + endpoint + '/' + crits_id + '/',
data=data,
params=params,
verify=attempt_certificate_validation)
else:
r = requests.patch(url + endpoint + '/' + crits_id + '/',
data=data,
params=params)
json_output = r.json()
result_code = json_output[u'return_code']
success = r.status_code in (200, 201) and result_code == 0
return(success)
开发者ID:Lambdanaut,项目名称:crits-adapter,代码行数:30,代码来源:crits.py
示例9: activate
def activate(self, channel):
domain = channel.org.get_brand_domain()
headers = {"Authorization": "Bearer %s" % channel.config[Channel.CONFIG_AUTH_TOKEN]}
# first set our callbacks
payload = {"webhooks": {"url": "https://" + domain + reverse("courier.wa", args=[channel.uuid, "receive"])}}
resp = requests.patch(
channel.config[Channel.CONFIG_BASE_URL] + "/v1/settings/application", json=payload, headers=headers
)
if resp.status_code != 200:
raise ValidationError(_("Unable to register callbacks: %s", resp.content))
# update our quotas so we can send at 15/s
payload = {
"messaging_api_rate_limit": ["15", "54600", "1000000"],
"contacts_scrape_rate_limit": "1000000",
"contacts_api_rate_limit": ["15", "54600", "1000000"],
}
resp = requests.patch(
channel.config[Channel.CONFIG_BASE_URL] + "/v1/settings/application", json=payload, headers=headers
)
if resp.status_code != 200:
raise ValidationError(_("Unable to configure channel: %s", resp.content))
开发者ID:mxabierto,项目名称:rapidpro,代码行数:25,代码来源:type.py
示例10: TestUser
def TestUser():
url = URL + "/users/me"
headers = {}
headers["Authorization"] = "Bearer " + access_token
values = {"state":"xxxx"}
data = json.dumps(values)
r = requests.patch(url, data = data, headers = headers)
print "set user state:", r.status_code
url = URL + "/users/me"
headers = {}
headers["Authorization"] = "Bearer " + access_token
values = {"name":"测试"}
data = json.dumps(values)
r = requests.patch(url, data = data, headers = headers)
print "set user name:", r.status_code
url = URL + "/users"
headers = {}
headers["Authorization"] = "Bearer " + access_token
obj = [{"zone":"86", "number":"13800000009", "name":"test9"},
{"zone":"86", "number":"13800000001", "name":"test1"}]
r = requests.post(url, data = json.dumps(obj), headers = headers)
print "upload contact list:", r.status_code
r = requests.get(url, headers = headers)
print "users:", r.text
开发者ID:richmonkey,项目名称:bauhinia_api,代码行数:30,代码来源:test_api.py
示例11: apply_label
def apply_label(label, issue, headers, dryrun=False):
current_labels = list(issue["labels"])
current_labels.append(label)
logger.debug("-> Adding a label via PATCH %s, labels=%r" % (issue["url"], current_labels))
if not dryrun:
requests.patch(issue["url"], headers=headers, data=json.dumps({"labels": current_labels}))
开发者ID:foosel,项目名称:GitIssueBot,代码行数:7,代码来源:autolabel.py
示例12: _exec_cmd_on_remote
def _exec_cmd_on_remote(self, cmd):
"""send HTTPS requests to **Github Issues** API
:param cmd: dict
:rtype: True or False
"""
success = False
url, payload = self._prepare_method_url_and_payload(cmd)
if cmd['CMD'] == 'ADD':
resp = requests.post(url=url,
data=json.dumps(payload),
auth=self.auth)
success = (resp.status_code == requests.codes.created)
elif cmd['CMD'] == 'EDIT':
resp = requests.patch(url=url,
data=json.dumps(payload),
auth=self.auth)
success = (resp.status_code == requests.codes.ok)
else: # REMOVE
resp = requests.patch(url=url,
data=json.dumps(payload),
auth=self.auth)
success = (resp.status_code == requests.codes.ok)
return success
开发者ID:keenhenry,项目名称:pda,代码行数:27,代码来源:ListDB.py
示例13: connect_socket
def connect_socket(self):
while(1):
try:
self.ws = create_connection(self.socket_url)
break
except:
logging.error('Sensa socket server connection error')
time.sleep(15)
logging.info('Connected to server')
# Set datastreams to listen from server
susc_ids = [ds['id'] for ds in self.datastream_suscriptions]
logging.debug('Suscribed to datastreams {}'.format(susc_ids))
msg = {
'device_id': self.device_id,
'datastream_suscriptions': susc_ids,
'api_token': self.api_token
}
self.ws.send(json.dumps(msg))
response = json.loads(self.ws.recv())
logging.debug('Connection response: {}'.format(response))
# Update status on web server
payload = json.dumps({'status': self.CONNECTED})
requests.patch(self.device_url, data=payload, headers=self.api_hdrs)
开发者ID:alealmuna,项目名称:sensa-client,代码行数:25,代码来源:sensa.py
示例14: save_user_data
def save_user_data(url, uid, name, balance, email):
p = {
"user[name]": name,
"user[email]": email,
"user[balance]": balance
}
requests.patch("{}/users/{}.json".format(url, uid), params=p)
开发者ID:YtvwlD,项目名称:qmleteroid,代码行数:7,代码来源:lib.py
示例15: test_invalid_status_transition
def test_invalid_status_transition(self):
task_url = '%srecipes/%s/tasks/%s/' % (self.get_proxy_url(),
self.recipe.id, self.recipe.tasks[0].id)
response = requests.patch(task_url, data=dict(status='Completed'))
self.assertEquals(response.status_code, 200)
response = requests.patch(task_url, data=dict(status='Running'))
self.assertEquals(response.status_code, 409)
开发者ID:ShaolongHu,项目名称:beaker,代码行数:7,代码来源:test_proxy.py
示例16: log_test_finish
def log_test_finish(self, nodeid: str, status: str, exception: str):
status = {'passed': 'COMPLETED', 'failed': 'FAILED'}[status]
requests.patch(f'{self._revizor_url}/api/tests/cases/{self._testcase_id}',
json={
'status': status,
'error_text': exception
}, headers={'Authorization': f'Token {self._token}'})
开发者ID:Scalr,项目名称:revizor-tests,代码行数:7,代码来源:surefire.py
示例17: test_start_and_complete_external_task
def test_start_and_complete_external_task(self):
# Alternative harnesses which support external tasks can report back
# the name and version of the task after they fetch it.
task_url = '%srecipes/%s/tasks/%s/' % (self.get_proxy_url(),
self.recipe.id, self.recipe.tasks[2].id)
name = '/external/example'
version = '3.14-1'
response = requests.patch(task_url, data=dict(status='Running',
name=name, version=version))
self.assertEquals(response.status_code, 200)
json = response.json()
self.assertEquals(json['status'], 'Running')
self.assertEquals(json['name'], name)
self.assertEquals(json['version'], version)
with session.begin():
session.expire_all()
task = self.recipe.tasks[2]
self.assertEquals(task.status, TaskStatus.running)
self.assertEquals(task.name, name)
self.assertEquals(task.version, version)
response = requests.patch(task_url, data=dict(status='Completed'))
self.assertEquals(response.status_code, 200)
self.assertEquals(response.json()['status'], 'Completed')
with session.begin():
session.expire_all()
task = self.recipe.tasks[2]
self.assertEquals(task.status, TaskStatus.completed)
开发者ID:ShaolongHu,项目名称:beaker,代码行数:27,代码来源:test_proxy.py
示例18: update_comment
def update_comment(request, comment_id):
headers=get_api_token(request)
headers['Content-type']='application/json'
data = dict()
url = "".join([API_BASE_LINK, 'comments', '/', comment_id, '/'])
if request.method == 'GET':
params = unquote(request.get_full_path()).split('?')[-1].split('&')
for param in params:
param=param.split('=')
if(len(param)<2):
return HttpResponse("Incorrect payload!!!")
data[param[0]]=param[1]
if request.method == 'POST':
method = request.POST.get('_method', '')
if str(method).upper() == 'DELETE':
requests.delete(url=url, headers=headers)
elif str(method).upper() == 'PATCH':
for key in request.POST:
data[key]=request.POST.get(key, '')
requests.patch(url=url, headers=headers, params=data).json()
else:
url = "".join([API_BASE_LINK, 'comments', '/'])
for key in request.POST:
data[key]=request.POST.get(key, '')
requests.post(url=url, headers=headers, params=data).json()
return redirect('/dashboard/')
开发者ID:Dikovinka,项目名称:HRCS,代码行数:26,代码来源:views.py
示例19: patch
def patch(**kwargs):
"""Summary
Args:
**kwargs: Description
Returns:
TYPE: Description
"""
path = kwargs.get('path')
url = kwargs.get('url')
client = kwargs.get('client')
oauth_key = kwargs.get('oauth_key', '')
fingerprint = kwargs.get('fingerprint')
payload = kwargs.get('payload')
headers = generate_headers(
client=client,
oauth_key=oauth_key,
fingerprint=fingerprint,
ip_address=kwargs.get('ip_address')
)
if path:
url = client.get_base_path() + path
response = requests.patch(url, headers=headers, data=json.dumps(payload))
else:
response = requests.patch(url, headers=headers, data=json.dumps(payload))
return determine_response(response=response)
开发者ID:sguron,项目名称:SynapsePay-Python,代码行数:27,代码来源:APIClient.py
示例20: listener
def listener(self, attr_name, value):
import requests
lat = str(value.global_relative_frame.lat)
lon = str(value.global_relative_frame.lon)
server ='http://m2gla-drone.istic.univ-rennes1.fr:8080'
url = server +'/intervention/' + args.idIntervention + '/drone'
data = '{ "battery": ' + str(vehicle.battery.current) +',"id":"'+ str(args.idDrone) +'","latitude": "' + lat + '","longitude": "' + lon + '"}'
headers = {"content-type": "application/json"}
#on met a jour la position du drone sur le serveur
requests.patch(url, data=data,headers=headers)
global nextPoint
import urllib,os
#si on est arrive a un nouveau point de la mission
if vehicle.commands.next != nextPoint:
filename = 'tmp/'+str(time.time())+'.png'
#recuperation de la limage de la carte a la position du drone
urllib.urlretrieve("https://maps.googleapis.com/maps/api/staticmap?center=" +lat +"," + lon + "&zoom=19&size=640x512&maptype=satellite&key=AIzaSyDMiGs7FfMIZANrYC6tBx6D-CFXMt0eY64&style=feature:road.local&scale=1",filename)
files = [('file', filename, open(filename, 'rb').read())]
data ={"idIntervention":str(args.idIntervention),"latitude":lat ,"longitude":lon}
url = server + '/photo/intervention/' + args.idIntervention
#on ajoute limage au serveur
post_multipart(url,data,files)
#suppression de limage
os.remove(filename)
nextPoint = vehicle.commands.next
开发者ID:tanguybernard,项目名称:drone-project,代码行数:32,代码来源:move_drone_boucle.py
注:本文中的requests.patch函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论