Chat (websocket) problems with pipelines

[josejachuf]

I have my test app using websocket, but I have problems with pipelines.

@gi0baro as you have suggested in https://github.com/emmett-framework/emmett/issues/257, I have the sockets module separated. It is a bloggy modification

# chat/__init__.py

#coding: utf8

from emmett import App, session, now, url, redirect
from emmett.orm import Database
# from emmett.tools import requires
from emmett.tools.auth import Auth
from emmett.sessions import SessionManager
from chat.modules.emmett_redis import RedisExtension
from chat.models.user import User
from chat.models.chat import Chat

app = App(__name__)
app.config.db.uri = 'sqlite://filename.sqlite'
app.config.auth.single_template = True
app.config.auth.registration_verification = False
app.config.auth.hmac_key = "MassiveDynamicRules"

# app.config.RedisExtension.host = "192.168.1.45"
app.config.RedisExtension.host = "127.0.0.1"
app.config.RedisExtension.port = '6379'
app.config.RedisExtension.password = "foobared"
app.use_extension(RedisExtension)

db = Database(app, auto_migrate=False)
auth = Auth(app, db, user_model=User)
db.define_models(Chat)

# app.pipeline = []
auth_routes = auth.module(__name__)
auth_routes.pipeline = [SessionManager.cookies('Walternate')]

from chat.controllers import *

# chat/controllers/main.py:

# coding: utf8
from emmett import request, response
from emmett.serializers import Serializers
from emmett.sessions import SessionManager
from chat import app, db, auth, Chat

routes = app.module(__name__, name='routes')
routes.pipeline = [SessionManager.cookies('Walternate'), db.pipe, auth.pipe]
json_dump = Serializers.get_for('json')

@routes.route("/")
async def index():
    return dict()

@routes.route("/chat/<str:room>")
async def chat(room):
    return dict(room=room)

@routes.route('/chat/<str:room>/messages', methods=['post'])
async def new_chat_message(room):
    response.status = 201
    params = await request.body_params
    res = Chat.create(**params)
    if res.errors:
        response.status = 422
        return {'error': res.errors.as_dict()}
    row = Chat.get(res.id)
    async with app.ext.RedisExtension.ctx() as redis:
        await redis.publish(f'chat:{room}:messages', json_dump(row))
    return json_dump(row)

# chat/controllers/main.py:

# coding: utf8
from emmett import websocket
from chat import app

sockets = app.module(__name__, name='sockets')

@sockets.websocket('/messages/<str:room>')
async def messages(room):
    async with app.ext.RedisExtension.ctx() as redis:
        channel, = await redis.subscribe(f'chat:{room}:messages')
        await websocket.accept()

        async for message in channel.iter():
            await websocket.send(message)

# chat/templates/chat.html:

{{extend 'layout.html'}}

<h2>Room: {{=room.upper()}}</h2>

<div id="messages" class="messages"></div>

<script>
    var socket = new WebSocket('ws://' + document.domain + ':' + location.port + '/messages/{{=asis(room)}}');
    var messages = document.createElement('ul');
    // Escucha por mensajes
    var divMessages = document.getElementById("messages");
    socket.addEventListener('message', function (event) {
        reader = new FileReader();
        reader.readAsText(event.data);

        reader.onload = () => {
            messages = document.getElementsByTagName('ul')[0];
            var message = document.createElement('li');
            var content = document.createTextNode('Mensaje: ' + reader.result);
            message.appendChild(content);
            messages.appendChild(message);
        };
    });

    divMessages.appendChild(messages);
</script>

As is the above works fine and the messages that are entered are displayed (from API tester plugin) view capture in: https://ibb.co/f1L3pCJ , but if I want to go to http://127.0.0.1:8000/auth/login I get the following error:

AttributeError: 'NoneType' object has no attribute '_csrf' return dir(self._get_robj()) except RuntimeError: return []

def getattr(self, name): return getattr(self._get_robj(), name)

def setitem(self, key, value): self._get_robj()[key] = value Application traceback | Full traceback | Frames ? Traceback (most recent call last): AttributeError: 'NoneType' object has no attribute '_csrf'

Now if I change in:

# chat/init.py change app.pipeline = [] for app.pipeline = [SessionManager.cookies('Walternate')] form login works fine, but the socket stops working:

In browser:

WebSocket connection to 'ws://127.0.0.1:8000/messages/python' failed: Error during WebSocket handshake: Unexpected response code: 500

In terminal emmett:

Traceback (most recent call last): File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/asgi/handlers.py", line 293, in handle_request await self.pre_handler(scope, receive, send) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/asgi/handlers.py", line 313, in dynamic_handler await self.router.dispatch() File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/routing/router.py", line 281, in dispatch await route.dispatcher.dispatch(websocket, reqargs) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/routing/dispatchers.py", line 96, in dispatch await self._parallel_flow(self.flow_open) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/routing/dispatchers.py", line 30, in _parallel_flow raise task.exception() File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/sessions.py", line 59, in open_ws current.session = self._load_session() File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/sessions.py", line 80, in _load_session cookie_data = request.cookies[self.cookie_name].value File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/_internal.py", line 70, in getattr return getattr(self._get_robj(), name) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/_internal.py", line 97, in _get_robj return getattr(self.obj.get(), self.name__) AttributeError: 'WSContext' object has no attribute 'request' ERROR: ASGI callable returned without sending handshake. DEBUG: server - event = connection_lost(None)

