Register a typeid with the manager type
(cls, typeid, callable=None, proxytype=None, exposed=None,
method_to_typeid=None, create_method=True)
| 700 | |
| 701 | @classmethod |
| 702 | def register(cls, typeid, callable=None, proxytype=None, exposed=None, |
| 703 | method_to_typeid=None, create_method=True): |
| 704 | ''' |
| 705 | Register a typeid with the manager type |
| 706 | ''' |
| 707 | if '_registry' not in cls.__dict__: |
| 708 | cls._registry = cls._registry.copy() |
| 709 | |
| 710 | if proxytype is None: |
| 711 | proxytype = AutoProxy |
| 712 | |
| 713 | exposed = exposed or getattr(proxytype, '_exposed_', None) |
| 714 | |
| 715 | method_to_typeid = method_to_typeid or \ |
| 716 | getattr(proxytype, '_method_to_typeid_', None) |
| 717 | |
| 718 | if method_to_typeid: |
| 719 | for key, value in list(method_to_typeid.items()): # isinstance? |
| 720 | assert type(key) is str, '%r is not a string' % key |
| 721 | assert type(value) is str, '%r is not a string' % value |
| 722 | |
| 723 | cls._registry[typeid] = ( |
| 724 | callable, exposed, method_to_typeid, proxytype |
| 725 | ) |
| 726 | |
| 727 | if create_method: |
| 728 | def temp(self, /, *args, **kwds): |
| 729 | util.debug('requesting creation of a shared %r object', typeid) |
| 730 | token, exp = self._create(typeid, *args, **kwds) |
| 731 | proxy = proxytype( |
| 732 | token, self._serializer, manager=self, |
| 733 | authkey=self._authkey, exposed=exp |
| 734 | ) |
| 735 | conn = self._Client(token.address, authkey=self._authkey) |
| 736 | dispatch(conn, None, 'decref', (token.id,)) |
| 737 | return proxy |
| 738 | temp.__name__ = typeid |
| 739 | setattr(cls, typeid, temp) |
| 740 | |
| 741 | # |
| 742 | # Subclass of set which get cleared after a fork |
no test coverage detected