Chào các bạn, ở bài trước chúng ta đã kết nối database và tạo account cho user. Trong bài hôm nay, chúng ta sẽ hoàn tất chương trình chat của chúng ta, các việc cần làm là:

• Login account vào trang room chat
• Tạo websocket kết nối với chương trình simple chat
• Khi user chat chúng ta sẽ lưu tin nhắn vào database và user login vào sẽ thấy những tin nhắn cũ

Để chuẩn bị cho bài hôm nay, chúng ta sẽ cài đặt một package mới là aiohttp_session. Đây là một package giúp chúng ta quản lý session của aiothttp. Bạn sẽ cài đặt package này với command sau:

pip install aiohttp_session
pip install cryptography

Chuẩn bị database

Ở bài trước chúng ta đã tạo table users,ở bài này chúng ta sẽ tạo thêm một table để lưu trữ nội dung chat của các users. Bây giờ, bạn cần mở file chat/model.py và thêm đoạn sau vào async def createdb(self):

        async with aiosqlite.connect(self.db_file) as db:
            await db.execute(
                "create table if not exists msg "
                "("
                "id integer primary key asc, "
                "user_id integer, created_at datetime,"
                "msg text"
                ")"
            )

Tìm class User(): và thêm 2 function mới này vào

    async def login_user(self, username, password):
        async with aiosqlite.connect(self.db_file) as db:
            password = hashlib.md5(password.encode('utf-8')).hexdigest()
            cursor = await db.execute("select * from users where username = '{0}' "
                                      "and password = '{1}'".format(username, password))

            rows = await cursor.fetchone()
            await cursor.close()
            return rows


    async def get_login_user(self, user_id):
        async with aiosqlite.connect(self.db_file) as db:
            cursor = await db.execute("select * from users where id = {0}".format(user_id))
            rows = await cursor.fetchone()
            await cursor.close()
            return rows

Tiếp tục tại chat/model.py, ta thêm đoạn sau để quản lý message cho user

class Message:
    def __init__(self):
        self.db_file = settings.DB_FILE

    async def save_msg(self, data):
        async with aiosqlite.connect(self.db_file) as db:
            await db.execute("insert into msg (user_id, created_at, msg) "
                             "values (?, ?, ?)",
                             [data.get('user_id'), data.get('created_at'),
                              data.get('msg')])
            await db.commit()

    async def load_msg(self):
        async with aiosqlite.connect(self.db_file) as db:
            cursor = await db.execute("SELECT users.username, msg.created_at, msg.msg FROM users"
                                      " inner join msg ON users.id = msg.user_id limit {0} OFFSET"
                                      " (SELECT COUNT(*) FROM msg)-{0};".format(settings.MAX_MSG))
            rows = await cursor.fetchall()
            await cursor.close()
            return rows

Sau khi bạn thêm tất cả thì file model.py sẽ như sau

import aiosqlite
import settings
import hashlib

class InitDB():
    def __init__(self):
        self.db_file = settings.DB_FILE

    async def createdb(self):
        async with aiosqlite.connect(self.db_file) as db:
            await db.execute(
                "create table if not exists users "
                "("
                "id integer primary key asc, "
                "username varchar(50), password varchar(50),"
                "email varchar(50)"
                ")"
            )

        async with aiosqlite.connect(self.db_file) as db:
            await db.execute(
                "create table if not exists msg "
                "("
                "id integer primary key asc, "
                "user_id integer, created_at datetime,"
                "msg text"
                ")"
            )