[gi0baro]

@josejachuf it's a bug in the session manager. Correct one should be the one with session manager on the entire application pipeline. Will push a fix in master branch in the next couple of days

[gi0baro]

@josejachuf should be fixed with ac7219c. Can you test it, please?

[josejachuf]

Thanks @gi0baro , the patch works fine, I have form login and socket. But now I have another problem in the submit, an error occurs.

ValueError: SELECT "users"."id", "users"."created_at", "users"."updated_at", "users"."email", "users"."password", "users"."registration_key", "users"."reset_password_key", "users"."registration_id", "users"."first_name", "users"."last_name" FROM "users" WHERE ("users"."email" = 'jose@test.com') LIMIT 1 OFFSET 0;

def with_connection_or_raise(f):
    def wrap(*args, **kwargs):
        if not args[0].connection:
            if len(args) > 1:
                raise ValueError(args[1])
            raise RuntimeError('no connection available')
        return f(*args, **kwargs)
    return wrap

Full traceback

Traceback (most recent call last): File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/asgi/handlers.py", line 226, in dynamic_handler http = await self.router.dispatch() File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/routing/router.py", line 212, in dispatch http_cls, output = await route.dispatcher.dispatch(request, reqargs) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/routing/dispatchers.py", line 77, in dispatch rv = await self.get_response(wrapper, reqargs) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/routing/dispatchers.py", line 44, in get_response return self.response_builders[wrapper.method](await self.f(reqargs)) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/tools/auth/exposer.py", line 101, in _login rv['form'] = await self.ext.forms.login(onvalidation=_validate_form) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/forms.py", line 149, in _process self.onvalidation(self) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/tools/auth/exposer.py", line 91, in _validate_form row = self.config.models['user'].get(email=form.params.email) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/orm/models.py", line 625, in get return cls.table(kwargs) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/objects.py", line 563, in call return self._db(query).select(limitby=(0, 1), File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/orm/objects.py", line 333, in select return obj._runselect(*fields, *options) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/emmett/orm/objects.py", line 317, in _runselect return super(Set, self).select(fields, **options) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/objects.py", line 2210, in select return adapter.select(self.query, fields, attributes) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/adapters/sqlite.py", line 82, in select return super(SQLite, self).select(query, fields, attributes) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/adapters/base.py", line 760, in select return self._select_aux(sql, fields, attributes, colnames) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/adapters/base.py", line 716, in _select_aux rows = self._select_aux_execute(sql) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/adapters/base.py", line 710, in _select_aux_execute self.execute(sql) File "/home/jose/emmett/venv38/lib/python3.8/site-packages/pydal/adapters/init.py", line 65, in wrap raise ValueError(args[1]) ValueError: SELECT "users"."id", "users"."created_at", "users"."updated_at", "users"."email", "users"."password", "users"."registration_key", "users"."reset_password_key", "users"."registration_id", "users"."first_name", "users"."last_name" FROM "users" WHERE ("users"."email" = 'jose@test.com') LIMIT 1 OFFSET 0;

the pipelines are: #chat/__init__.py:

app.pipeline = [SessionManager.cookies('Walternate')]
auth_routes = auth.module(__name__)
# with and without auth_routes.pipeline
# auth_routes.pipeline = [db.pipe, auth.pipe]

#chat/main.py:

routes = app.module(__name__, name='routes')
routes.pipeline = [db.pipe, auth.pipe]

[gi0baro]

@josejachuf yeah, the issue is that auth module need database pipe. Now, with the actual apis is not possible to attach the auth module to an upper module or to provide a specific pipeline for it. Gonna publish a patch for it later today.

In the meantime, if you move the database pipe to the application pipeline, it should work, even if it waste connections.

[josejachuf]

Hi @gi0baro, this work fine:

In the meantime, if you move the database pipe to the application pipeline, it should work, even if it waste connections.

I want to know if I understood correctly. In this case, for each connection to the socket, is a new connection to the database created?

[gi0baro]

Hi @gi0baro, this work fine:

In the meantime, if you move the database pipe to the application pipeline, it should work, even if it waste connections.

I want to know if I understood correctly. In this case, for each connection to the socket, is a new connection to the database created?

Exactly.

[gi0baro]

@josejachuf since f50b5ae you should be able to customise the pipeline for the auth module, eg:

auth_routes = auth.module(__name__, pipeline=[db.pipe])

It will be released with 2.0.0a5, I'm closing this.

[josejachuf]

@gi0baro thanks, this works fine.

my pipelines stayed this way:

app: [<emmett.sessions.CookieSessionPipe object at 0x7ff9110f8c10>] auth_routes: [<emmett.orm.base.DatabasePipe object at 0x7ff9110f8fd0>] routes: [<emmett.orm.base.DatabasePipe object at 0x7ff91110e730>]