Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add type hints #22

Merged
merged 20 commits into from
Aug 24, 2023
Merged
Show file tree
Hide file tree
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
4 changes: 0 additions & 4 deletions docs/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,10 +31,6 @@
# Be strict about any broken references
nitpicky = True

nitpick_ignore = [
('py:class', 'jaraco.functools.CallableT'),
]

# Include Python intersphinx mapping to prevent failures
# jaraco/skeleton#51
extensions += ['sphinx.ext.intersphinx']
Expand Down
119 changes: 56 additions & 63 deletions jaraco/functools.py → jaraco/functools/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import collections
from __future__ import annotations

import collections.abc
import functools
import inspect
import itertools
Expand All @@ -9,10 +11,7 @@

import more_itertools

from typing import Callable, TypeVar


CallableT = TypeVar("CallableT", bound=Callable[..., object])
_DEFAULT_CACHE_WRAPPER = functools.lru_cache()


def compose(*funcs):
Expand All @@ -39,24 +38,6 @@ def compose_two(f1, f2):
return functools.reduce(compose_two, funcs)


def method_caller(method_name, *args, **kwargs):
"""
Return a function that will call a named method on the
target object with optional positional and keyword
arguments.

>>> lower = method_caller('lower')
>>> lower('MyString')
'mystring'
"""

def call_method(target):
func = getattr(target, method_name)
return func(*args, **kwargs)

return call_method


def once(func):
"""
Decorate func so it's only ever called the first time.
Expand Down Expand Up @@ -99,12 +80,7 @@ def wrapper(*args, **kwargs):
return wrapper


def method_cache(
method: CallableT,
cache_wrapper: Callable[
[CallableT], CallableT
] = functools.lru_cache(), # type: ignore[assignment]
) -> CallableT:
def method_cache(method, cache_wrapper=_DEFAULT_CACHE_WRAPPER):
"""
Wrap lru_cache to support storing the cache data in the object instances.

