Py-Gems Core API

pygems.core.functools

class pygems.core.functools.MultiMethod(name)

Manage function versions distinguieshed by argument types

Based on Guido van Rossum’s Five-minute Multimethods in Python

register(types, function): Register function signature.

class pygems.core.functools.MultiMethodRegistry

Map function to multimethod

classmethod register(types: list, function: Callable) → pygems.core.functools.MultiMethod: Register a function as mutimethod.

pygems.core.functools.multimethod(*types)

Decorator to register a multimethod signature

You can specify function to be mapped to call signature. For example:

We can define one implementation for the scenario when the function receives an int argument.

We can also define another implementation for the scenario when the function receives a str argument:

>>> @multimethod(int)
... def myfunc(a:int):
...     print(a)
>>> @multimethod(str)
... def myfunc(name: str):
...     print(f'Hello, {name}')

Trying to register duplicate signature results in TypeError error:

>>> @multimethod(str)
... def myfunc(s:str):
...     print(s)
Traceback (most recent call last):
...
TypeError: duplicate registration

Calling myfunc with int argument:

>>> myfunc(12)
12

Calling myfunc with str argument:

>>> myfunc('Ivan')
Hello, Ivan

Trying to call with unregistered signature results in error: >>> myfunc([]) Traceback (most recent call last): … TypeError: no match

pygems.core.namespace

class pygems.core.namespace.Namespace(*args, **kwargs)

Namespace class

asdict(sorted_keys=False)

Get dictionary representation of the data.

Only custom attributes are exported:

>>> ns = Namespace(name='Joan')
>>> ns.asdict()
{'name': 'Joan'}

Set the sorted_keys parameter to True to get dictionary withsorted keys. Useful for docstring tests.

>>> ns = Namespace(name='Joan', city='Arc')
>>> ns.asdict(sorted_keys=True)
{'city': 'Arc', 'name': 'Joan'}

update(*args, **kwargs)

Update the object attributes

Positional arguments are treated as dictionary:

>>> ns = Namespace()
>>> ns.update({'name':'John'})
>>> ns.name
'John'

Keyword arguments are treated as dictionary elements:

>>> ns = Namespace()
>>> ns.update(name='Jane')
>>> ns.name
'Jane'

Positional and keyword arguments can be combined:

>>> address = {'city':'London'}
>>> name = {'first': 'Sponge'}
>>> ns = Namespace()
>>> ns.update(name, address, age=23)
>>> ns.asdict(sorted_keys=True)
{'age': 23, 'city': 'London', 'first': 'Sponge'}

pygems.core.plugin

Tools for implementing plugin architecture in Python applicaitons.

class pygems.core.plugin.PluginCollection(plugin_class=None)

Plugin Collection

Here is example

We define an event listener class which listens for load events. When load event notification is received the on_load event is called which simply prints the arguments passed to it.

>>> class EventListener(BaseEventListener):
...    def on_load(self, *args, **kwargs):
...       print(*args)

We create a PluginCollection instance which accepts only BaseEventListener instances:

>>> plugins = PluginCollection(BaseEventListener)

We register an instance of our event listener:

>>> _ = plugins.append(EventListener())

Notifying plugins on load event will call the on_load method of our plugin instance:

>>> _ = plugins.notify('load', 'myfile.txt')
myfile.txt

Events which are not expected are ignored. In other words event xyz is ignored if our event listener doesn’t impmlement on_xyz method:

>>> _ = plugins.notify('shutdown')

Most methods return the plugin collection instance which makes it possible to write code like this:

>>> plugins = PluginCollection().append(EventListener()).notify('init')
>>> _ = plugins.notify('load', 'bigfile.dat')
bigfile.dat

append(*plugins)

Appending a plugin stores it in collection:

>>> plugins = PluginCollection()
>>> _ = plugins.append(print)
>>> plugins._plugins
[<built-in function print>]

Multiple plugins can be appended at the same time. Order is preserved:

>>> plugins = PluginCollection()
>>> _ = plugins.append(print, dir)
>>> plugins._plugins
[<built-in function print>, <built-in function dir>]

Appending same plugin more than once is ignored:

>>> plugins = PluginCollection()
>>> _ = plugins.append(print, dir)
>>> _ = plugins.append(print)
>>> plugins._plugins
[<built-in function print>, <built-in function dir>]

Tryhing to append non-callable plugin raises error:

>>> plugins = PluginCollection()
>>> plugins.append(3)
Traceback (most recent call last):
...
AssertionError: plugin argument should be callable

You can restrict type of plugins so that adding callable which is not instance of the given class raises assertion error. Note that instances still should be callable which in Python means that class implements the __call__ method:

>>> class BasePlugin:
...    def __call__(self, *args, **kwargs):
...       print(*args, **kwargs)
>>> plugins = PluginCollection(BasePlugin)
>>> _ = plugins.append(BasePlugin())
>>> _ = plugins.append(print)
Traceback (most recent call last):
   ...
AssertionError: plugin argument should be intance of <class 'core.plugin.BasePlugin'>

insert(*plugins)

Insert one or more plugins at the beginning of the collection.

>>> plugins = PluginCollection().append(print)
>>> plugins.insert(dir)._plugins
[<built-in function dir>, <built-in function print>]

Plugins are validated:

>>> plugins.insert(3)
Traceback (most recent call last):
   ...
AssertionError: plugin argument should be callable

notify(*args, **kwargs)

Runs all plugins from collection passing positional and keyword arguments:

>>> plugins = PluginCollection()
>>> _ = plugins.append(print)
>>> _ = plugins.notify('click', 3, sep='|')
click|3

remove(*plugins)

Remove one or more plugins

>>> plugins = PluginCollection().append(print, dir)
>>> plugins.remove(print)._plugins
[<built-in function dir>]

