generic_func.register(cls, func) -> func Registers a new implementation for the given *cls* on a *generic_func*.
(cls, func=None)
| 941 | all(isinstance(arg, type) for arg in cls.__args__)) |
| 942 | |
| 943 | def register(cls, func=None): |
| 944 | """generic_func.register(cls, func) -> func |
| 945 | |
| 946 | Registers a new implementation for the given *cls* on a *generic_func*. |
| 947 | |
| 948 | """ |
| 949 | nonlocal cache_token |
| 950 | if _is_valid_dispatch_type(cls): |
| 951 | if func is None: |
| 952 | return lambda f: register(cls, f) |
| 953 | else: |
| 954 | if func is not None: |
| 955 | raise TypeError( |
| 956 | f"Invalid first argument to `register()`. " |
| 957 | f"{cls!r} is not a class or union type." |
| 958 | ) |
| 959 | ann = getattr(cls, '__annotate__', None) |
| 960 | if ann is None: |
| 961 | raise TypeError( |
| 962 | f"Invalid first argument to `register()`: {cls!r}. " |
| 963 | f"Use either `@register(some_class)` or plain `@register` " |
| 964 | f"on an annotated function." |
| 965 | ) |
| 966 | func = cls |
| 967 | |
| 968 | # only import typing if annotation parsing is necessary |
| 969 | from typing import get_type_hints |
| 970 | from annotationlib import Format, ForwardRef |
| 971 | argname, cls = next(iter(get_type_hints(func, format=Format.FORWARDREF).items())) |
| 972 | if not _is_valid_dispatch_type(cls): |
| 973 | if isinstance(cls, UnionType): |
| 974 | raise TypeError( |
| 975 | f"Invalid annotation for {argname!r}. " |
| 976 | f"{cls!r} not all arguments are classes." |
| 977 | ) |
| 978 | elif isinstance(cls, ForwardRef): |
| 979 | raise TypeError( |
| 980 | f"Invalid annotation for {argname!r}. " |
| 981 | f"{cls!r} is an unresolved forward reference." |
| 982 | ) |
| 983 | else: |
| 984 | raise TypeError( |
| 985 | f"Invalid annotation for {argname!r}. " |
| 986 | f"{cls!r} is not a class." |
| 987 | ) |
| 988 | |
| 989 | if isinstance(cls, UnionType): |
| 990 | for arg in cls.__args__: |
| 991 | registry[arg] = func |
| 992 | else: |
| 993 | registry[cls] = func |
| 994 | if cache_token is None and hasattr(cls, '__abstractmethods__'): |
| 995 | cache_token = get_cache_token() |
| 996 | dispatch_cache.clear() |
| 997 | return func |
| 998 | |
| 999 | def wrapper(*args, **kw): |
| 1000 | if not args: |
nothing calls this directly
no test coverage detected
searching dependent graphs…