Expand Down Expand Up @@ -168,26 +144,22 @@ def method_cache(
as ``@property``, which changes the semantics of the function.

See also
--------
http://code.activestate.com/recipes/577452-a-memoize-decorator-for-instance-methods/
for another implementation and additional justification.
"""

def wrapper(self: object, *args: object, **kwargs: object) -> object:
def wrapper(self, *args, **kwargs):
# it's the first call, replace the method with a cached, bound method
bound_method: CallableT = types.MethodType( # type: ignore[assignment]
method, self
)
bound_method = types.MethodType(method, self)
cached_method = cache_wrapper(bound_method)
setattr(self, method.__name__, cached_method)
return cached_method(*args, **kwargs)

# Support cache clear even before cache has been created.
wrapper.cache_clear = lambda: None # type: ignore[attr-defined]
wrapper.cache_clear = lambda: None

return (
_special_method_cache(method, cache_wrapper) # type: ignore[return-value]
or wrapper
)
return _special_method_cache(method, cache_wrapper) or wrapper


def _special_method_cache(method, cache_wrapper):
Expand All @@ -203,12 +175,13 @@ def _special_method_cache(method, cache_wrapper):
"""
name = method.__name__
special_names = '__getattr__', '__getitem__'

if name not in special_names:
return
return None

wrapper_name = '__cached' + name

def proxy(self, *args, **kwargs):
def proxy(self, /, *args, **kwargs):
if wrapper_name not in vars(self):
bound = types.MethodType(method, self)
cache = cache_wrapper(bound)
Expand Down Expand Up @@ -245,7 +218,7 @@ def result_invoke(action):
r"""
Decorate a function with an action function that is
invoked on the results returned from the decorated
function (for its side-effect), then return the original
function (for its side effect), then return the original
result.

>>> @result_invoke(print)
Expand All @@ -269,7 +242,7 @@ def wrapper(*args, **kwargs):
return wrap


def invoke(f, *args, **kwargs):
def invoke(f, /, *args, **kwargs):
"""
Call a function for its side effect after initialization.

Expand Down Expand Up @@ -304,25 +277,26 @@ def invoke(f, *args, **kwargs):
Use functools.partial to pass parameters to the initial call

>>> @functools.partial(invoke, name='bingo')
... def func(name): print("called with", name)
... def func(name): print('called with', name)
called with bingo
"""
f(*args, **kwargs)
return f


def call_aside(*args, **kwargs):
"""
Deprecated name for invoke.
"""
warnings.warn("call_aside is deprecated, use invoke", DeprecationWarning)
return invoke(*args, **kwargs)
def call_aside(f, *args, **kwargs):
"""Deprecated name for invoke."""
warnings.warn(
'`jaraco.functools.call_aside` is deprecated, '
'use `jaraco.functools.invoke` instead',
DeprecationWarning,
stacklevel=2,
)
return invoke(f, *args, **kwargs)


class Throttler:
"""
Rate-limit a function (or other callable)
"""
"""Rate-limit a function (or other callable)."""

def __init__(self, func, max_rate=float('Inf')):
if isinstance(func, Throttler):
Expand All @@ -339,20 +313,20 @@ def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)

def _wait(self):
"ensure at least 1/max_rate seconds from last call"
"""Ensure at least 1/max_rate seconds from last call."""
elapsed = time.time() - self.last_called
must_wait = 1 / self.max_rate - elapsed
time.sleep(max(0, must_wait))
self.last_called = time.time()

def __get__(self, obj, type=None):
def __get__(self, obj, owner=None):
return first_invoke(self._wait, functools.partial(self.func, obj))


def first_invoke(func1, func2):
"""
Return a function that when invoked will invoke func1 without
any parameters (for its side-effect) and then invoke func2
any parameters (for its side effect) and then invoke func2
with whatever parameters were passed, returning its result.
"""

Expand All @@ -363,6 +337,17 @@ def wrapper(*args, **kwargs):
return wrapper


method_caller = first_invoke(
lambda: warnings.warn(
'`jaraco.functools.method_caller` is deprecated, '
'use `operator.methodcaller` instead',
DeprecationWarning,
stacklevel=3,
),
operator.methodcaller,
)


def retry_call(func, cleanup=lambda: None, retries=0, trap=()):
"""
Given a callable func, trap the indicated exceptions
Expand All @@ -371,7 +356,7 @@ def retry_call(func, cleanup=lambda: None, retries=0, trap=()):
to propagate.
"""
attempts = itertools.count() if retries == float('inf') else range(retries)
for attempt in attempts:
for _ in attempts:
try:
return func()
except trap:
Expand All @@ -380,7 +365,7 @@ def retry_call(func, cleanup=lambda: None, retries=0, trap=()):
return func()


def retry(*r_args, **r_kwargs):
def retry(*args, **kwargs):
"""
Decorator wrapper for retry_call. Accepts arguments to retry_call
except func and then returns a decorator for the decorated function.
Expand All @@ -399,7 +384,7 @@ def decorate(func):
@functools.wraps(func)
def wrapper(*f_args, **f_kwargs):
bound = functools.partial(func, *f_args, **f_kwargs)
return retry_call(bound, *r_args, **r_kwargs)
return retry_call(bound, *args, **kwargs)

return wrapper

Expand All @@ -408,7 +393,7 @@ def wrapper(*f_args, **f_kwargs):

def print_yielded(func):
"""
Convert a generator into a function that prints all yielded elements
Convert a generator into a function that prints all yielded elements.

>>> @print_yielded
... def x():
Expand All @@ -424,7 +409,7 @@ def print_yielded(func):

def pass_none(func):
"""
Wrap func so it's not called if its first param is None
Wrap func so it's not called if its first param is None.

>>> print_text = pass_none(print)
>>> print_text('text')
Expand All @@ -433,9 +418,10 @@ def pass_none(func):
"""

@functools.wraps(func)
def wrapper(param, *args, **kwargs):
def wrapper(param, /, *args, **kwargs):
if param is not None:
return func(param, *args, **kwargs)
return None

return wrapper

Expand Down Expand Up @@ -509,7 +495,7 @@ def save_method_args(method):
args_and_kwargs = collections.namedtuple('args_and_kwargs', 'args kwargs')

@functools.wraps(method)
def wrapper(self, *args, **kwargs):
def wrapper(self, /, *args, **kwargs):
attr_name = '_saved_' + method.__name__
attr = args_and_kwargs(args, kwargs)
setattr(self, attr_name, attr)
Expand Down Expand Up @@ -559,6 +545,13 @@ def wrapper(*args, **kwargs):


def identity(x):
"""
Return the argument.

>>> o = object()
>>> identity(o) is o
True
"""
return x


Expand All @@ -580,7 +573,7 @@ def bypass_when(check, *, _op=identity):

def decorate(func):
@functools.wraps(func)
def wrapper(param):
def wrapper(param, /):
return param if _op(check) else func(param)

return wrapper
Expand Down
Loading
Loading