class User():
    def __init__(self):
        self.db_file = settings.DB_FILE


    async def check_user(self, username):
        async with aiosqlite.connect(self.db_file) as db:
            cursor = await db.execute("select * from users where username = '{}'".format(username))
            rows = await cursor.fetchone()
            await cursor.close()
            return rows

    async def create_user(self, data):
        result = False
        user = await self.check_user(data.get('username'))

        if not user and data.get('username'):
            async with aiosqlite.connect(self.db_file) as db:
                password = hashlib.md5(data.get('password').encode('utf-8')).hexdigest()
                results = await db.execute("insert into users (username, password, email) "
                                           "values (?, ?, ?)",
                                 [data.get('username'), password,
                                  data.get('email')])
                await db.commit()
                if results.lastrowid:
                    result = await self.get_login_user(results.lastrowid)
                await results.close()

        return result

    async def login_user(self, username, password):
        async with aiosqlite.connect(self.db_file) as db:
            password = hashlib.md5(password.encode('utf-8')).hexdigest()
            cursor = await db.execute("select * from users where username = '{0}' "
                                      "and password = '{1}'".format(username, password))

            rows = await cursor.fetchone()
            await cursor.close()
            return rows


    async def get_login_user(self, user_id):
        async with aiosqlite.connect(self.db_file) as db:
            cursor = await db.execute("select * from users where id = {0}".format(user_id))
            rows = await cursor.fetchone()
            await cursor.close()
            return rows

class Message:
    def __init__(self):
        self.db_file = settings.DB_FILE

    async def save_msg(self, data):
        async with aiosqlite.connect(self.db_file) as db:
            await db.execute("insert into msg (user_id, created_at, msg) "
                             "values (?, ?, ?)",
                             [data.get('user_id'), data.get('created_at'),
                              data.get('msg')])
            await db.commit()

    async def load_msg(self):
        async with aiosqlite.connect(self.db_file) as db:
            cursor = await db.execute("SELECT users.username, msg.created_at, msg.msg FROM users"
                                      " inner join msg ON users.id = msg.user_id limit {0} OFFSET"
                                      " (SELECT COUNT(*) FROM msg)-{0};".format(settings.MAX_MSG))
            rows = await cursor.fetchall()
            await cursor.close()
            return rows

Bây giờ ta sẽ tiếp tục với phần giao diên, bạn hãy mở chat/views.py tìm phần sau

import json
from aiohttp import web
import aiohttp_jinja2
from time import time
from .model import User

def redirect(request, router_name):
    url = request.app.router[router_name].url_for()
    raise web.HTTPFound(url)

def set_session(session, user_data):
    session['user'] = user_data
    session['last_visit'] = time()

def convert_json(message):
    return json.dumps({'error': message})

Và thay thế bằng đoạn này

import json
import settings
from aiohttp import web, WSMsgType #1
from aiohttp_session import get_session #2
import aiohttp_jinja2
from time import time #3
from datetime import datetime #4
from .model import User, Message #5
from settings import log #6

history = [] #7

def redirect(request, router_name):
    url = request.app.router[router_name].url_for()
    raise web.HTTPFound(url)

def set_session(session, user_data):
    session['user'] = user_data
    session['last_visit'] = time()

def convert_json(message):
    return json.dumps({'error': message})

async def load_msg():
    if not history: #8
        message = Message()
        messages = await message.load_msg() #9
        for msg in messages:
            history.append(
                {'time': datetime.strptime(msg[1], '%Y-%m-%d %H:%M:%S.%f'), 'user': msg[0], 'msg': msg[2]}) #10

Đây là những module chúng ta cần sử dụng cho bài hôm nay

1. WSMsgType: module này giúp chúng ta kiểm tra loại của websocket trả về là text hoặc error
2. get_session: module của package aiohttp_session giúp chúng ta truy cập vào session của aiohttp
3. time: built-in module của python để ta lấy thời gian hiện tại dạng milisecond set vào session
4. datetime: built-in module của python lấy thời gian hiện tại dạng ngày tháng năm để lưu vào database, history, và hiển thị ra room chat
5. Message: class mới dùng để quản lý tin nhắn (lấy tin nhắn hiện tại, lưu tin nhắn vào database)
6. log: dùng để hiển thị lỗi ra terminal
7. Ta tạo một list history để lưu trữ tin nhắn và hiển thị cho các user đăng nhập vào thấy
8. Mỗi khi ta restart server history sẽ mất hết, và tránh để mỗi khi có user mới login hay register load lại history, chúng ta kiểm tra nếu history rỗng thì mới load message từ databasse
9. message.load_msg() sẽ load 20 message mới nhất và show cho user vừa login xem
10. Ta lưu history dưới dạng dictionary gồm có: time - thời gian chat, dùng hàm datetime.strptime() để định dạng, user - username được lưu trữ trong table msg, msg - nội dung chat

