Source code for pywebio.session.base

import logging
import sys
import traceback
from collections import defaultdict

import user_agents

from ..utils import catch_exp_call

logger = logging.getLogger(__name__)


[docs]class Session: """ 会话对象,由Backend创建 属性: info 表示会话信息的对象 save 会话的数据对象,提供用户在对象上保存一些会话相关数据 由Task在当前Session上下文中调用: get_current_session get_current_task_id get_scope_name pop_scope push_scope send_task_command next_client_event on_task_exception register_callback defer_call 由Backend调用: send_client_event get_task_commands close Task和Backend都可调用: closed Session是不同的后端Backend与协程交互的桥梁: 后端Backend在接收到用户浏览器的数据后,会通过调用 ``send_client_event`` 来通知会话,进而由Session驱动协程的运行。 Task内在调用输入输出函数后,会调用 ``send_task_command`` 向会话发送输入输出消息指令, Session将其保存并留给后端Backend处理。 """ @staticmethod def get_current_session() -> "Session": raise NotImplementedError @staticmethod def get_current_task_id(): raise NotImplementedError def __init__(self, session_info): """ :param session_info: 会话信息。可以通过 Session.info 访问 """ self.internal_save = dict(info=session_info) # some session related info, just for internal used self.save = {} # underlying implement of `pywebio.session.data` self.scope_stack = defaultdict(lambda: ['ROOT']) # task_id -> scope栈 self.deferred_functions = [] # 会话结束时运行的函数 self._closed = False
[docs] def get_scope_name(self, idx): """获取当前任务的scope栈检索scope名 :param int idx: scope栈的索引 :return: scope名,不存在时返回 None """ task_id = type(self).get_current_task_id() try: return self.scope_stack[task_id][idx] except IndexError: raise ValueError("Scope not found")
[docs] def pop_scope(self): """弹出当前scope :return: 当前scope名 """ task_id = type(self).get_current_task_id() try: return self.scope_stack[task_id].pop() except IndexError: raise ValueError("ROOT Scope can't pop")
[docs] def push_scope(self, name): """进入新scope""" task_id = type(self).get_current_task_id() self.scope_stack[task_id].append(name)
def send_task_command(self, command): raise NotImplementedError
[docs] def next_client_event(self) -> dict: """获取来自客户端的下一个事件。阻塞调用,若在等待过程中,会话被用户关闭,则抛出SessionClosedException异常""" raise NotImplementedError
def send_client_event(self, event): raise NotImplementedError def get_task_commands(self) -> list: raise NotImplementedError
[docs] def close(self, nonblock=False): """Close current session :param bool nonblock: Don't block thread. Used in closing from backend. """ if self._closed: return self._closed = True self.deferred_functions.reverse() while self.deferred_functions: func = self.deferred_functions.pop() catch_exp_call(func, logger)
def closed(self) -> bool: return self._closed def on_task_exception(self): from ..output import toast from . import run_js from . import info as session_info logger.exception('Error') toast_msg = "应用发生内部错误" if 'zh' in session_info.user_language else "An internal error occurred in the application" type, value, tb = sys.exc_info() lines = traceback.format_exception(type, value, tb) traceback_msg = ''.join(lines) traceback_msg = 'Internal Server Error\n' + traceback_msg try: toast(toast_msg, duration=1, color='error') run_js("console.error(traceback_msg)", traceback_msg=traceback_msg) except Exception: pass
[docs] def register_callback(self, callback, **options): """ 向Session注册一个回调函数,返回回调id Session需要保证当收到前端发送的事件消息 ``{event: "callback",task_id: 回调id, data:...}`` 时, ``callback`` 回调函数被执行, 并传入事件消息中的 ``data`` 字段值作为参数 """ raise NotImplementedError
[docs] def defer_call(self, func): """设置会话结束时调用的函数。可以用于资源清理。 在会话中可以多次调用 `defer_call()` ,会话结束后将会顺序执行设置的函数。 :param func: 话结束时调用的函数 """ """设置会话结束时调用的函数。可以用于资源清理。""" self.deferred_functions.append(func)
def get_session_info_from_headers(headers): """从Http请求头中获取会话信息 :param headers: 字典类型的Http请求头 :return: 表示会话信息的对象,属性有: * ``user_agent`` : 用户浏览器信息。可用字段见 https://github.com/selwin/python-user-agents#usage * ``user_language`` : 用户操作系统使用的语言 * ``server_host`` : 当前会话的服务器host,包含域名和端口,端口为80时可以被省略 * ``origin`` : 当前用户的页面地址. 包含 协议、主机、端口 部分. 比如 ``'http://localhost:8080'`` . 可能为空,但保证当用户的页面地址不在当前服务器下(即 主机、端口部分和 ``server_host`` 不一致)时有值. """ ua_str = headers.get('User-Agent', '') ua = user_agents.parse(ua_str) user_language = headers.get('Accept-Language', '').split(',', 1)[0].split(' ', 1)[0].split(';', 1)[0] server_host = headers.get('Host', '') origin = headers.get('Origin', '') session_info = dict(user_agent=ua, user_language=user_language, server_host=server_host, origin=origin) return session_info