Python falcon 模块,HTTPConflict() 实例源码
我们从Python开源项目中,提取了以下19个代码示例,用于说明如何使用falcon.HTTPConflict()。
def on_post(self, request, response):
query = dict()
try:
raw_json = request.stream.read()
except Exception as e:
raise falcon.HTTPError(falcon.HTTP_400, 'Error', e.message)
try:
data = json.loads(raw_json, encoding='utf-8')
except ValueError:
raise falcon.HTTPError(falcon.HTTP_400, 'Malformed JSON')
if "id" not in data:
raise falcon.HTTPConflict('Task creation', "ID is not specified.")
if "type" not in data:
raise falcon.HTTPConflict('Task creation', "Type is not specified.")
transaction = self.client.push_task({ "task" : "vertex", "data" : data })
response.body = json.dumps({ "transaction" : transaction })
response.status = falcon.HTTP_202
def on_post(self, req, resp, *args, **kwargs):
data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
data, errors = self.clean(data)
if errors:
result = {'errors': errors}
status_code = falcon.HTTP_BAD_REQUEST
self.render_response(result, req, resp, status_code)
return
try:
with self.session_scope(self.db_engine) as db_session:
result = self.create(req, resp, data, db_session=db_session)
except IntegrityError:
raise HTTPConflict('Conflict', 'Unique constraint violated')
except ProgrammingError as err:
# Cases such as unallowed NULL value should have been checked before we got here (e.g. validate against
# schema using falconjsonio) - therefore assume this is a UNIQUE constraint violation
if len(err.orig.args) > 1 and err.orig.args[1] == self.VIOLATION_UNIQUE:
raise HTTPConflict('Conflict', 'Unique constraint violated')
raise
status_code = falcon.HTTP_CREATED
self.render_response(result, req, resp, status_code)
def delete(self, req, resp, obj, db_session=None):
"""
Delete an existing record.
:param req: Falcon request
:type req: falcon.request.Request
:param resp: Falcon response
:type resp: falcon.response.Response
:param obj: the object to delete
:param db_session: SQLAlchemy session
:type db_session: sqlalchemy.orm.session.Session
"""
deleted = db_session.delete(obj)
if deleted == 0:
raise falcon.HTTPConflict('Conflict', 'Resource found but conditions violated')
def on_put(self, req, resp, *args, **kwargs):
status_code = falcon.HTTP_OK
try:
with self.session_scope(self.db_engine) as db_session:
obj = self.get_object(req, resp, kwargs, for_update=True, db_session=db_session)
data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
data, errors = self.clean(data)
if errors:
result = {'errors': errors}
status_code = falcon.HTTP_BAD_REQUEST
else:
result = self.update(req, resp, data, obj, db_session)
except (IntegrityError, ProgrammingError) as err:
# Cases such as unallowed NULL value should have been checked before we got here (e.g. validate against
# schema using falconjsonio) - therefore assume this is a UNIQUE constraint violation
if isinstance(err, IntegrityError) or err.orig.args[1] == self.VIOLATION_FOREIGN_KEY:
raise HTTPConflict('Conflict', 'Unique constraint violated')
else:
raise
self.render_response(result, req, resp, status_code)
def _retrieve_layering_policy(self):
try:
# NOTE(fmontei): Layering policies exist system-wide, across all
# revisions, so no need to filter by revision.
layering_policy_filters = {
'deleted': False,
'schema': types.LAYERING_POLICY_SCHEMA
}
layering_policy = db_api.document_get(**layering_policy_filters)
except errors.DocumentNotFound as e:
error_msg = (
'No layering policy found in the system so could not render '
'the documents.')
LOG.error(error_msg)
LOG.exception(six.text_type(e))
raise falcon.HTTPConflict(description=error_msg)
else:
return layering_policy
def dataset_untrained_status(req, resp, resource, params):
"""Raises an error if dataset is not on an untrained state
Must be executed after check_dataset_exsistence. This will not inform
about dataset existence, instead will return an undefined error.
If query param ignore_status is true, it will not raise any error
"""
status, dataset_dto = _get_dataset_status(params['dataset_id'])
ignore_status = req.get_param_as_bool("ignore_status")
# Dataset is trained if 0b0010 bit is on
if status & 0b0010 != 0 and not ignore_status:
raise falcon.HTTPConflict(
title="The dataset is not in a correct state",
description=("The dataset {id} has an status {status}, which "
"is not valid to insert triples. Required is 0 "
).format(**dataset_dto.to_dict()))
def dataset_trained_status(req, resp, resource, params):
"""Raises an error if dataset is not on an trained state
Must be executed after check_dataset_exsistence. This will not inform
about dataset existence, instead will return an undefined error.
If query param ignore_status is true, it will not raise any error
"""
status, dataset_dto = _get_dataset_status(params['dataset_id'])
ignore_status = req.get_param_as_bool("ignore_status")
# Trained status is 0b0010.
# The status must be pair for the task to be finished
if status & 0b0010 != 0 and status & 0b0001 != 0 and not ignore_status:
raise falcon.HTTPConflict(
title="The dataset is not in a correct state",
description=("The dataset {id} has an status {status}, which "
"is not valid to generate an index. Required is 1 "
).format(**dataset_dto.to_dict()))
def delete(self, req, resp, obj):
deleted = obj.delete(using=self.connection)
if deleted == 0:
raise HTTPConflict('Conflict', 'Resource found but conditions violated')
def on_delete(self, req, resp, *args, **kwargs):
try:
with self.session_scope(self.db_engine) as db_session:
obj = self.get_object(req, resp, kwargs, for_update=True, db_session=db_session)
self.delete(req, resp, obj, db_session)
except (IntegrityError, ProgrammingError) as err:
# This should only be caused by foreign key constraint being violated
if isinstance(err, IntegrityError) or err.orig.args[1] == self.VIOLATION_FOREIGN_KEY:
raise HTTPConflict('Conflict', 'Other content links to this')
else:
raise
self.render_response({}, req, resp)
def _create_revision_documents(self, bucket_name, documents,
validations):
try:
created_documents = db_api.documents_create(
bucket_name, documents, validations=validations)
except (deckhand_errors.DocumentExists,
deckhand_errors.SingletonDocumentConflict) as e:
raise falcon.HTTPConflict(description=e.format_message())
except Exception as e:
raise falcon.HTTPInternalServerError(description=six.text_type(e))
return created_documents
def test_misc(self, client):
self._misc_test(client, falcon.HTTPBadRequest, falcon.HTTP_400)
self._misc_test(client, falcon.HTTPNotAcceptable, falcon.HTTP_406,
needs_title=False)
self._misc_test(client, falcon.HTTPConflict, falcon.HTTP_409)
self._misc_test(client, falcon.HTTPPreconditionFailed, falcon.HTTP_412)
self._misc_test(client, falcon.HTTPUnsupportedMediaType, falcon.HTTP_415,
needs_title=False)
self._misc_test(client, falcon.HTTPUnprocessableEntity, falcon.HTTP_422)
self._misc_test(client, falcon.HTTPUnavailableForLegalReasons, falcon.HTTP_451,
needs_title=False)
self._misc_test(client, falcon.HTTPInternalServerError, falcon.HTTP_500)
self._misc_test(client, falcon.HTTPBadGateway, falcon.HTTP_502)
def on_post(self, request, response, vertex_id):
query = dict()
try:
raw_json = request.stream.read()
except Exception as e:
raise falcon.HTTPError(falcon.HTTP_400, 'Error', e.message)
try:
data = json.loads(raw_json, encoding='utf-8')
except ValueError:
raise falcon.HTTPError(falcon.HTTP_400, 'Malformed JSON')
data["id"] = vertex_id
try:
query = list(graph.query_vertices({ "id" : vertex_id }))
except Exception as e:
raise falcon.HTTPError(falcon.HTTP_400, 'Error', e.message)
if len(query) > 0:
raise falcon.HTTPConflict('Vertex Creation', "Vertex already exists.")
try:
result = graph.update_vertex(**data)
except Exception as e:
raise falcon.HTTPError(falcon.HTTP_400, 'Error', e.message)
response.status = falcon.HTTP_200
response.body = json.dumps({ "created" : result }, encoding='utf-8')
def on_post(self, req, resp, dataset_id, dataset_dto, entities_pair):
"""This method return the true distance between two entities
{"distance":
["http://www.wikidata.org/entity/Q1492",
"http://www.wikidata.org/entity/Q2807"]
}
:param int dataset_id: The dataset identifier on database
:param DTO dataset_dto: The Dataset DTO from dataset_id (from hook)
:param tuple entities_pair: A pair of entities (from hook)
:returns: A distance attribute, float number
:rtype: dict
"""
dataset_dao = data_access.DatasetDAO()
dataset = dataset_dao.build_dataset_object(dataset_dto) # TODO: design
# Get server to do 'queries'
search_index, err = dataset_dao.get_search_index(dataset_dto)
if search_index is None:
msg_title = "Dataset not ready perform search operation"
raise falcon.HTTPConflict(title=msg_title, description=str(err))
# TODO: Maybe extract server management anywhere to simplify this
search_server = server.Server(search_index)
entity_x, entity_y = entities_pair
id_x = dataset.get_entity_id(entity_x)
id_y = dataset.get_entity_id(entity_y)
if id_x is None or id_y is None:
raise falcon.HTTPNotFound(
description=("The {} id from entity {} or the {} id from {} "
"entity can't be found on the dataset")
.format(id_x, entity_x, id_y, entity_y))
dist = search_server.distance_between_entities(id_x, id_y)
resp.body = json.dumps({"distance": dist})
resp.content_type = 'application/json'
resp.status = falcon.HTTP_200
def on_post(self, req, resp, dataset_info, **kwargs):
"""Create a new dataset on the service
This method will create a new empty dataset, and returns a 201 CREATED
with Location header filled with the URI of new dataset.
:param HTTPUserDatasetDTO dataset_info: HTTP Client dataset information
:query int dataset_type: The dataset type (optional)
:returns: Location header with new path to dataset object
"""
dao = data_access.DatasetDAO()
# Get dataset type
dts_type = req.get_param_as_int("dataset_type")
dataset_type = dao.get_dataset_types()[dts_type]["class"]
id_dts, err = dao.insert_empty_dataset(
dataset_type, name=dataset_info.name,
description=dataset_info.description)
if id_dts is None and err[0] == 409:
raise falcon.HTTPConflict(
title="The dataset name is already used", description=err[1])
elif id_dts is None and err[0] == 500:
raise falcon.HTTPInternalServerError(description=err[1])
else:
# Dataset created, evrything is done
resp.status = falcon.HTTP_201
resp.body = json.dumps({"dataset": {"id": id_dts}})
resp.location = "/datasets/" + str(id_dts)
def on_post(self, req, resp, dataset_id, dataset_dto, entities):
"""Get the embedding given an entity or a list of entities (URI)
{"entities": ["Q1492", "Q2807", "Q1"]}
:param integer dataset_id: Unique ID of dataset
:param integer dataset_dto: Dataset DTO (from hook)
:param list entities: List of entities to get embeddings (from hook)
:returns: A list of list with entities and its embeddings
:rtype: list
"""
istrained = dataset_dto.is_trained()
if istrained is None or not istrained:
raise falcon.HTTPConflict(
title="Dataset has not a valid state",
description="Dataset {} has a {} state".format(
dataset_id, dataset_dto.status))
try:
result = async_tasks.find_embeddings_on_model(dataset_id, entities)
except OSError as err:
filerr = err.filename
raise falcon.HTTPNotFound(
title="The file on database couldn't be located",
description=("A file ({}) has been found on database, but it "
"does not exist on filesystem").format(filerr))
textbody = {"embeddings": result}
resp.body = json.dumps(textbody)
resp.status = falcon.HTTP_200
def on_post(self, req: Request, res: Response):
with self.make_session() as session:
try:
item_id = self.post_item(req.context["doc"], session)
put_json_to_context(res, {"created": item_id})
except IntegrityError as e:
raise falcon.HTTPConflict("Conflict", str(e))
def on_put(self, req: Request, res: Response, item_id):
with self.make_session() as session:
try:
self.put_item(item_id, req.context["doc"], session)
put_json_to_context(res, {"created": item_id})
except IntegrityError as e:
raise falcon.HTTPConflict("Conflict", str(e))
def on_patch(self, req: Request, res: Response, item_id):
with self.make_session() as session:
try:
ok = self.update_item(item_id, req.context["doc"], session)
if not ok:
raise falcon.HTTPNotFound()
except IntegrityError as e:
raise falcon.HTTPConflict("Conflict", str(e))