Create a new dict for the current thread, and return it.
(self)
| 40 | return self.dicts[id(thread)][1] |
| 41 | |
| 42 | def create_dict(self): |
| 43 | """Create a new dict for the current thread, and return it.""" |
| 44 | localdict = {} |
| 45 | key = self.key |
| 46 | thread = current_thread() |
| 47 | idt = id(thread) |
| 48 | def local_deleted(_, key=key): |
| 49 | # When the localimpl is deleted, remove the thread attribute. |
| 50 | thread = wrthread() |
| 51 | if thread is not None: |
| 52 | del thread.__dict__[key] |
| 53 | def thread_deleted(_, idt=idt): |
| 54 | # When the thread is deleted, remove the local dict. |
| 55 | # Note that this is suboptimal if the thread object gets |
| 56 | # caught in a reference loop. We would like to be called |
| 57 | # as soon as the OS-level thread ends instead. |
| 58 | local = wrlocal() |
| 59 | if local is not None: |
| 60 | local.dicts.pop(idt) |
| 61 | wrlocal = ref(self, local_deleted) |
| 62 | wrthread = ref(thread, thread_deleted) |
| 63 | thread.__dict__[key] = wrlocal |
| 64 | self.dicts[idt] = wrthread, localdict |
| 65 | return localdict |
| 66 | |
| 67 | |
| 68 | @contextmanager |
no test coverage detected