Skip to content
Merged
2 changes: 1 addition & 1 deletion .github/workflows/lint.yml
Original file line number Diff line number Diff line change
Expand Up @@ -47,4 +47,4 @@ jobs:

- name: Run mypy for tests
shell: bash
run: mypy tests
run: mypy tests --exclude tests/typing
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -21,3 +21,4 @@ uv.lock
node_modules
mutants
CLAUDE.md
AGENTS.md
81 changes: 72 additions & 9 deletions pristan/common_types.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,76 @@
from typing import Callable, Dict, List, Optional, TypeVar, Union
import sys
from typing import (
Any,
Callable,
Dict,
Iterator,
List,
Optional,
Protocol,
Tuple,
TypeVar,
Union,
overload,
)

try:
# TODO: ParamSpec appeared with python 3.10, so delete this try-except if python 3.9 deleted from the matrix
from typing import ParamSpec # type: ignore[attr-defined, unused-ignore]
except ImportError: # pragma: no cover
from typing_extensions import ParamSpec # type: ignore[assignment, unused-ignore]
if sys.version_info >= (3, 10):
from typing import ParamSpec # pragma: no cover
else:
from typing_extensions import ParamSpec # pragma: no cover

SlotParameters = ParamSpec('SlotParameters')

SlotPapameters = ParamSpec('SlotPapameters')
PluginResult = TypeVar('PluginResult')
SlotCallResult = TypeVar('SlotCallResult')
SlotCallResultCovariant = TypeVar('SlotCallResultCovariant', covariant=True) # noqa: PLC0105
PluginResultCovariant = TypeVar('PluginResultCovariant', covariant=True) # noqa: PLC0105

SlotResult = Optional[Union[List[PluginResult], Dict[str, PluginResult]]]
SlotFunction = Callable[SlotPapameters, Optional[Union[List[PluginResult], Dict[str, PluginResult]]]] # type: ignore[valid-type, misc, unused-ignore]
PluginFunction = Callable[SlotPapameters, PluginResult] # type: ignore[valid-type, misc, unused-ignore]
SlotFunction = Callable[SlotParameters, SlotCallResult]
PluginFunction = Callable[SlotParameters, PluginResult]


class PluginProtocol(Protocol[SlotParameters, PluginResultCovariant]): # pragma: no cover
name: str
requested_name: str

def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> PluginResultCovariant: ...


class BaseSlotViewProtocol(Protocol[SlotParameters, SlotCallResultCovariant, PluginResultCovariant]): # pragma: no cover
def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotCallResultCovariant: ...

def __iter__(self) -> Iterator[PluginProtocol[SlotParameters, PluginResultCovariant]]: ...

def __bool__(self) -> bool: ...

def __len__(self) -> int: ...


class SlotSelectionProtocol(BaseSlotViewProtocol[SlotParameters, SlotCallResultCovariant, PluginResultCovariant], Protocol[SlotParameters, SlotCallResultCovariant, PluginResultCovariant]):
pass


