commit
5b06bcc57d
|
@ -292,15 +292,22 @@ def html(body, status=200, headers=None):
|
||||||
content_type="text/html; charset=utf-8")
|
content_type="text/html; charset=utf-8")
|
||||||
|
|
||||||
|
|
||||||
async def file(location, mime_type=None, headers=None, _range=None):
|
async def file(
|
||||||
|
location, mime_type=None, headers=None, filename=None, _range=None):
|
||||||
"""Return a response object with file data.
|
"""Return a response object with file data.
|
||||||
|
|
||||||
:param location: Location of file on system.
|
:param location: Location of file on system.
|
||||||
:param mime_type: Specific mime_type.
|
:param mime_type: Specific mime_type.
|
||||||
:param headers: Custom Headers.
|
:param headers: Custom Headers.
|
||||||
|
:param filename: Override filename.
|
||||||
:param _range:
|
:param _range:
|
||||||
"""
|
"""
|
||||||
filename = path.split(location)[-1]
|
headers = headers or {}
|
||||||
|
if filename:
|
||||||
|
headers.setdefault(
|
||||||
|
'Content-Disposition',
|
||||||
|
'attachment; filename="{}"'.format(filename))
|
||||||
|
filename = filename or path.split(location)[-1]
|
||||||
|
|
||||||
async with open_async(location, mode='rb') as _file:
|
async with open_async(location, mode='rb') as _file:
|
||||||
if _range:
|
if _range:
|
||||||
|
@ -312,24 +319,30 @@ async def file(location, mime_type=None, headers=None, _range=None):
|
||||||
out_stream = await _file.read()
|
out_stream = await _file.read()
|
||||||
|
|
||||||
mime_type = mime_type or guess_type(filename)[0] or 'text/plain'
|
mime_type = mime_type or guess_type(filename)[0] or 'text/plain'
|
||||||
|
|
||||||
return HTTPResponse(status=200,
|
return HTTPResponse(status=200,
|
||||||
headers=headers,
|
headers=headers,
|
||||||
content_type=mime_type,
|
content_type=mime_type,
|
||||||
body_bytes=out_stream)
|
body_bytes=out_stream)
|
||||||
|
|
||||||
|
|
||||||
async def file_stream(location, chunk_size=4096, mime_type=None, headers=None,
|
async def file_stream(
|
||||||
_range=None):
|
location, chunk_size=4096, mime_type=None, headers=None,
|
||||||
|
filename=None, _range=None):
|
||||||
"""Return a streaming response object with file data.
|
"""Return a streaming response object with file data.
|
||||||
|
|
||||||
:param location: Location of file on system.
|
:param location: Location of file on system.
|
||||||
:param chunk_size: The size of each chunk in the stream (in bytes)
|
:param chunk_size: The size of each chunk in the stream (in bytes)
|
||||||
:param mime_type: Specific mime_type.
|
:param mime_type: Specific mime_type.
|
||||||
:param headers: Custom Headers.
|
:param headers: Custom Headers.
|
||||||
|
:param filename: Override filename.
|
||||||
:param _range:
|
:param _range:
|
||||||
"""
|
"""
|
||||||
filename = path.split(location)[-1]
|
headers = headers or {}
|
||||||
|
if filename:
|
||||||
|
headers.setdefault(
|
||||||
|
'Content-Disposition',
|
||||||
|
'attachment; filename="{}"'.format(filename))
|
||||||
|
filename = filename or path.split(location)[-1]
|
||||||
|
|
||||||
_file = await open_async(location, mode='rb')
|
_file = await open_async(location, mode='rb')
|
||||||
|
|
||||||
|
|
|
@ -149,7 +149,22 @@ def test_file_response(file_name, static_file_directory):
|
||||||
request, response = app.test_client.get('/files/{}'.format(file_name))
|
request, response = app.test_client.get('/files/{}'.format(file_name))
|
||||||
assert response.status == 200
|
assert response.status == 200
|
||||||
assert response.body == get_file_content(static_file_directory, file_name)
|
assert response.body == get_file_content(static_file_directory, file_name)
|
||||||
|
assert 'Content-Disposition' not in response.headers
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('source,dest', [
|
||||||
|
('test.file', 'my_file.txt'), ('decode me.txt', 'readme.md'), ('python.png', 'logo.png')])
|
||||||
|
def test_file_response_custom_filename(source, dest, static_file_directory):
|
||||||
|
app = Sanic('test_file_helper')
|
||||||
|
@app.route('/files/<filename>', methods=['GET'])
|
||||||
|
def file_route(request, filename):
|
||||||
|
file_path = os.path.join(static_file_directory, filename)
|
||||||
|
file_path = os.path.abspath(unquote(file_path))
|
||||||
|
return file(file_path, filename=dest)
|
||||||
|
|
||||||
|
request, response = app.test_client.get('/files/{}'.format(source))
|
||||||
|
assert response.status == 200
|
||||||
|
assert response.body == get_file_content(static_file_directory, source)
|
||||||
|
assert response.headers['Content-Disposition'] == 'attachment; filename="{}"'.format(dest)
|
||||||
|
|
||||||
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
|
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
|
||||||
def test_file_head_response(file_name, static_file_directory):
|
def test_file_head_response(file_name, static_file_directory):
|
||||||
|
@ -191,7 +206,22 @@ def test_file_stream_response(file_name, static_file_directory):
|
||||||
request, response = app.test_client.get('/files/{}'.format(file_name))
|
request, response = app.test_client.get('/files/{}'.format(file_name))
|
||||||
assert response.status == 200
|
assert response.status == 200
|
||||||
assert response.body == get_file_content(static_file_directory, file_name)
|
assert response.body == get_file_content(static_file_directory, file_name)
|
||||||
|
assert 'Content-Disposition' not in response.headers
|
||||||
|
|
||||||
|
@pytest.mark.parametrize('source,dest', [
|
||||||
|
('test.file', 'my_file.txt'), ('decode me.txt', 'readme.md'), ('python.png', 'logo.png')])
|
||||||
|
def test_file_stream_response_custom_filename(source, dest, static_file_directory):
|
||||||
|
app = Sanic('test_file_helper')
|
||||||
|
@app.route('/files/<filename>', methods=['GET'])
|
||||||
|
def file_route(request, filename):
|
||||||
|
file_path = os.path.join(static_file_directory, filename)
|
||||||
|
file_path = os.path.abspath(unquote(file_path))
|
||||||
|
return file_stream(file_path, chunk_size=32, filename=dest)
|
||||||
|
|
||||||
|
request, response = app.test_client.get('/files/{}'.format(source))
|
||||||
|
assert response.status == 200
|
||||||
|
assert response.body == get_file_content(static_file_directory, source)
|
||||||
|
assert response.headers['Content-Disposition'] == 'attachment; filename="{}"'.format(dest)
|
||||||
|
|
||||||
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
|
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
|
||||||
def test_file_stream_head_response(file_name, static_file_directory):
|
def test_file_stream_head_response(file_name, static_file_directory):
|
||||||
|
|
Loading…
Reference in New Issue
Block a user