Login account vào room chat

Trong chat/views.py, bạn copy function post vào để user có thể login với account đã tạo, chúng ta sẽ cùng xem qua class Login để hiểu rõ hơn đoạn code này thực hiện như thế nào

class Login(web.View): #1
    @aiohttp_jinja2.template('chat/login.html')
    async def get(self):
        session_user = await get_session(self.request) #2
        if session_user.get('user'): #3
            redirect(self.request, 'room_chat') #4
        return None

    async def post(self): #5
        data = await self.request.post() #6
        user = User()
        result = await user.login_user(data.get('username'), data.get('password')) #7
        if isinstance(result, tuple): #8
            session = await get_session(self.request)
            set_session(session, {'id': result[0], 'username': result[1]}) #9

            message = Message()
            messages = await message.load_msg()
            history.clear()
            for msg in messages:
                history.append({'time': datetime.strptime(msg[1], '%Y-%m-%d %H:%M:%S.%f'), 'user': msg[0], 'msg': msg[2]})

            redirect(self.request, 'room_chat')
        else:
            return web.Response(text="Can't login")

1. Các tính năng đăng nhập chúng ta sẽ viết tại class Login này.
2. function get() ở bài trước chúng ta đã sử dụng làm giao diện cho trang login, ở bài này ta sẽ bổ sung tính năng kiểm xem user này có đăng nhập chưa, nếu đã đăng nhập thì chúng ta không cho user này vào lại trang login mà redirect thẳng tới trang room chat. Để lấy session của user hiện tại ta dùng get_session và truyền vào tham số self.request
3. Sau khi truyền tham số self.request vào get_session, get_session sẽ trả về 1 dictionary, ta có thể kiểm tra xem user có tồn tại chưa bằng session_user.get('user')
4. redirect là function chúng ta thêm vào, giúp chuyển trang hiện tại của user đến trang cần đến. Chúng ta sẽ xem chi tiết ở phần sau.
5. Để đăng nhập ta phải nhận dữ liệu được post từ form login.
6. self.request.post() là function để chúng ta lấy dữ liệu post từ form, kết quả trả về là một dictionary
7. Sau khi lấy được username và password, ta sẽ đưa vào function login_user để tìm user trong database. Nếu username và password nhập vào trùng với trong database sẽ trả về kết quả thông tin user.
8. Sqlite trả về các record là tuple, không có column name, nên chúng ta sẽ kiểm tra kết quả trả về nếu là tuple sẽ lưu session, nếu không sẽ trả về message "Can't login"
9. Thông tin user chúng ta cần lưu vào session là id của user và username. Như đã nói trên, sqlite trả về record là tuple, thứ tự sắp xếp theo cột trong database

class Logout(web.View):

    async def get(self):
        session = await get_session(self.request)
        if session.get('user'):
            del session['user'] #

        redirect(self.request, 'homepage') #2

1. Chúng ta sẽ kiểm tra xem session có user hay không, nếu có chúng ta sẽ xóa bằng các dùng del session['user']. Sau khi xóa session, user buộc phải login lại để vào room chat. Bạn có thể tìm hiểu thêm del trong dictionary tại đây
2. Nếu user không login nhưng cố tình vào trang logout chúng ta sẽ buộc user quay về homepage

Tiếp theo bạn tạo page room chat trong chat/views.py

class RoomChat(web.View):
    @aiohttp_jinja2.template('chat/room_chat.html')
    async def get(self):
        session = await get_session(self.request)
        if not session.get('user'):
            redirect(self.request, 'homepage') #1

        return {'messages': history} #2

1. Nếu user chưa login thì ta redirect user về homepage
2. Nếu user login thành công ta load history và hiển thị ra room chat

Tạo file templates/chat/room_chat.html với nội dung như sau

{% extends "base.html" %}

{% block title %}Begin chat{% endblock %}

