jc211
/
tractor
forked from goodboy/tractor
1
0
Fork 0
Code Pull Requests Activity

Can we ever really appease mypy?

implicit_runtime
Tyler Goodlet 2021-01-03 11:18:31 -05:00
parent 0bb2163b0c
commit 333ddcf93f
1 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tractor/_state.py
View File

@ -7,7 +7,7 @@ import multiprocessing as mp
import trio
_current_actor: Optional['Actor'] = None # type: ignore
_current_actor: Optional['Actor'] = None # type: ignore # noqa
_runtime_vars: Dict[str, Any] = {
'_debug_mode': False,
'_is_root': False,
@ -15,7 +15,7 @@ _runtime_vars: Dict[str, Any] = {
}
def current_actor(err_on_no_runtime: bool = True) -> 'Actor': # type: ignore
def current_actor(err_on_no_runtime: bool = True) -> 'Actor': # type: ignore # noqa
"""Get the process-local actor instance.
"""
if _current_actor is None and err_on_no_runtime:
@ -24,6 +24,12 @@ def current_actor(err_on_no_runtime: bool = True) -> 'Actor': # type: ignore
return _current_actor
_conc_name_getters = {
'task': trio.lowlevel.current_task,
'actor': current_actor
}
class ActorContextInfo(Mapping):
"Dyanmic lookup for local actor and task names"
_context_keys = ('task', 'actor')
@ -34,12 +40,9 @@ class ActorContextInfo(Mapping):
def __iter__(self):
return iter(self._context_keys)
def __getitem__(self, key: str):
def __getitem__(self, key: str) -> str:
try:
return {
'task': trio.lowlevel.current_task,
'actor': current_actor
}[key]().name
return _conc_name_getters[key]().name # type: ignore
except RuntimeError:
# no local actor/task context initialized yet
return f'no {key} context'