-
Notifications
You must be signed in to change notification settings - Fork 60
/
_modules.py
109 lines (82 loc) · 3.75 KB
/
_modules.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
__all__ = ("namespaced_id", "module_ui", "module_server")
import copy
import sys
from typing import Any, Callable, TypeVar
if sys.version_info < (3, 10):
from typing_extensions import ParamSpec, Concatenate
else:
from typing import ParamSpec, Concatenate
from ._namespaces import namespaced_id
from .session import Inputs, Outputs, Session, require_active_session, session_context
class ModuleInputs(Inputs):
"""
A class representing a module's outputs.
Warning
-------
An instance of this class is created for each request and passed as an argument to
the :class:`shiny.modules.Module`'s ``server`` function. For this reason, you
shouldn't need to create instances of this class yourself. Furthermore, you
probably shouldn't need this class for type checking either since it has the same
signature as :class:`shiny.session.Session`.
"""
def __init__(self, ns: str, parent_inputs: Inputs):
self._namespaces = copy.copy(parent_inputs._namespaces)
self._namespaces.insert(0, ns)
# Don't set _parent attribute like the other classes since Inputs redefines
# __setattr__
self._map = parent_inputs._map
class ModuleOutputs(Outputs):
"""
A class representing a module's outputs.
Warning
-------
An instance of this class is created for each request and passed as an argument to
the :class:`shiny.modules.Module`'s ``server`` function. For this reason, you
shouldn't need to create instances of this class yourself. Furthermore, you
probably shouldn't need this class for type checking either since it has the same
signature as :class:`shiny.session.Session`.
"""
def __init__(self, ns: str, parent_outputs: Outputs):
self._namespaces = copy.copy(parent_outputs._namespaces)
self._namespaces.insert(0, ns)
self._parent = parent_outputs
def __getattr__(self, attr: str) -> Any:
return getattr(self._parent, attr)
class ModuleSession(Session):
"""
A class representing a module's outputs.
Warning
-------
An instance of this class is created for each request and passed as an argument to
the :class:`shiny.modules.Module`'s ``server`` function. For this reason, you
shouldn't need to create instances of this class yourself. Furthermore, you
probably shouldn't need this class for type checking either since it has the same
signature as :class:`shiny.session.Session`.
"""
def __init__(self, ns: str, parent_session: Session):
self._namespaces = copy.copy(parent_session._namespaces)
self._namespaces.append(ns)
self._parent: Session = parent_session
self.input: ModuleInputs = ModuleInputs(ns, parent_session.input)
self.output: ModuleOutputs = ModuleOutputs(ns, parent_session.output)
def __getattr__(self, attr: str) -> Any:
return getattr(self._parent, attr)
class MockModuleSession(ModuleSession):
def __init__(self, ns: str):
self._namespaces = [ns]
P = ParamSpec("P")
R = TypeVar("R")
def module_ui(fn: Callable[P, R]) -> Callable[Concatenate[str, P], R]:
def wrapper(ns: str, *args: P.args, **kwargs: P.kwargs) -> R:
# TODO: what should happen if this is called *inside* of a session? Do we pass in the parent session?
with session_context(MockModuleSession(ns)):
return fn(*args, **kwargs)
return wrapper
def module_server(
fn: Callable[Concatenate[Inputs, Outputs, Session, P], R]
) -> Callable[Concatenate[str, P], R]:
def wrapper(ns: str, *args: P.args, **kwargs: P.kwargs) -> R:
mod_sess = ModuleSession(ns, require_active_session(None))
with session_context(mod_sess):
return fn(mod_sess.input, mod_sess.output, mod_sess, *args, **kwargs)
return wrapper