``put_xxx()`` 类函数的返回值 若 ``put_xxx()`` 调用的返回值没有被变量接收,则直接将消息发送到会话; 否则消息则作为其他消息的一部分
| 32 | |
| 33 | |
| 34 | class Output: |
| 35 | """ ``put_xxx()`` 类函数的返回值 |
| 36 | |
| 37 | 若 ``put_xxx()`` 调用的返回值没有被变量接收,则直接将消息发送到会话; |
| 38 | 否则消息则作为其他消息的一部分 |
| 39 | """ |
| 40 | |
| 41 | @staticmethod |
| 42 | def json_encoder(obj, ignore_error=False): |
| 43 | """json序列化与输出相关消息的Encoder函数 """ |
| 44 | if isinstance(obj, Output): |
| 45 | return obj.embed_data() |
| 46 | elif isinstance(obj, OutputList): |
| 47 | return obj.data |
| 48 | |
| 49 | if not ignore_error: |
| 50 | raise TypeError('Object of type %s is not JSON serializable' % obj.__class__.__name__) |
| 51 | |
| 52 | @classmethod |
| 53 | def dump_dict(cls, data): |
| 54 | # todo 使用其他方式来转换spec |
| 55 | return json.loads(json.dumps(data, default=cls.json_encoder)) |
| 56 | |
| 57 | @classmethod |
| 58 | def safely_destruct(cls, obj): |
| 59 | """安全销毁 OutputReturn 对象/包含OutputReturn对象的dict/list, 使 OutputReturn.__del__ 不进行任何操作""" |
| 60 | try: |
| 61 | json.dumps(obj, default=partial(cls.json_encoder, ignore_error=True)) |
| 62 | except Exception: |
| 63 | pass |
| 64 | |
| 65 | def __init__(self, spec, on_embed=None): |
| 66 | self.processed = False |
| 67 | self.on_embed = on_embed or (lambda d: d) |
| 68 | try: |
| 69 | self.spec = type(self).dump_dict(spec) # this may raise TypeError |
| 70 | except TypeError: |
| 71 | self.processed = True |
| 72 | type(self).safely_destruct(spec) |
| 73 | raise |
| 74 | |
| 75 | # For Context manager |
| 76 | self.enabled_context_manager = False |
| 77 | self.container_selector = None |
| 78 | self.container_dom_id = None # todo: this name is ambiguous, rename it to `scope_name` or others |
| 79 | self.after_exit = None |
| 80 | |
| 81 | # Try to make sure current session exist. |
| 82 | # If we leave the session interaction in `Output.__del__`, |
| 83 | # the Exception raised from there will be ignored by python interpreter, |
| 84 | # thus we can't end some session in some cases. |
| 85 | # See also: https://github.com/pywebio/PyWebIO/issues/243 |
| 86 | get_current_session() |
| 87 | |
| 88 | def enable_context_manager(self, container_selector=None, container_dom_id=None, after_exit=None): |
| 89 | self.enabled_context_manager = True |
| 90 | self.container_selector = container_selector |
| 91 | self.container_dom_id = container_dom_id |
no outgoing calls
no test coverage detected
searching dependent graphs…