def index(request):
form = UploadForm(request)
if form.validate_on_submit():
image = form.image.data
# NOTE: trusting user submitted file names here, the name should be
# sanitized in production.
uploaded_file = Path(request.app.config.UPLOAD_DIR) / image.name
uploaded_file.write_bytes(image.body)
description = form.description.data or 'no description'
session.setdefault('files', []).append((image.name, description))
return response.redirect('/')
img = '<section><img src="/img/{}"><p>{}</p><hr></section>'
images = ''.join(img.format(i, d) for i, d in session.get('files', []))
content = f"""
<h1>Sanic-WTF file field validators example</h1>
{images}
<form action="" method="POST" enctype="multipart/form-data">
{'<br>'.join(form.csrf_token.errors)}
{form.csrf_token}
{'<br>'.join(form.image.errors)}
{'<br>'.join(form.description.errors)}
<br> {form.image.label}
<br> {form.image}
<br> {form.description.label}
<br> {form.description(size=20, placeholder="description")}
<br> {form.submit}
</form>
"""
return response.html(content)
python类Sanic()的实例源码
def get_sanic_application():
"""
Sets up django and returns a Sanic application
"""
if sys.version_info < (3, 5):
raise RuntimeError("The SanicDjango Adaptor may only be used with python 3.5 and above.")
django.setup()
from django.conf import settings
DEBUG = getattr(settings, 'DEBUG', False)
INSTALLED_APPS = getattr(settings, 'INSTALLED_APPS', [])
do_static = DEBUG and 'django.contrib.staticfiles' in INSTALLED_APPS
app = Sanic(__name__)
if do_static:
static_url = getattr(settings, 'STATIC_URL', "/static/")
static_root = getattr(settings, 'STATIC_ROOT', "./static")
app.static(static_url, static_root)
app.handle_request = SanicHandler(app) # patch the app to use the django adaptor handler
return app
def test_query_string():
app = Sanic('test_query_string')
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app, params=[("test1", "1"), ("test2", "false"), ("test2", "true")])
assert request.args.get('test1') == '1'
assert request.args.get('test2') == 'false'
# ------------------------------------------------------------ #
# POST
# ------------------------------------------------------------ #
def test_post_form_multipart_form_data():
app = Sanic('test_post_form_multipart_form_data')
@app.route('/')
async def handler(request):
return text('OK')
payload = '------sanic\r\n' \
'Content-Disposition: form-data; name="test"\r\n' \
'\r\n' \
'OK\r\n' \
'------sanic--\r\n'
headers = {'content-type': 'multipart/form-data; boundary=----sanic'}
request, response = sanic_endpoint_test(app, data=payload, headers=headers)
assert request.form.get('test') == 'OK'
def test_middleware_request():
app = Sanic('test_middleware_request')
results = []
@app.middleware
async def handler(request):
results.append(request)
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app)
assert response.text == 'OK'
assert type(results[0]) is Request
def test_middleware_response():
app = Sanic('test_middleware_response')
results = []
@app.middleware('request')
async def process_response(request):
results.append(request)
@app.middleware('response')
async def process_response(request, response):
results.append(request)
results.append(response)
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app)
assert response.text == 'OK'
assert type(results[0]) is Request
assert type(results[1]) is Request
assert issubclass(type(results[2]), HTTPResponse)
def test_cookie_options():
app = Sanic('test_text')
@app.route('/')
def handler(request):
response = text("OK")
response.cookies['test'] = 'at you'
response.cookies['test']['httponly'] = True
response.cookies['test']['expires'] = datetime.now() + timedelta(seconds=10)
return response
request, response = sanic_endpoint_test(app)
response_cookies = SimpleCookie()
response_cookies.load(response.headers.get('Set-Cookie', {}))
assert response_cookies['test'].value == 'at you'
assert response_cookies['test']['httponly'] == True
def test_bad_request_response():
app = Sanic('test_bad_request_response')
lines = []
async def _request(sanic, loop):
connect = asyncio.open_connection('127.0.0.1', 42101)
reader, writer = await connect
writer.write(b'not http')
while True:
line = await reader.readline()
if not line:
break
lines.append(line)
app.stop()
app.run(host='127.0.0.1', port=42101, debug=False, after_start=_request)
assert lines[0] == b'HTTP/1.1 400 Bad Request\r\n'
assert lines[-1] == b'Error: Bad Request'
def test_storage():
app = Sanic('test_text')
@app.middleware('request')
def store(request):
request['user'] = 'sanic'
request['sidekick'] = 'tails'
del request['sidekick']
@app.route('/')
def handler(request):
return json({ 'user': request.get('user'), 'sidekick': request.get('sidekick') })
request, response = sanic_endpoint_test(app)
response_json = loads(response.text)
assert response_json['user'] == 'sanic'
assert response_json.get('sidekick') is None
def test_with_middleware():
app = Sanic('test_with_middleware')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
results = []
@app.middleware
async def handler(request):
results.append(request)
request, response = sanic_endpoint_test(app)
assert response.text == 'I am get method'
assert type(results[0]) is Request
def test_static_routes():
app = Sanic('test_dynamic_route')
@app.route('/test')
async def handler1(request):
return text('OK1')
@app.route('/pizazz')
async def handler2(request):
return text('OK2')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.text == 'OK1'
request, response = sanic_endpoint_test(app, uri='/pizazz')
assert response.text == 'OK2'
def test_dynamic_route_string():
app = Sanic('test_dynamic_route_string')
results = []
@app.route('/folder/<name:string>')
async def handler(request, name):
results.append(name)
return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
request, response = sanic_endpoint_test(app, uri='/folder/favicon.ico')
assert response.text == 'OK'
assert results[1] == 'favicon.ico'
def test_dynamic_route_int():
app = Sanic('test_dynamic_route_int')
results = []
@app.route('/folder/<folder_id:int>')
async def handler(request, folder_id):
results.append(folder_id)
return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/12345')
assert response.text == 'OK'
assert type(results[0]) is int
request, response = sanic_endpoint_test(app, uri='/folder/asdf')
assert response.status == 404
def test_dynamic_route_number():
app = Sanic('test_dynamic_route_number')
results = []
@app.route('/weight/<weight:number>')
async def handler(request, weight):
results.append(weight)
return text('OK')
request, response = sanic_endpoint_test(app, uri='/weight/12345')
assert response.text == 'OK'
assert type(results[0]) is float
request, response = sanic_endpoint_test(app, uri='/weight/1234.56')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/weight/1234-56')
assert response.status == 404
def test_dynamic_route_unhashable():
app = Sanic('test_dynamic_route_unhashable')
@app.route('/folder/<unhashable:[A-Za-z0-9/]+>/end/')
async def handler(request, unhashable):
return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/test/asdf/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test///////end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/nope/')
assert response.status == 404
def test_route_duplicate():
app = Sanic('test_route_duplicate')
with pytest.raises(RouteExists):
@app.route('/test')
async def handler1(request):
pass
@app.route('/test')
async def handler2(request):
pass
with pytest.raises(RouteExists):
@app.route('/test/<dynamic>/')
async def handler1(request, dynamic):
pass
@app.route('/test/<dynamic>/')
async def handler2(request, dynamic):
pass
def test_static_add_route():
app = Sanic('test_static_add_route')
async def handler1(request):
return text('OK1')
async def handler2(request):
return text('OK2')
app.add_route(handler1, '/test')
app.add_route(handler2, '/test2')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.text == 'OK1'
request, response = sanic_endpoint_test(app, uri='/test2')
assert response.text == 'OK2'
def test_dynamic_add_route_int():
app = Sanic('test_dynamic_add_route_int')
results = []
async def handler(request, folder_id):
results.append(folder_id)
return text('OK')
app.add_route(handler, '/folder/<folder_id:int>')
request, response = sanic_endpoint_test(app, uri='/folder/12345')
assert response.text == 'OK'
assert type(results[0]) is int
request, response = sanic_endpoint_test(app, uri='/folder/asdf')
assert response.status == 404
def test_dynamic_add_route_number():
app = Sanic('test_dynamic_add_route_number')
results = []
async def handler(request, weight):
results.append(weight)
return text('OK')
app.add_route(handler, '/weight/<weight:number>')
request, response = sanic_endpoint_test(app, uri='/weight/12345')
assert response.text == 'OK'
assert type(results[0]) is float
request, response = sanic_endpoint_test(app, uri='/weight/1234.56')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/weight/1234-56')
assert response.status == 404
def test_dynamic_add_route_regex():
app = Sanic('test_dynamic_route_int')
async def handler(request, folder_id):
return text('OK')
app.add_route(handler, '/folder/<folder_id:[A-Za-z0-9]{0,4}>')
request, response = sanic_endpoint_test(app, uri='/folder/test')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test1')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/test-123')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/')
assert response.status == 200
def test_dynamic_add_route_unhashable():
app = Sanic('test_dynamic_add_route_unhashable')
async def handler(request, unhashable):
return text('OK')
app.add_route(handler, '/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = sanic_endpoint_test(app, uri='/folder/test/asdf/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test///////end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/nope/')
assert response.status == 404
def test_add_route_duplicate():
app = Sanic('test_add_route_duplicate')
with pytest.raises(RouteExists):
async def handler1(request):
pass
async def handler2(request):
pass
app.add_route(handler1, '/test')
app.add_route(handler2, '/test')
with pytest.raises(RouteExists):
async def handler1(request, dynamic):
pass
async def handler2(request, dynamic):
pass
app.add_route(handler1, '/test/<dynamic>/')
app.add_route(handler2, '/test/<dynamic>/')
def test_several_bp_with_url_prefix():
app = Sanic('test_text')
bp = Blueprint('test_text', url_prefix='/test1')
bp2 = Blueprint('test_text2', url_prefix='/test2')
@bp.route('/')
def handler(request):
return text('Hello')
@bp2.route('/')
def handler2(request):
return text('Hello2')
app.blueprint(bp)
app.blueprint(bp2)
request, response = sanic_endpoint_test(app, uri='/test1/')
assert response.text == 'Hello'
request, response = sanic_endpoint_test(app, uri='/test2/')
assert response.text == 'Hello2'
def test_bp_middleware():
app = Sanic('test_middleware')
blueprint = Blueprint('test_middleware')
@blueprint.middleware('response')
async def process_response(request, response):
return text('OK')
@app.route('/')
async def handler(request):
return text('FAIL')
app.blueprint(blueprint)
request, response = sanic_endpoint_test(app)
assert response.status == 200
assert response.text == 'OK'
def test_list_default():
app = Sanic('test_get')
app.blueprint(openapi_blueprint)
@app.put('/test')
@doc.consumes(doc.List(int, description="All the numbers"), location="body")
def test(request):
return json({"test": True})
request, response = app.test_client.get('/openapi/spec.json')
response_schema = json_loads(response.body.decode())
parameter = response_schema['paths']['/test']['put']['parameters'][0]
assert response.status == 200
assert parameter['type'] == 'array'
assert parameter['items']['type'] == 'integer'
def __init__(self, name='default', testing=False):
"""Initialize Jawaf instance. Set up routes, database connections, and session.
:param name: String. Sanic instance name. (Default: 'default')
:param testing: Boolean. Whether or not testing framework is active.
"""
self.name = name
self.server = Sanic(name)
self.testing = testing
self._db_pools = {}
self._session_pool = None
self._smtp = None
global _active_instance
_active_instance = self
self.add_routes(routes_import=os.path.join(settings.PROJECT_DIR, 'routes.py'), base_path=settings.BASE_DIR)
if 'STATIC' in settings:
self.server.static(settings.STATIC['uri'], settings.STATIC['directory'])
self.init_databases()
self.init_session()
self.init_smtp()
self.init_apps()
def init_databases(self):
"""Initialize database connection pools from settings.py,
setting up Sanic blueprints for server start and stop."""
for database in settings.DATABASES:
db_blueprint = Blueprint(f'{self.name}_db_blueprint_{database}')
connection_settings = settings.DATABASES[database].copy()
if not connection_settings['user']:
connection_settings.pop('user')
connection_settings.pop('password')
@db_blueprint.listener('before_server_start')
async def setup_connection_pool(app, loop):
self._db_pools[database] = await create_pool(**connection_settings)
@db_blueprint.listener('after_server_stop')
async def close_connection_pool(app, loop):
if database in self._db_pools and self._db_pools[database]:
await self._db_pools[database].close()
self.server.blueprint(db_blueprint)
helper.py 文件源码
项目：python-tarantool-benchmark-and-bootstrap
作者: valentinmk
项目源码
文件源码
阅读 28
收藏 0
点赞 0
评论 0
def start(self):
"""Start sanic web server."""
self.app = Sanic('sanic_server')
# GZip support
# Compress(self.app)
# self.app.config['COMPRESS_MIMETYPES'] = {'text/html',
# 'application/json'}
# self.app.config['COMPRESS_LEVEL'] = 4
# self.app.config['COMPRESS_MIN_SIZE'] = 300
# Session support
self.session_interface = InMemorySessionInterface()
self.app.response_middleware.appendleft(self.save_session)
self.app.request_middleware.append(self.add_session_to_request)
self.add_routes()
return await self.app.create_server(loop=self.loop,
host='0.0.0.0',
port=self.port,
debug=False)
def test_query_string():
app = Sanic('test_query_string')
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app, params=[("test1", 1), ("test2", "false"), ("test2", "true")])
assert request.args.get('test1') == '1'
assert request.args.get('test2') == 'false'
# ------------------------------------------------------------ #
# POST
# ------------------------------------------------------------ #
def test_middleware_request():
app = Sanic('test_middleware_request')
results = []
@app.middleware
async def handler(request):
results.append(request)
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app)
assert response.text == 'OK'
assert type(results[0]) is Request