{% block body_content %}
    <div id='subscribe'>
        {% for mes in messages%}
        <p>[{{ mes['time'].strftime('%H:%M:%S') }}] ({{ mes['user'] }}) {{ mes['msg'] }}</p>
        {% endfor %}
    </div>
    <div style="clear:both;width: 100%;">&nbsp;</div>
    <div>
        <input  type="text" id="message" autocomplete="off" style="border: 1px solid">
        <input class="btn btn-primary" type="submit" id='submit' value="Send">
        <input class="btn btn-default" type="button" id="signout" value="Sign Out">
    </div>
{% endblock %}

{% block footerjs %}
    <script src="{{ url('static', filename='js/socket.js') }}"></script>
{% endblock %}

Mở file static/css/main.css và thêm đoạn này ở cuối file

/* Custom css */
#subscribe {
  height: 200px;
  width: 100%;
  overflow: scroll;
}

Tạo file static/js/socket.js với nội dung như sau:

$(document).ready(function(){
        try{
            var sock = new WebSocket('ws://' + window.location.host + '/ws'); #1
        }
        catch(err){
            var sock = new WebSocket('wss://' + window.location.host + '/ws');#1.1
        }

        // show message in div#subscribe
        function showMessage(message) {
            var messageElem = $('#subscribe'),
                height = 0,
                date = new Date();
                options = {hour12: false};
            messageElem.append($('<p>').html('[' + date.toLocaleTimeString('en-US', options) + '] ' + message + '\n'));
            messageElem.find('p').each(function(i, value){
                height += parseInt($(this).height());
            });
            if(messageElem.find('p').length>20){
                messageElem.find('p:first').remove();
            }

            messageElem.animate({scrollTop: height});
        }

        function sendMessage(){
            var msg = $('#message');
            sock.send(msg.val());
            msg.val('').focus();
        }

        sock.onopen = function(){
            showMessage('Connection to server started');
        }

        // send message from form
        $('#submit').click(function() {
            sendMessage();
        });

        $('#message').keyup(function(e){
            if(e.keyCode == 13){
                sendMessage();
            }
        });

        // income message handler
        sock.onmessage = function(event) {
          showMessage(event.data);
        };

        $('#signout').click(function(){
            window.location.href = "/logout"
        });

        sock.onclose = function(event){
            if(event.wasClean){
                showMessage('Clean connection end')
            }else{
                showMessage('Connection broken')
            }
        };

        sock.onerror = function(error){
            showMessage(error);
        }
    });

1 và 1.1 là địa chỉ websocket của chúng ta. Nếu trình duyệt bạn đang dùng không hỗ trợ protocol ws://, chúng ta sẽ thử wss://

Và đây, cuối cùng cũng tới class WebSocket. WebSocket sẽ đảm nhận việc push các thông báo khi có user mới login, logout hoặc push tin nhắn mới tới tất cả user trong room chat theo thời gian thật. Mình sẽ giải thích những function mới trong class này.

class WebSocket(web.View):
    async def get(self):
        ws = web.WebSocketResponse() #1
        await ws.prepare(self.request)#2

        session = await get_session(self.request)
        message = Message()

        session_user = session.get('user')
        username = session_user.get('username')

        for _ws in self.request.app['websockets']: #3
            await _ws.send_str('%s joined' % username) #4

        self.request.app['websockets'].append(ws)

        async for msg in ws: #5
            if msg.type == WSMsgType.text: #6
                for _ws in self.request.app['websockets']:
                    time_chat = datetime.now()
                    if len(history) > settings.MAX_MSG:
                        del history[0] #7
                    history.append({'time': time_chat, 'user': username, 'msg': msg.data})
                    await message.save_msg({'user_id': session_user.get('id'), 'created_at': time_chat, 'msg': msg.data}) #8
                    await _ws.send_str('(%s) %s' % (username, msg.data)) #9
                    

            elif msg.type == WSMsgType.error:
                log.debug('ws connection closed with exception %s' % ws.exception()) #10

        self.request.app['websockets'].remove(ws) #11

        for _ws in self.request.app['websockets']:
            await _ws.send_str('%s disconected' % username) #12

        log.debug('websocket connection closed')

        return ws

