我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sanic.Sanic()。
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)
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)
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_dynamic_route_number(): app = Sanic('test_dynamic_route_int') 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_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.register_blueprint(bp) app.register_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.register_blueprint(blueprint) request, response = sanic_endpoint_test(app) assert response.status == 200 assert response.text == 'OK'
def create_app(path='/graphql', **kwargs): app = Sanic(__name__) app.debug = True schema = kwargs.pop('schema', None) or Schema async_executor = kwargs.pop('async_executor', False) if async_executor: @app.listener('before_server_start') def init_async_executor(app, loop): executor = AsyncioExecutor(loop) app.add_route(GraphQLView.as_view(schema=schema, executor=executor, **kwargs), path) @app.listener('before_server_stop') def remove_graphql_endpoint(app, loop): app.remove_route(path) else: app.add_route(GraphQLView.as_view(schema=schema, **kwargs), path) app.client = SanicTestClient(app) return app
def _setup_sanic(self): from sanic import Sanic from sanic import response as sanic_response from sanic.response import redirect as sanic_redirect singletons.register('sanic_app', lambda: Sanic('omnic')) self.app = singletons.sanic_app self.response = sanic_response self.redirect = sanic_redirect # Set up all routes for all services for service in singletons.settings.load_all('SERVICES'): service_name = service.SERVICE_NAME for partial_url, view in service.urls.items(): if isinstance(view, str): view = getattr(service, view) url = '%s/%s' % (service_name, partial_url) self.app.add_route(view, url)
def __init__(self, ethpconnect='127.0.0.1', ethpport=9500, rpcconnect='127.0.0.1', rpcport=8545, ipcconnect=None, blocknotify=None, walletnotify=None, alertnotify=None, tls=False, *, loop=None): self._loop = loop or asyncio.get_event_loop() self._app = Sanic(__name__, log_config=None, error_handler=SentryErrorHandler()) self._host = ethpconnect self._port = ethpport self._rpc_host = rpcconnect self._rpc_port = rpcport self._unix_socket = ipcconnect self._blocknotify = blocknotify self._walletnotify = walletnotify self._alertnotify = alertnotify self._tls = tls self._log = logging.getLogger('rpc_server') self.routes()
def test_content_type(): app = Sanic('test_content_type') @app.route('/') async def handler(request): return text(request.content_type) request, response = app.test_client.get('/') assert request.content_type == DEFAULT_HTTP_CONTENT_TYPE assert response.text == DEFAULT_HTTP_CONTENT_TYPE headers = { 'content-type': 'application/json', } request, response = app.test_client.get('/', headers=headers) assert request.content_type == 'application/json' assert response.text == 'application/json'
def test_remote_addr(): app = Sanic('test_content_type') @app.route('/') async def handler(request): return text(request.remote_addr) headers = { 'X-Forwarded-For': '127.0.0.1, 127.0.1.2' } request, response = app.test_client.get('/', headers=headers) assert request.remote_addr == '127.0.0.1' assert response.text == '127.0.0.1' request, response = app.test_client.get('/') assert request.remote_addr == '' assert response.text == '' headers = { 'X-Forwarded-For': '127.0.0.1, , ,,127.0.1.2' } request, response = app.test_client.get('/', headers=headers) assert request.remote_addr == '127.0.0.1' assert response.text == '127.0.0.1'
def test_match_info(): app = Sanic('test_match_info') @app.route('/api/v1/user/<user_id>/') async def handler(request, user_id): return json(request.match_info) request, response = app.test_client.get('/api/v1/user/sanic_user/') assert request.match_info == {"user_id": "sanic_user"} assert json_loads(response.text) == {"user_id": "sanic_user"} # ------------------------------------------------------------ # # POST # ------------------------------------------------------------ #
def test_url_attributes_no_ssl(path, query, expected_url): app = Sanic('test_url_attrs_no_ssl') async def handler(request): return text('OK') app.add_route(handler, path) request, response = app.test_client.get(path + '?{}'.format(query)) assert request.url == expected_url.format(HOST, PORT) parsed = urlparse(request.url) assert parsed.scheme == request.scheme assert parsed.path == request.path assert parsed.query == request.query_string assert parsed.netloc == request.host
def test_url_attributes_with_ssl(path, query, expected_url): app = Sanic('test_url_attrs_with_ssl') current_dir = os.path.dirname(os.path.realpath(__file__)) context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH) context.load_cert_chain( os.path.join(current_dir, 'certs/selfsigned.cert'), keyfile=os.path.join(current_dir, 'certs/selfsigned.key')) async def handler(request): return text('OK') app.add_route(handler, path) request, response = app.test_client.get( 'https://{}:{}'.format(HOST, PORT) + path + '?{}'.format(query), server_kwargs={'ssl': context}) assert request.url == expected_url.format(HOST, PORT) parsed = urlparse(request.url) assert parsed.scheme == request.scheme assert parsed.path == request.path assert parsed.query == request.query_string assert parsed.netloc == request.host
def test_multiprocessing(): """Tests that the number of children we produce is correct""" # Selects a number at random so we can spot check num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1)) app = Sanic('test_multiprocessing') process_list = set() def stop_on_alarm(*args): for process in multiprocessing.active_children(): process_list.add(process.pid) process.terminate() signal.signal(signal.SIGALRM, stop_on_alarm) signal.alarm(1) app.run(HOST, PORT, workers=num_workers) assert len(process_list) == num_workers
def test_static_content_range_correct(file_name, static_file_directory): app = Sanic('test_static') app.static( '/testing.file', get_file_path(static_file_directory, file_name), use_content_range=True) headers = { 'Range': 'bytes=12-19' } request, response = app.test_client.get('/testing.file', headers=headers) assert response.status == 200 assert 'Content-Length' in response.headers assert 'Content-Range' in response.headers static_content = bytes(get_file_content( static_file_directory, file_name))[12:19] assert int(response.headers[ 'Content-Length']) == len(static_content) assert response.body == static_content
def test_static_content_range_front(file_name, static_file_directory): app = Sanic('test_static') app.static( '/testing.file', get_file_path(static_file_directory, file_name), use_content_range=True) headers = { 'Range': 'bytes=12-' } request, response = app.test_client.get('/testing.file', headers=headers) assert response.status == 200 assert 'Content-Length' in response.headers assert 'Content-Range' in response.headers static_content = bytes(get_file_content( static_file_directory, file_name))[12:] assert int(response.headers[ 'Content-Length']) == len(static_content) assert response.body == static_content
def test_static_content_range_back(file_name, static_file_directory): app = Sanic('test_static') app.static( '/testing.file', get_file_path(static_file_directory, file_name), use_content_range=True) headers = { 'Range': 'bytes=-12' } request, response = app.test_client.get('/testing.file', headers=headers) assert response.status == 200 assert 'Content-Length' in response.headers assert 'Content-Range' in response.headers static_content = bytes(get_file_content( static_file_directory, file_name))[-12:] assert int(response.headers[ 'Content-Length']) == len(static_content) assert response.body == static_content
def test_method_not_allowed(): app = Sanic('method_not_allowed') @app.get('/') async def test(request): return response.json({'hello': 'world'}) request, response = app.test_client.head('/') assert response.headers['Allow']== 'GET' @app.post('/') async def test(request): return response.json({'hello': 'world'}) request, response = app.test_client.head('/') assert response.status == 405 assert set(response.headers['Allow'].split(', ')) == set(['GET', 'POST']) assert response.headers['Content-Length'] == '0'
def test_file_head_response(file_name, static_file_directory): app = Sanic('test_file_helper') @app.route('/files/<filename>', methods=['GET', 'HEAD']) async def file_route(request, filename): file_path = os.path.join(static_file_directory, filename) file_path = os.path.abspath(unquote(file_path)) stats = await async_os.stat(file_path) headers = dict() headers['Accept-Ranges'] = 'bytes' headers['Content-Length'] = str(stats.st_size) if request.method == "HEAD": return HTTPResponse( headers=headers, content_type=guess_type(file_path)[0] or 'text/plain') else: return file(file_path, headers=headers, mime_type=guess_type(file_path)[0] or 'text/plain') request, response = app.test_client.head('/files/{}'.format(file_name)) assert response.status == 200 assert 'Accept-Ranges' in response.headers assert 'Content-Length' in response.headers assert int(response.headers[ 'Content-Length']) == len( get_file_content(static_file_directory, file_name))
def test_unauthorized_exception(exception_app): """Test the built-in Unauthorized exception""" request, response = exception_app.test_client.get('/401') assert response.status == 401 request, response = exception_app.test_client.get('/401/basic') assert response.status == 401 assert response.headers.get('WWW-Authenticate') is not None assert response.headers.get('WWW-Authenticate') == "Basic realm='Sanic'" request, response = exception_app.test_client.get('/401/digest') assert response.status == 401 auth_header = response.headers.get('WWW-Authenticate') assert auth_header is not None assert auth_header.startswith('Digest') assert "qop='auth, auth-int'" in auth_header assert "algorithm='MD5'" in auth_header assert "nonce='abcdef'" in auth_header assert "opaque='zyxwvu'" in auth_header request, response = exception_app.test_client.get('/401/bearer') assert response.status == 401 assert response.headers.get('WWW-Authenticate') == "Bearer"
