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)
python类HTTP_OK的实例源码
def on_put(self, req, resp, *args, **kwargs):
"""
Updates a single record.
This should set all missing fields to default values, but we're not going to be so strict.
:param req: Falcon request
:type req: falcon.request.Request
:param resp: Falcon response
:type resp: falcon.response.Response
"""
obj = self.get_object(req, resp, kwargs, for_update=True)
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)
status_code = falcon.HTTP_OK
self.render_response(result, req, resp, status_code)
def test_parse_form_as_params(client):
class Resource:
def on_post(self, req, resp, **kwargs):
assert req.get_param('simple') == 'ok'
assert req.get_param('afile').file.read() == b'filecontent'
assert req.get_param('afile').filename == 'afile.txt'
resp.body = 'parsed'
resp.content_type = 'text/plain'
application.add_route('/route', Resource())
resp = client.post('/route', data={'simple': 'ok'},
files={'afile': ('filecontent', 'afile.txt')})
assert resp.status == falcon.HTTP_OK
assert resp.body == 'parsed'
def test_with_binary_file(client):
here = os.path.dirname(os.path.realpath(__file__))
filepath = os.path.join(here, 'image.jpg')
image = open(filepath, 'rb')
class Resource:
def on_post(self, req, resp, **kwargs):
resp.data = req.get_param('afile').file.read()
resp.content_type = 'image/jpg'
application.add_route('/route', Resource())
resp = client.post('/route', data={'simple': 'ok'},
files={'afile': image})
assert resp.status == falcon.HTTP_OK
image.seek(0)
assert resp.body == image.read()
def test_parse_non_ascii_filename_in_headers(client):
class Resource:
def on_post(self, req, resp, **kwargs):
assert req.get_param('afile').file.read() == b'name,code\nnom,2\n'
assert req.get_param('afile').filename == 'Na%C3%AFve%20file.txt'
resp.body = 'parsed'
resp.content_type = 'text/plain'
application.add_route('/route', Resource())
# Simulate browser sending non ascii filename.
body = ('--boundary\r\nContent-Disposition: '
'form-data; name="afile"; filename*=utf-8\'\'Na%C3%AFve%20file.txt'
'\r\nContent-Type: text/csv\r\n\r\nname,code\nnom,2\n\r\n'
'--boundary--\r\n')
headers = {'Content-Type': 'multipart/form-data; boundary=boundary'}
resp = client.post('/route', body=body, headers=headers)
assert resp.status == falcon.HTTP_OK
assert resp.body == 'parsed'
def test_list_images(client):
doc = {
'images': [
{
'href': '/images/1eaf6ef1-7f2d-4ecc-a8d5-6e8adba7cc0e.png'
}
]
}
response = client.simulate_get('/images')
result_doc = msgpack.unpackb(response.content, encoding='utf-8')
assert result_doc == doc
assert response.status == falcon.HTTP_OK
# With clever composition of fixtures, we can observe what happens with
# the mock injected into the image resource.
def test_should_report_healthy_if_kafka_healthy(self, kafka_check):
kafka_check.healthcheck.return_value = healthcheck.CheckResult(True,
'OK')
self.resource._kafka_check = kafka_check
ret = self.simulate_request(ENDPOINT,
headers={
'Content-Type': 'application/json'
},
decode='utf8',
method='GET')
self.assertEqual(falcon.HTTP_OK, self.srmock.status)
ret = json.loads(ret)
self.assertIn('kafka', ret)
self.assertEqual('OK', ret.get('kafka'))
def render_response(result, req, resp, status=falcon.HTTP_OK):
"""
:param result: Data to be returned in the response
:param req: Falcon request
:type req: falcon.request.Request
:param resp: Falcon response
:type resp: falcon.response.Response
:param status: HTTP status code
:type status: str
"""
resp.body = result
resp.status = status
def test_parse_multiple_values(client):
class Resource:
def on_post(self, req, resp, **kwargs):
assert req.get_param_as_list('multi') == ['1', '2']
resp.body = 'parsed'
resp.content_type = 'text/plain'
application.add_route('/route', Resource())
resp = client.post('/route', data={'multi': ['1', '2']},
files={'afile': ('filecontent', 'afile.txt')})
assert resp.status == falcon.HTTP_OK
assert resp.body == 'parsed'
def test_get_folder_tree(self, client):
'''Test we can serialize a directory tree structure
'''
# Get path within designated storage directory
fid = id_from_path(settings['MODEL_DATA_DIR'])
response = client.simulate_get('/files/tree/{}'.format(fid))
assert response.status == falcon.HTTP_OK
assert response.text == json.dumps(directory_tree_serializer(settings['MODEL_DATA_DIR']))
def on_get(self, req, resp, fid):
''' Return a JSON representation of the directory tree. The JSON response has the
following attributes:
- ``type``: file or folder
- ``name``: The base name of the file/folder
- ``path``: Absolute path to the object.
- ``id``: URL-safe base64 encoding of the ``path``
- ``children``: Only present if ``type`` is ``folder``. A list of all
children of this folder, each having the same representation.
Args:
fid (str): the base64 encoded url safe ID for the path to the root folder
Example::
http localhost:8000/files/tree/{fid}
'''
try:
self._data_store.validate_fid(fid)
resp.status = falcon.HTTP_OK
path = serializers.path_from_id(fid)
resp.body = json.dumps(serializers.directory_tree_serializer(path))
except PermissionError as error:
resp.status = falcon.HTTP_BAD_REQUEST
resp.body = str(error)
def on_get(self, req, resp, name):
'''Retrieve a JSON representation of a single ``Model`` based on its'
name.
Example::
http localhost:8000/models/{name}
'''
model = self.get_object(name)
resp.status = falcon.HTTP_OK
resp.body = model.to_json()