1. WebSocketResponse: class xử lý server-side websocket, class này được kế thừa từ StreamResponose. Xem thêm chi tiết tại đây
2. Bạn bắt buộc cần ws.prepare() để có thể sử dụng các method của websocket
3. Các websocket sẽ được lưu trữ trong list để sử dụng push các tin nhắn đến các user trong room chat
4. Để push tin nhắn đến user, ta dùng function _ws.send_str("tin nhắn cần push")
5. Để biết được tin message của websocket thuộc loại nào, ta sẽ dùng vòng lặp for các message trong websocket để xử lý
6. Khi tin nhắn được nhắn tới chúng ta sẽ biết nó thuộc loại nào với method msg.type và lấy được nội dung tin nhắn với msg.data
7. Để giữ cho list history không quá 20 item, chúng ta sẽ kiểm tra nếu history > settings.MAX_MSG thì sẽ xóa item đầu tiên
8. Ta lưu tin nhắn user đã gửi vào database bằng function save_msg
9. Push tin nhắn của user đến room chat
10. Nếu websocket gặp bất kỳ lỗi nào ngoài ý muốn thì chúng ta sẽ log lại lỗi này để debug
11. Khi user mất kết nối với room chat như: mất internet, refresh page, logout... thì ta sẽ xóa socket của user này
12. Đồng thời thông báo cho toàn bộ user khác là "User A" disconnected

Sau khi bạn thêm tất cả function, thì chat/views.py sẽ như sau

import json
import settings
from aiohttp import web, WSMsgType
from aiohttp_session import get_session
import aiohttp_jinja2
from time import time
from datetime import datetime
from .model import User, Message
from settings import log

history = []

def redirect(request, router_name):
    url = request.app.router[router_name].url_for()
    raise web.HTTPFound(url)

def set_session(session, user_data):
    session['user'] = user_data
    session['last_visit'] = time()

def convert_json(message):
    return json.dumps({'error': message})

async def load_msg():
    if not history:
        message = Message()
        messages = await message.load_msg()

        for msg in messages:
            history.append(
                {'time': datetime.strptime(msg[1], '%Y-%m-%d %H:%M:%S.%f'), 'user': msg[0], 'msg': msg[2]})

class Login(web.View):
    @aiohttp_jinja2.template('chat/login.html')
    async def get(self):
        session_user = await get_session(self.request)
        if session_user.get('user'):
            redirect(self.request, 'room_chat')
        return None

    async def post(self):
        data = await self.request.post()
        user = User()
        result = await user.login_user(data.get('username'), data.get('password'))
        if isinstance(result, tuple):
            session = await get_session(self.request)
            set_session(session, {'id': result[0], 'username': result[1]})

            await load_msg()

            redirect(self.request, 'room_chat')
        else:
            return web.Response(text="Can't login")

class CreateUser(web.View):
    @aiohttp_jinja2.template('chat/create_user.html')
    async def get(self):
        return None

    async def post(self):
        data = await self.request.post()

        user = User()
        post = {'username': data.get('username'),
                'password': data.get('password'),
                'email': data.get('email')}
        result = await user.create_user(data=post)
        if isinstance(result, tuple):
            session = await get_session(self.request)
            set_session(session, {'id': result[0], 'username': result[1]})

            await load_msg()

            redirect(self.request, 'room_chat')
        else:
            return web.Response(text="Can't register")

class Logout(web.View):

    async def get(self):
        session = await get_session(self.request)
        if session.get('user'):
            del session['user']

        redirect(self.request, 'homepage')

class RoomChat(web.View):
    @aiohttp_jinja2.template('chat/room_chat.html')
    async def get(self):
        session = await get_session(self.request)
        if not session.get('user'):
            redirect(self.request, 'homepage')

        return {'messages': history}


