Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def page_identity(self, response, request_json=None):
"""Takes a `requests.Response` and
returns a new __item_class__ instance if the request method is not a get, or returns
a __class__ instance if the request path is different than the caller's `endpoint`.
"""
request_path = response.request.path_url
if request_path == '/migrations_notran/':
raise exc.IsMigrating('You have been redirected to the migration-in-progress page.')
request_method = response.request.method.lower()
self.last_elapsed = response.elapsed
if isinstance(request_json, dict) and 'ds' in request_json:
ds = request_json.ds
else:
ds = None
try:
data = response.json()
except ValueError as e: # If there was no json to parse
data = dict()
if response.text or response.status_code not in (200, 202, 204):
text = response.text
if len(text) > 1024:
def cancel(self):
cancel = self.get_related('cancel')
if not cancel.can_cancel:
return
try:
cancel.post()
except exc.MethodNotAllowed as e:
# Race condition where job finishes between can_cancel
# check and post.
if not any("not allowed" in field for field in e.msg.values()):
raise(e)
return self.get()
json = self.json if json is None else json
r = self.connection.put(self.endpoint, json=json)
return self.page_identity(r, request_json=json)
def get_related(self, related_name, **kwargs):
assert related_name in self.json.get('related', [])
endpoint = self.json['related'][related_name]
return self.walk(endpoint, **kwargs)
def walk(self, endpoint, **kw):
page_cls = get_registered_page(endpoint)
return page_cls(self.connection, endpoint=endpoint).get(**kw)
_exception_map = {http.NO_CONTENT: exc.NoContent,
http.NOT_FOUND: exc.NotFound,
http.INTERNAL_SERVER_ERROR: exc.InternalServerError,
http.BAD_GATEWAY: exc.BadGateway,
http.METHOD_NOT_ALLOWED: exc.MethodNotAllowed,
http.UNAUTHORIZED: exc.Unauthorized,
http.PAYMENT_REQUIRED: exc.PaymentRequired,
http.CONFLICT: exc.Conflict}
def exception_from_status_code(status_code):
return _exception_map.get(status_code, None)
class PageList(object):
@property
def __item_class__(self):
return self.page_identity(r, request_json=json)
def get_related(self, related_name, **kwargs):
assert related_name in self.json.get('related', [])
endpoint = self.json['related'][related_name]
return self.walk(endpoint, **kwargs)
def walk(self, endpoint, **kw):
page_cls = get_registered_page(endpoint)
return page_cls(self.connection, endpoint=endpoint).get(**kw)
_exception_map = {http.NO_CONTENT: exc.NoContent,
http.NOT_FOUND: exc.NotFound,
http.INTERNAL_SERVER_ERROR: exc.InternalServerError,
http.BAD_GATEWAY: exc.BadGateway,
http.METHOD_NOT_ALLOWED: exc.MethodNotAllowed,
http.UNAUTHORIZED: exc.Unauthorized,
http.PAYMENT_REQUIRED: exc.PaymentRequired,
http.CONFLICT: exc.Conflict}
def exception_from_status_code(status_code):
return _exception_map.get(status_code, None)
class PageList(object):
@property
def __item_class__(self):
"""Returns the class representing a single 'Page' item
With an inheritence of OrgListSubClass -> OrgList -> PageList -> Org -> Base -> Page, the following
def launch(self, payload={}):
"""Launch using related->launch endpoint."""
# get related->launch
launch_pg = self.get_related('launch')
# launch the workflow_job_template
result = launch_pg.post(payload)
# return job
jobs_pg = self.related.workflow_jobs.get(id=result.workflow_job)
if jobs_pg.count != 1:
msg = "workflow_job_template launched (id:{}) but job not found in response at {}/workflow_jobs/" % \
(result.json['workflow_job'], self.url)
raise exc.UnexpectedAWXState(msg)
return jobs_pg.results[0]