(
self,
data: Any,
*args: Any,
_CHAIN: "type[chain[_PayloadRegistryItem]]" = chain,
**kwargs: Any,
)
| 97 | self._normal_lookup: dict[Any, PayloadType] = {} |
| 98 | |
| 99 | def get( |
| 100 | self, |
| 101 | data: Any, |
| 102 | *args: Any, |
| 103 | _CHAIN: "type[chain[_PayloadRegistryItem]]" = chain, |
| 104 | **kwargs: Any, |
| 105 | ) -> "Payload": |
| 106 | if self._first: |
| 107 | for factory, type_ in self._first: |
| 108 | if isinstance(data, type_): |
| 109 | return factory(data, *args, **kwargs) |
| 110 | # Try the fast lookup first |
| 111 | if lookup_factory := self._normal_lookup.get(type(data)): |
| 112 | return lookup_factory(data, *args, **kwargs) |
| 113 | # Bail early if its already a Payload |
| 114 | if isinstance(data, Payload): |
| 115 | return data |
| 116 | # Fallback to the slower linear search |
| 117 | for factory, type_ in _CHAIN(self._normal, self._last): |
| 118 | if isinstance(data, type_): |
| 119 | return factory(data, *args, **kwargs) |
| 120 | raise LookupError() |
| 121 | |
| 122 | def register( |
| 123 | self, factory: PayloadType, type: Any, *, order: Order = Order.normal |
no test coverage detected