class SlotProtocol(BaseSlotViewProtocol[SlotParameters, SlotCallResultCovariant, PluginResult], Protocol[SlotParameters, SlotCallResultCovariant, PluginResult]): # pragma: no cover
@overload
def plugin(self, plugin_function_or_name: Optional[str] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[[Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]]: ...

@overload
def plugin(self, plugin_function_or_name: Callable[SlotParameters, PluginResult], unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[SlotParameters, PluginResult]: ...

def keys(self) -> Tuple[str, ...]: ...

def __getitem__(self, key: str) -> SlotSelectionProtocol[SlotParameters, SlotCallResultCovariant, PluginResult]: ...

def __contains__(self, item: object) -> bool: ...


class SlotDecoratorProtocol(Protocol): # pragma: no cover
@overload
def __call__(self, function: Callable[SlotParameters, List[PluginResult]], /) -> SlotProtocol[SlotParameters, List[PluginResult], PluginResult]: ...

@overload
def __call__(self, function: Callable[SlotParameters, Dict[str, PluginResult]], /) -> SlotProtocol[SlotParameters, Dict[str, PluginResult], PluginResult]: ...

@overload
def __call__(self, function: Callable[SlotParameters, None], /) -> SlotProtocol[SlotParameters, None, Any]: ...
12 changes: 6 additions & 6 deletions pristan/components/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,15 @@
from printo import repred
from simtypes import check

from pristan.common_types import PluginFunction, PluginResult, SlotPapameters
from pristan.common_types import PluginFunction, PluginResult, SlotParameters
from pristan.components.slot_code_representer import sentinel as return_type_sentinel
from pristan.errors import NumberOfCallsError


@repred(positionals=['name'])
class Plugin(Generic[PluginResult]):
# TODO: consider to delete this "type: ignore" if python 3.9 deleted from the matrix
def __init__(self, name: str, plugin_function: PluginFunction[SlotPapameters, PluginResult], expected_result_type: Union[InnerNoneType, Type[Any]], type_check: bool, unique: bool, run_once: bool = False) -> None: # type: ignore[type-arg, unused-ignore] # noqa: PLR0913
def __init__(self, name: str, plugin_function: PluginFunction[SlotParameters, PluginResult], expected_result_type: Union[InnerNoneType, Type[Any]], type_check: bool, unique: bool, run_once: bool = False) -> None: # noqa: PLR0913
self.plugin_function = plugin_function
self.requested_name = name
self.name = name
Expand All @@ -24,20 +24,20 @@ def __init__(self, name: str, plugin_function: PluginFunction[SlotPapameters, Pl
self.call_count = 0
self.lock = Lock()

def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> PluginResult:
def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> PluginResult:
if self.run_once:
with self.lock:
if self.call_count:
raise NumberOfCallsError(f'A limit of 1 has been set on the number of calls for plugin "{self.name}". And this plugin has already been called previously.')
self.call_count += 1

# TODO: try to delete this "type: ignore" comments if python 3.8 deleted from CI
result = self.plugin_function(*args, **kwargs) # type: ignore[arg-type, unused-ignore]
result = self.plugin_function(*args, **kwargs) # type: ignore[arg-type]

if self.type_check and self.expected_result_type is not return_type_sentinel and not check(result, self.expected_result_type, strict=True): # type: ignore[arg-type]
raise TypeError(f'The type {type(result).__name__} of the plugin\'s "{self.name}" return value {result!r} does not match the expected type {self._get_class_name(self.expected_result_type)}.') # type: ignore[union-attr, unused-ignore]
raise TypeError(f'The type {type(result).__name__} of the plugin\'s "{self.name}" return value {result!r} does not match the expected type {self._get_class_name(self.expected_result_type)}.')

return result # type: ignore[no-any-return, unused-ignore]
return result

def set_name(self, name: str) -> None:
self.name = name
Expand Down
4 changes: 2 additions & 2 deletions pristan/components/plugins_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ class PluginsGroup(Generic[PluginResult]):
All collection operations are totally not thread-safe.
"""

def __init__(self, caller: 'SlotCaller', plugins: Optional[List[Plugin[PluginResult]]] = None) -> None: # type: ignore[name-defined] # noqa: F821
def __init__(self, caller: 'SlotCaller[PluginResult]', plugins: Optional[List[Plugin[PluginResult]]] = None) -> None: # type: ignore[name-defined] # noqa: F821
self.caller = caller
self.plugins: List[Plugin[PluginResult]] = []
self.plugins_by_requested_names: DefaultDict[str, List[Plugin[PluginResult]]] = defaultdict(list)
Expand Down Expand Up @@ -57,7 +57,7 @@ def __contains__(self, item: Any) -> bool:

raise TypeError('Checking for inclusion is only possible for strings of a valid format or for plugin objects.')

def __getitem__(self, key: str) -> 'CallerWithPlugins': # type: ignore[name-defined] # noqa: F821
def __getitem__(self, key: str) -> 'CallerWithPlugins[PluginResult]': # type: ignore[name-defined] # noqa: F821
from pristan.components.slot_caller import CallerWithPlugins # noqa: PLC0415

if isinstance(key, str):
Expand Down
36 changes: 18 additions & 18 deletions pristan/components/slot.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
try:
from importlib_metadata import ( # type: ignore[import-not-found, unused-ignore]
entry_points, # type: ignore[import-not-found, unused-ignore]
entry_points,
)
except ImportError: # type: ignore[assignment, unused-ignore] # pragma: no cover
except ImportError: # pragma: no cover
from importlib.metadata import ( # type: ignore[assignment, unused-ignore]
entry_points, # type: ignore[assignment, unused-ignore]
entry_points,
)

from threading import RLock
Expand All @@ -26,9 +26,10 @@

from pristan.common_types import (
PluginFunction,
PluginProtocol,
PluginResult,
SlotFunction,
SlotPapameters,
SlotParameters,
SlotResult,
)
from pristan.components.plugin import Plugin
Expand All @@ -55,7 +56,7 @@
},
)
class Slot(Generic[PluginResult]):
def __init__(self, slot_function: SlotFunction[SlotPapameters, SlotResult[PluginResult]], signature: Optional[str], slot_name: Optional[str], max: Optional[int], type_check: bool, entrypoint_group: str) -> None: # type: ignore[type-arg, unused-ignore] # noqa: PLR0913, A002
def __init__(self, slot_function: SlotFunction[SlotParameters, SlotResult[PluginResult]], signature: Optional[str], slot_name: Optional[str], max: Optional[int], type_check: bool, entrypoint_group: str) -> None: # noqa: PLR0913, A002
if max is not None and max < 0:
raise ValueError('The maximum number of plugins cannot be less than zero.')

Expand All @@ -78,15 +79,15 @@ def __init__(self, slot_function: SlotFunction[SlotPapameters, SlotResult[Plugin
self.plugins: PluginsGroup[PluginResult] = PluginsGroup(self.caller)
self.backed_caller = CallerWithPlugins(self.caller, self.plugins.plugins)

self._compare_signatures(self.slot_function, self.slot_function) # type: ignore[arg-type, unused-ignore]
self._compare_signatures(self.slot_function, self.slot_function) # type: ignore[arg-type]

self.loaded = False

def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> SlotResult[PluginResult]:
def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotResult[PluginResult]:
self._load_entrypoints()
return self.backed_caller(*args, **kwargs)

def __iter__(self) -> Generator[Plugin[PluginResult], None, None]:
def __iter__(self) -> Generator[PluginProtocol[SlotParameters, PluginResult], None, None]:
self._load_entrypoints()
yield from self.plugins

Expand All @@ -101,18 +102,17 @@ def __len__(self) -> int:
return len(self.plugins)

@overload
def plugin(self, plugin_function_or_name: Optional[str], unique: bool = False, engine: Optional[Union[List[str], str]] = None) -> Callable[[PluginFunction[SlotPapameters, PluginResult]], PluginFunction[SlotPapameters, PluginResult]]: # type: ignore[type-arg, unused-ignore]
def plugin(self, plugin_function_or_name: Optional[str] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[[Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]]:
... # pragma: no cover

@overload
def plugin(self, plugin_function_or_name: PluginFunction[SlotPapameters, PluginResult], unique: bool = False, engine: Optional[Union[List[str], str]] = None) -> PluginFunction[SlotPapameters, PluginResult]: # type: ignore[type-arg, unused-ignore]
... # pragma: no cover
def plugin(self, plugin_function_or_name: Callable[SlotParameters, PluginResult], unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Callable[SlotParameters, PluginResult]: ... # pragma: no cover

def plugin(self, plugin_function_or_name: Optional[Union[PluginFunction[SlotPapameters, PluginResult], str]] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Union[Callable[[PluginFunction[SlotPapameters, PluginResult]], PluginFunction[SlotPapameters, PluginResult]], PluginFunction[SlotPapameters, PluginResult]]: # type: ignore[type-arg, unused-ignore]
def plugin(self, plugin_function_or_name: Optional[Union[PluginFunction[SlotParameters, PluginResult], str]] = None, unique: bool = False, engine: Optional[Union[List[str], str]] = None, run_once: bool = False) -> Union[Callable[[Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]], Callable[SlotParameters, PluginResult]]:
if isinstance(plugin_function_or_name, str):
if not plugin_function_or_name.isidentifier():
raise ValueError('The plugin name must be a valid Python identifier.')
get_plugin_name: Callable[[PluginFunction[SlotPapameters, PluginResult]], str] = lambda function: plugin_function_or_name # type: ignore[type-arg, unused-ignore] # noqa: E731, ARG005
get_plugin_name: Callable[[PluginFunction[SlotParameters, PluginResult]], str] = lambda function: plugin_function_or_name # noqa: E731, ARG005

elif callable(plugin_function_or_name):
get_plugin_name = lambda function: plugin_function_or_name.__name__ # noqa: E731, ARG005
Expand All @@ -123,9 +123,9 @@ def plugin(self, plugin_function_or_name: Optional[Union[PluginFunction[SlotPapa
else:
raise TypeError('Only a function or plugin name followed by a function can be passed to the decorator.')

def decorator(plugin_function: PluginFunction[SlotPapameters, PluginResult]) -> PluginFunction[SlotPapameters, PluginResult]: # type: ignore[type-arg, unused-ignore]
def decorator(plugin_function: Callable[SlotParameters, PluginResult]) -> Callable[SlotParameters, PluginResult]:
# TODO: consider to delete this "type: ignore" if python 3.8 deleted from the matrix
self._compare_signatures(self.slot_function, plugin_function) # type: ignore[arg-type, unused-ignore]
self._compare_signatures(self.slot_function, plugin_function) # type: ignore[arg-type]
self._add_plugin(get_plugin_name(plugin_function), plugin_function, unique, engine, run_once)
return plugin_function

Expand All @@ -145,8 +145,8 @@ def _load_entrypoints(self) -> None:
point.load()
self.loaded = True

def _add_plugin(self, name: str, function: PluginFunction[SlotPapameters, PluginResult], unique: bool, engine: Optional[Union[str, List[str]]], run_once: bool) -> None: # type: ignore[type-arg, unused-ignore]
plugin: Plugin = Plugin(name, function, self.code_representation.returning_type, self.type_check, unique, run_once) # type: ignore[type-arg]
def _add_plugin(self, name: str, function: PluginFunction[SlotParameters, PluginResult], unique: bool, engine: Optional[Union[str, List[str]]], run_once: bool) -> None:
plugin: Plugin[PluginResult] = Plugin(name, function, self.code_representation.returning_type, self.type_check, unique, run_once)

with self.lock:
if len(self.plugins) == self.max_number_of_plugins:
Expand All @@ -161,7 +161,7 @@ def _add_plugin(self, name: str, function: PluginFunction[SlotPapameters, Plugin
self.plugins.delete_last_by_name(name)
raise PrimadonnaPluginError(f'Plugin "{other_plugin.name}" claims to be unique, but there are other plugins with the same name.')

def _compare_signatures(self, slot_function: SlotFunction[SlotPapameters, SlotResult[PluginResult]], plugin_function: PluginFunction[SlotPapameters, PluginResult]) -> None: # type: ignore[type-arg, unused-ignore]
def _compare_signatures(self, slot_function: SlotFunction[SlotParameters, SlotResult[PluginResult]], plugin_function: PluginFunction[SlotParameters, PluginResult]) -> None:
if self.signature is not None:
PossibleCallMatcher(self.signature).match(plugin_function, raise_exception=True)
elif not PossibleCallMatcher.from_callable(slot_function) & PossibleCallMatcher.from_callable(plugin_function):
Expand Down
10 changes: 5 additions & 5 deletions pristan/components/slot_caller.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from pristan.common_types import (
PluginResult,
SlotFunction,
SlotPapameters,
SlotParameters,
SlotResult,
)
from pristan.components.plugin import Plugin
Expand All @@ -18,13 +18,13 @@
@repred
class SlotCaller(Generic[PluginResult]):
# TODO: consider to delete this "type: ignore" if python 3.8 deleted from the matrix
def __init__(self, code_representation: SlotCodeRepresenter, slot_name: Optional[str], slot_function: SlotFunction[SlotPapameters, SlotResult[PluginResult]], type_check: bool) -> None: # type: ignore[type-arg, unused-ignore]
def __init__(self, code_representation: SlotCodeRepresenter, slot_name: Optional[str], slot_function: SlotFunction[SlotParameters, SlotResult[PluginResult]], type_check: bool) -> None:
self.code_representation = code_representation
self.slot_name = slot_name
self.slot_function = slot_function
self.type_check = type_check

def __call__(self, plugins: Union[PluginsGroup[PluginResult], List[Plugin[PluginResult]]], *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> SlotResult[PluginResult]: # type: ignore[return]
def __call__(self, plugins: Union[PluginsGroup[PluginResult], List[Plugin[PluginResult]]], *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotResult[PluginResult]: # type: ignore[return]
if not self.code_representation.is_empty and not plugins:
if self.code_representation.returns_list:
if self.code_representation.returning_type is return_type_sentinel:
Expand All @@ -40,7 +40,7 @@ def __call__(self, plugins: Union[PluginsGroup[PluginResult], List[Plugin[Plugin
returns_type = self.code_representation.returning_type

# TODO: consider to delete this "type: ignore" if python 3.9 deleted from the matrix
result: SlotResult[PluginResult] = Plugin(self.slot_name if self.slot_name is not None else self.slot_function.__name__, self.slot_function, returns_type, self.type_check, False)(*args, **kwargs) # type: ignore[assignment, unused-ignore]
result: SlotResult[PluginResult] = Plugin(self.slot_name if self.slot_name is not None else self.slot_function.__name__, self.slot_function, returns_type, self.type_check, False)(*args, **kwargs)

if self.code_representation.returning_type is return_type_sentinel and not self.code_representation.returns_dict and not self.code_representation.returns_list:
result = None
Expand All @@ -63,7 +63,7 @@ def __init__(self, caller: SlotCaller[PluginResult], plugins: List[Plugin[Plugin
self.caller = caller
self.plugins = plugins

def __call__(self, *args: SlotPapameters.args, **kwargs: SlotPapameters.kwargs) -> SlotResult[PluginResult]:
def __call__(self, *args: SlotParameters.args, **kwargs: SlotParameters.kwargs) -> SlotResult[PluginResult]:
return self.caller(self.plugins, *args, **kwargs)

def __iter__(self) -> Generator[Plugin[PluginResult], None, None]:
Expand Down
Loading
Loading