class WebSocket(web.View):
    async def get(self):
        ws = web.WebSocketResponse()
        await ws.prepare(self.request)

        session = await get_session(self.request)
        message = Message()

        session_user = session.get('user')
        username = session_user.get('username')

        for _ws in self.request.app['websockets']:
            await _ws.send_str('%s joined' % username)

        self.request.app['websockets'].append(ws)

        async for msg in ws:
            if msg.type == WSMsgType.text:
                for _ws in self.request.app['websockets']:
                    time_chat = datetime.now()
                    if len(history) > settings.MAX_MSG:
                        del history[0]

                    history.append({'time': time_chat, 'user': username, 'msg': msg.data})
                    await message.save_msg(
                        {'user_id': session_user.get('id'), 'created_at': time_chat, 'msg': msg.data})
                    await _ws.send_str('(%s) %s' % (username, msg.data))

            elif msg.type == WSMsgType.error:
                log.debug('ws connection closed with exception %s' % ws.exception())

        self.request.app['websockets'].remove(ws)

        for _ws in self.request.app['websockets']:
            await _ws.send_str('%s disconected' % username)

        log.debug('websocket connection closed')

        return ws

Bây giờ bạn mở file routes.py và thay thế bằng nội dung như sau

from aiohttp import web
from chat.views import CreateUser, Login, Logout, WebSocket, RoomChat

routes = [
    web.get('/', Login, name='homepage'),
    web.get('/createuser', CreateUser, name='createuser'),
    web.post('/createuser', CreateUser),
    web.post('/login', Login, name='login'),
    web.get('/logout', Logout, name='logout'),
    web.get('/room_chat', RoomChat, name='room_chat'),
    web.get('/ws', WebSocket)
]

Middleware

aiohttp.web cung cấp một cơ chế mạnh mẽ để tùy biến request handlers thông qua middleware. Đọc thêm chi tiết tại đây. Bạn hãy mở file server.py và chỉnh sửa như sau

from aiohttp_session import session_middleware #1
from aiohttp_session.cookie_storage import EncryptedCookieStorage #2
from cryptography import fernet #3

fernet_key = fernet.Fernet.generate_key()
secret_key = base64.urlsafe_b64decode(fernet_key)
app = web.Application(
    middlewares=[
        session_middleware(EncryptedCookieStorage(secret_key)), #4
    ]
)
app['websockets'] = [] #5

1. session_middleware là module của aiohttp_session, cho phép chúng ta thêm vào middleware của aiohttp khả năng quản lý session
2. EncryptedCookieStorage sẽ mã hóa dữ liệu session và lưu vào HTTP Cookies. Xem thêm tại đây
3. fernet dùng để tạo secret key dùng cho EncryptedCookieStorage
4. Để thêm session_middleware, bạn cần đặt session_middleware trong 1 list cho param middlewares. session_middleware nhận vào một param là mã hóa của EncryptedCookieStorage
5. Ta cần tạo một list websocket để quản lý socket của các user

Sau khi chỉnh sửa hoàn tất, file server.py như sausau

import jinja2
import base64
from aiohttp import web
import aiohttp_jinja2 as jtemplate
from routes import routes
import settings
from chat.model import InitDB
import asyncio
from aiohttp_session import session_middleware
from aiohttp_session.cookie_storage import EncryptedCookieStorage
from cryptography import fernet

fernet_key = fernet.Fernet.generate_key()
secret_key = base64.urlsafe_b64decode(fernet_key)
app = web.Application(
    middlewares=[
        session_middleware(EncryptedCookieStorage(secret_key)),
    ]
)
app['websockets'] = []
app.add_routes(routes)
app.router.add_static('/static', settings.STATIC_PATH, name='static')
app.router.add_static('/media', settings.MEDIA_PATH, name='media')
jtemplate.setup(app, loader=jinja2.FileSystemLoader(settings.TEMPLATE_PATH))

if __name__ == '__main__':
    initdb = InitDB()

    loop = asyncio.get_event_loop()
    loop.run_until_complete(initdb.createdb())
    web.run_app(app)
    loop.close()

Đến đây là kết thúc series simple chat với aiohttp. Để test thử chương trình bạn hãy truy cập vào workspace virtualenv và chạy command

python server.py
======== Running on http://0.0.0.0:8080 ========
(Press CTRL+C to quit)

Bạn có thể download project tại link https://github.com/kikyo2006/simplechat-aiohttp

Nếu các bạn có thắc mắc hoặc góp ý cho series này, xin hãy post câu hỏi bên dưới hoặc join group Python Community Viet Nam để cùng thảo luận. Chúc các bạn thành công.

• ← Bài trước
• Bài kế tiếp →