Multiple plugins could be removed at the same time:

>>> plugins = PluginCollection().append(print, dir)
>>> plugins.remove(print, dir)._plugins
[]

pygems.core.shortcuts

Various functions

pygems.core.shortcuts.drop_fields(representation: dict, fields: Union[str, list], error_handler=None): Remove one or more keys from a dictionary

pygems.core.shortcuts.get_ignore_errors(errors: list): Return error handler which ignores error from one or more classes and raises other errors

pygems.core.shortcuts.getattr_nested(obj: Any, attr: str, default=None): Retrieve attribute from nested objects using dot notation

pygems.core.timer

class pygems.core.timer.StringMessageCallback(message_template=None, message_func=None, arg_separator=None)

Simple string message callback for the Timer’s stop() method

Example 1::

>>> timer = Timer(name='MyTimer', stop_func=StringMessageCallback())
>>> timer.stop('loaded')          
MyTimer: 2.7700000000019376e-05s loaded
>>> timer.stop('transferred')     
MyTimer: 5.060000000001175e-05s transferred

Example 2: Custom template could be used::

>>> timer = Timer(name='MyTimer', stop_func=StringMessageCallback(message_template='{args[0]} at {timer.elapsed}s'))
>>> timer.stop('Loaded')          
Loaded at 2.7700000000019376e-05s
>>> timer.stop('Transfered')      
Transfered at 5.060000000001175e-05s

arg_separator: str = ' ': Separator to use when joining the callback arguments before passing to message_func

message_func: Callable: Function to be used to print the output. Default is the print function

message_template: str = '{timer.name}: {timer.elapsed}s {args_str}': Template to use to format the output string

class pygems.core.timer.Timer(name=None, time_func: Optional[Callable[[], float]] = None, stop_func=None)

Generic Timer class.

namestr: Name of the timer. Available as name attribute.
time_funccallback, optional: Function to be used to retrieve current time (timeit.default_timer is used by default). Also available as time_func property.
stop_funccallback, optional: Function to be called when the stop() method is called. Also available a stop_func property.

The Timer class is useful for capturing the execution time of long-running operations.

Examples:

In this example we will create a timer which prints the elapsed time each time the stop() method is called using a callback function stop_func.

Another callback function time_func is used to retrieve the current time.

>>> from pygems.core.timer import *
>>> from functools import partial
>>> time_func = partial([1, 9.1234, 30.789].pop, 0)
>>> stop_func = lambda t, *args: print(f'{t.name}:', *args, f'{round(t.elapsed,6)}s')

>>> timer = Timer(name='MyTimer', time_func=time_func, stop_func=stop_func)

Call the stop() method, passing a stop point name prints the formatted message, passing the stop point name::

>>> timer.stop('load finished at')
MyTimer: load finished at 8.1234s

>>> timer.stop('parse finished at')
MyTimer: parse finished at 29.789s

In this example a fake time_func timer function and custom stop_func callback are used for illustrative purposes.

# We use partial to make our fake timer function.
from functools import partial

# A fake timer function which will pop and return the first element of a list
# each time it has been called.
time_func = partial([1, 9.1234, 30.789].pop, 0)

# Custom stop_func which prints a formatted message.
stop_func = lambda t, *args: print(f'{t.name}:', *args, f'{round(t.elapsed,6)}s')

property elapsed: float

Returns elapsed time between sarted and stopped or started and current time.

If timer is not stopped, elapsed returns current - started time::

>>> from functools import partial
>>> timer = Timer(time_func=partial([1,5].pop, 0))
>>> timer.elapsed
4

If timer is stopped, elapsed returns stopped time - started time::

>>> timer = Timer(time_func=partial([1,3,5].pop, 0))
>>> timer.stop()
>>> timer.stopped_at
3
>>> timer.elapsed
2

start()

Starts the timer by setting the started_at to current time and stopped_at to None.

Example::

>>> from functools import partial
>>> timer = Timer(time_func=partial([10,20,30].pop, 0))

started_at is set during intialization::

>>> timer.started_at
10

When we stop the timer, stopped_at is set::

>>> timer.stop()
>>> timer.stopped_at
20

Starting a stopped timer sets started_at to current time and stopped_at to None::

>>> timer.start()
>>> timer.started_at
30
>>> timer.stopped_at is None
True

stop(*args)

Stop the timer by setting the stopped_at attribute to current time

When the stop() method is called:

Set the stopped_at attribute to the current time
Determine if stop_func attribute is set to a callback
If the callback is set, call it passing the Timer object as a first argument, followed by all arguments, received by the stop() method.

Example::

>>> from functools import partial
>>> timer = Timer(time_func=partial([1,9, 50].pop, 0))

When timer is not stopped, stopped time stopped_at is None::

>>> timer.stopped_at is None
True

After we call the stop() method, the stopped_at time is set to current time::

>>> timer.stop()
>>> timer.stopped_at
9

Calling stop() method multipe times is safe. Each call remembers current time when stop() was called::

>>> timer.stop()
>>> timer.stopped_at
50

When stop_func attribute is set, it is called when timer stop() metod is called with timer passed as first artument::

>>> timer = Timer(name='MyTimer', time_func=partial([1,9.1234,30.789].pop, 0), 
...               stop_func=lambda t, *args: print(f'{t.name}:', *args, f'{round(t.elapsed,6)}s'))
>>> timer.stop('load finished at')
MyTimer: load finished at 8.1234s
>>> timer.stop('parse finished at')
MyTimer: parse finished at 29.789s

property stop_func: Callable: Function to be called after at the end of the stop() method.

property time

The current time as returned by the time_func function.

Example:

>>> timer = Timer(time_func=lambda : 5)
>>> timer.time
5

property time_func: Callable[[], float]: The function used to get the current time.