Skip to content
This repository has been archived by the owner on Nov 1, 2024. It is now read-only.

Generalize Dispatcher class #505

Open
wants to merge 1 commit into
base: fbsync
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 16 additions & 13 deletions torcharrow/dispatcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

from typing import Callable, ClassVar, Dict, Tuple
from typing import Callable, ClassVar, Dict, Generic, Tuple, TypeVar

# ---------------------------------------------------------------------------
# column factory (class methods only!)
Expand All @@ -13,21 +13,24 @@
Typecode = str # one of dtype.typecode


class Dispatcher:
T = TypeVar("T")

# singelton, append only, registering is idempotent
_calls: ClassVar[Dict[Tuple[Typecode, Device], Callable]] = {}

@classmethod
def register(cls, key: Tuple[Typecode, Device], call: Callable):
# key is tuple: (device,typecode)
if key in Dispatcher._calls:
if call == Dispatcher._calls[key]:
class _Dispatcher(Generic[T]):
def __init__(self):
# append only, registering is idempotent
self._calls: Dict[T, Callable] = {}

def register(self, key: T, call: Callable):
if key in self._calls:
if call == self._calls[key]:
return
else:
raise ValueError("keys for calls can only be registered once")
Dispatcher._calls[key] = call
self._calls[key] = call

def lookup(self, key: T) -> Callable:
return self._calls[key]


@classmethod
def lookup(cls, key: Tuple[Typecode, Device]) -> Callable:
return Dispatcher._calls[key]
Dispatcher: _Dispatcher[Tuple[Typecode, Device]] = _Dispatcher()