Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.send(text_data="Hello world!")
if django.VERSION >= (2, 0):
from django.urls import path
router = URLRouter(
[
path("basic-http/", BasicHttpConsumer),
path("basic-ws/", BasicWebsocketConsumer),
path("", AsgiHandler),
]
)
else:
from django.conf.urls import url
router = URLRouter(
[
url(r"^basic-http/$", BasicHttpConsumer),
url(r"^basic-ws/$", BasicWebsocketConsumer),
url(r"^$", AsgiHandler),
]
)
return AuthMiddlewareStack(router)
def test_path_remaining():
"""
Resolving continues in outer router if an inner router has no matching
routes
"""
inner_router = URLRouter([url(r"^no-match/$", MagicMock(return_value=1))])
test_app = MagicMock(return_value=2)
outer_router = URLRouter(
[url(r"^prefix/", inner_router), url(r"^prefix/stuff/$", test_app)]
)
outermost_router = URLRouter([url(r"", outer_router)])
assert outermost_router({"type": "http", "path": "/prefix/stuff/"}) == 2
with path().
"""
from django.urls import path
test_app = MagicMock(return_value=1)
inner_router = URLRouter([path("test//", test_app)])
def asgi_middleware(inner):
# Some middleware which hides the fact that we have an inner URLRouter
def app(scope):
return inner(scope)
app._path_routing = True
return app
outer_router = URLRouter(
[path("number//", asgi_middleware(inner_router))]
)
assert inner_router({"type": "http", "path": "/test/3/"}) == 1
assert outer_router({"type": "http", "path": "/number/42/test/3/"}) == 1
assert test_app.call_args[0][0]["url_route"] == {
"args": (),
"kwargs": {"number": 42, "page": 3},
}
with pytest.raises(ValueError):
assert outer_router({"type": "http", "path": "/number/42/test/3/bla/"})
with pytest.raises(ValueError):
assert outer_router({"type": "http", "path": "/number/42/blub/"})
from django.conf.urls import url
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import backend.middleware
import backend.consumers
application = ProtocolTypeRouter({
"websocket": AuthMiddlewareStack(backend.middleware.JWTAuthMiddleware(
URLRouter([
url(r"^measurements/subscribe/(?P[\.a-z0-9]+)/$", backend.consumers.MeasurementSubscribeConsumer),
url(r"^kit/$", backend.consumers.KitConsumer),
])
# route('authuser.sub', authuser_sub),
# ]
from django.conf.urls import url
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
# from chat.consumers import AdminChatConsumer, PublicChatConsumer
# from aprs_news.consumers import APRSNewsConsumer
application = ProtocolTypeRouter({
# WebSocket chat handler
"websocket": AuthMiddlewareStack(
URLRouter([
# url(r"^chat/admin/$", AdminChatConsumer),
# -*- coding:utf-8 -*-
# !/usr/bin/env python
# Time 18-1-11
# Author Yo
# Email YoLoveLife@outlook.com
from __future__ import absolute_import,unicode_literals
from ops.urls.socket_urls import ops_routing
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
application = ProtocolTypeRouter({
'websocket': AuthMiddlewareStack(
URLRouter(
ops_routing,
)
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.conf.urls import url
from django.urls import path
from celery_api.ws import CeleryLogWebsocket
from kubeops_api.ws import F2OWebsocket
application = ProtocolTypeRouter({
# Empty for now (http->django views is added by default)
'websocket': AuthMiddlewareStack(
URLRouter([
path('ws/tasks//log/', CeleryLogWebsocket, name='task-log-ws'),
path('ws/progress//', F2OWebsocket, name='execution-progress-ws'),
]),
if token_key:
token = Token.objects.get(key=token_key)
scope["user"] = token.user
close_old_connections()
except Token.DoesNotExist:
scope["user"] = AnonymousUser()
return self.inner(scope)
TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))
application = ProtocolTypeRouter(
{
# (http->django views is added by default)
"websocket": TokenAuthMiddlewareStack(
URLRouter(app.routing.websocket_urlpatterns)
)
# The channel routing defines what connections get handled by what consumers,
# selecting on either the connection type (ProtocolTypeRouter) or properties
# of the connection's scope (like URLRouter, which looks at scope["path"])
# For more, see http://channels.readthedocs.io/en/latest/topics/routing.html
application = ProtocolTypeRouter({
# Channels will do this for you automatically. It's included here as an example.
# "http": AsgiHandler,
# Route all WebSocket requests to our custom chat handler.
# We actually don't need the URLRouter here, but we've put it in for
# illustration. Also note the inclusion of the AuthMiddlewareStack to
# add users and sessions - see http://channels.readthedocs.io/en/latest/topics/authentication.html
"websocket": AuthMiddlewareStack(
URLRouter([
# URLRouter just takes standard Django path() or url() entries.
path("chat/stream/", ChatConsumer),
]),
from channels.routing import ProtocolTypeRouter
from channels.routing import URLRouter
import apps.tasks.api.routing
from apps.accounts.ws_token import TokenAuthMiddleware
application = ProtocolTypeRouter({
# Empty for now (http->django views is added by default)
'websocket': TokenAuthMiddleware(
URLRouter(apps.tasks.api.routing.websocket_urlpatterns)
)