r""" .. autofunction:: run_async .. autofunction:: run_asyncio_coroutine .. autofunction:: download .. autofunction:: run_js .. autofunction:: eval_js .. autofunction:: register_thread .. autofunction:: defer_call .. autofunction:: hold .. autofunction:: data .. autofunction:: get_info .. autoclass:: pywebio.session.coroutinebased.TaskHandle :members: """ import threading from base64 import b64encode from functools import wraps from .base import Session from .coroutinebased import CoroutineBasedSession from .threadbased import ThreadBasedSession, ScriptModeSession from ..exceptions import SessionNotFoundException from ..utils import iscoroutinefunction, isgeneratorfunction, run_as_function, to_coroutine # 当前进程中正在使用的会话实现的列表 _active_session_cls = [] __all__ = ['run_async', 'run_asyncio_coroutine', 'register_thread', 'hold', 'defer_call', 'data', 'get_info', 'run_js', 'eval_js', 'download'] def register_session_implement_for_target(target_func): """根据target_func函数类型注册会话实现，并返回会话实现""" if iscoroutinefunction(target_func) or isgeneratorfunction(target_func): cls = CoroutineBasedSession else: cls = ThreadBasedSession if ScriptModeSession in _active_session_cls: raise RuntimeError("Already in script mode, can't start server") if cls not in _active_session_cls: _active_session_cls.append(cls) return cls def get_session_implement(): """获取当前会话实现。仅供内部实现使用。应在会话上下文中调用""" if not _active_session_cls: _active_session_cls.append(ScriptModeSession) _start_script_mode_server() # 当前正在使用的会话实现只有一个 if len(_active_session_cls) == 1: return _active_session_cls[0] # 当前有多个正在使用的会话实现 for cls in _active_session_cls: try: cls.get_current_session() return cls except SessionNotFoundException: pass raise SessionNotFoundException def _start_script_mode_server(): from ..platform.tornado import start_server_in_current_thread_session start_server_in_current_thread_session() def get_current_session() -> "Session": return get_session_implement().get_current_session() def get_current_task_id(): return get_session_implement().get_current_task_id() def check_session_impl(session_type): def decorator(func): """装饰器：在函数调用前检查当前会话实现是否满足要求""" @wraps(func) def inner(*args, **kwargs): curr_impl = get_session_implement() # Check if 'now_impl' is a derived from session_type or is the same class if not issubclass(curr_impl, session_type): func_name = getattr(func, '__name__', str(func)) require = getattr(session_type, '__name__', str(session_type)) curr = getattr(curr_impl, '__name__', str(curr_impl)) raise RuntimeError("Only can invoke `{func_name:s}` in {require:s} context." " You are now in {curr:s} context".format(func_name=func_name, require=require, curr=curr)) return func(*args, **kwargs) return inner return decorator def chose_impl(gen_func): """根据当前会话实现来将 gen_func 转化为协程对象或直接以函数运行""" @wraps(gen_func) def inner(*args, **kwargs): gen = gen_func(*args, **kwargs) if get_session_implement() == CoroutineBasedSession: return to_coroutine(gen) else: return run_as_function(gen) return inner @chose_impl def next_client_event(): res = yield get_current_session().next_client_event() return res @chose_impl def hold(): """保持会话，直到用户关闭浏览器， 此时函数抛出 `SessionClosedException ` 异常。 注意⚠️：在 :ref:`基于协程 ` 的会话上下文中，需要使用 ``await hold()`` 语法来进行调用。 """ while True: yield next_client_event() def download(name, content): """下载文件 :param str name: 下载保存为的文件名 :param content: 文件内容. 类型为 bytes-like object """ from ..io_ctrl import send_msg content = b64encode(content).decode('ascii') send_msg('download', spec=dict(name=name, content=content)) def run_js(code): """运行js代码. 代码运行在浏览器的JS全局作用域中 :param str code: js代码 """ from ..io_ctrl import send_msg send_msg('run_script', spec=code) @chose_impl def eval_js(expression): """执行js表达式，并获取表达式的值 :param str expression: js表达式. 表达式的值需要能JSON序列化 :return: js表达式的值 注意⚠️：在 :ref:`基于协程 ` 的会话上下文中，需要使用 ``await eval_js(expression)`` 语法来进行调用。 Example:: current_url = eval_js("window.location.href") function_res = eval_js('''(function(){ var a = 1; a += 100; return a; })()''') """ script = r""" (function(WebIO){ let result = null; try{ result = eval(%r); }catch{}; WebIO.sendMessage({ event: "js_yield", task_id: WebIOCurrentTaskID, // local var in run_script command data: result || null }); })(WebIO);""" % expression run_js(script) res = yield next_client_event() assert res[ 'event'] == 'js_yield', "Internal Error, please report this bug on https://github.com/wang0618/PyWebIO/issues" return res['data'] @check_session_impl(CoroutineBasedSession) def run_async(coro_obj): """异步运行协程对象。协程中依然可以调用 PyWebIO 交互函数。 仅能在 :ref:`基于协程 ` 的会话上下文中调用 :param coro_obj: 协程对象 :return: An instance of `TaskHandle ` is returned, which can be used later to close the task. """ return get_current_session().run_async(coro_obj) @check_session_impl(CoroutineBasedSession) async def run_asyncio_coroutine(coro_obj): """若会话线程和运行事件的线程不是同一个线程，需要用 run_asyncio_coroutine 来运行asyncio中的协程。 仅能在 :ref:`基于协程 ` 的会话上下文中调用。 :param coro_obj: 协程对象 """ return await get_current_session().run_asyncio_coroutine(coro_obj) @check_session_impl(ThreadBasedSession) def register_thread(thread: threading.Thread): """注册线程，以便在线程内调用 PyWebIO 交互函数。仅能在默认的基于线程的会话上下文中调用。 :param threading.Thread thread: 线程对象 """ return get_current_session().register_thread(thread) def defer_call(func): """设置会话结束时调用的函数。无论是用户主动关闭会话还是任务结束会话关闭，设置的函数都会被运行。 可以用于资源清理等工作。 在会话中可以多次调用 `defer_call()` ,会话结束后将会顺序执行设置的函数。 `defer_call` 同样支持以装饰器的方式使用:: @defer_call def cleanup(): pass :param func: 话结束时调用的函数 .. attention:: 通过 `defer_call()` 设置的函数被调用时会话已经关闭，所以在函数体内不可以调用 PyWebIO 的交互函数 """ get_current_session().defer_call(func) return func def data(): """获取当前会话的数据对象，用于在对象上保存一些会话相关的数据。访问数据对象不存在的属性时会返回None而不是抛出异常。 """ return get_current_session().save def get_info(): """ 获取当前会话的相关信息 :return: 表示会话信息的对象，属性有： * ``user_agent`` : 表示用户浏览器信息的对象，属性有 * ``is_mobile`` (bool): 用户使用的设备是否为手机 (比如 iPhone, Android phones, Blackberry, Windows Phone 等设备) * ``is_tablet`` (bool): 用户使用的设备是否为平板 (比如 iPad, Kindle Fire, Nexus 7 等设备) * ``is_pc`` (bool): 用户使用的设备是否为桌面电脑 (比如运行 Windows, OS X, Linux 的设备) * ``is_touch_capable`` (bool): 用户使用的设备是否支持触控 * ``browser.family`` (str): 浏览器家族. 比如 'Mobile Safari' * ``browser.version`` (tuple): 浏览器版本元组. 比如 (5, 1) * ``browser.version_string`` (str): 浏览器版本字符串. 比如 '5.1' * ``os.family`` (str): 操作系统家族. 比如 'iOS' * ``os.version`` (tuple): 操作系统版本元组. 比如 (5, 1) * ``os.version_string`` (str): 操作系统版本字符串. 比如 '5.1' * ``device.family`` (str): 设备家族. 比如 'iPhone' * ``device.brand`` (str): 设备品牌. 比如 'Apple' * ``device.model`` (str): 设备幸好. 比如 'iPhone' * ``user_language`` (str): 用户操作系统使用的语言. 比如 ``'zh-CN'`` * ``server_host`` (str): 当前会话的服务器host，包含域名和端口，端口为80时可以被省略 * ``origin`` (str): 当前用户的页面地址. 包含 协议、主机、端口 部分. 比如 ``'http://localhost:8080'`` . 可能为空，但保证当用户的页面地址不在当前服务器下(即 主机、端口部分和 ``server_host`` 不一致)时有值. * ``user_ip`` (str): 用户的ip地址. * ``backend`` (str): PyWebIO使用的Web框架名. 目前可用值有 ``'tornado'`` , ``'flask'`` , ``'django'`` , ``'aiohttp'`` . * ``request`` (object): 创建当前会话时的Web请求对象. 根据PyWebIO使用的后端Server不同，``request`` 的类型也不同: * 使用Tornado后端时, ``request`` 为 `tornado.httputil.HTTPServerRequest `_ 实例 * 使用Flask后端时, ``request`` 为 `flask.Request `_ 实例 * 使用Django后端时, ``request`` 为 `django.http.HttpRequest `_ 实例 * 使用aiohttp后端时, ``request`` 为 `aiohttp.web.BaseRequest `_ 实例 返回值的 ``user_agent`` 属性是通过 user-agents 库进行解析生成的。参见 https://github.com/selwin/python-user-agents#usage """ return get_current_session().info