ps-py 1.0.0.1

PS_PY: Pub-Sub-Python

Dependancies

• python>=3.5.9

QuickStart

Installation and Basic Configuration

1. Install PS_PY: Pub-Sub-Python by running pip install ps_py.

2. There is no specific configuration needed to use ps_py.

# **non-callable subject**
# publisher code
from ps_py.publisher import Publisher
publisher = Publisher()
subject = publisher.add('example_subject', initial_value='example_value')
subject.next('example_value_1')  # to publish new value.

# subscriber code
from ps_py.publisher import Publisher
publisher = Publisher()
subject = publisher.get_subject('example_subject')
# NOTE: The above will create a new subject **if and only if** the subject doesn't exist.

"""Assuming ``on_success``, ``on_error`` are already defined somewhere
before the below statement.
``onSuccess`` is mandatory to subscribe; whereas ``onError`` is optional.
``onSuccess``, ``onError``: can be either callable or a ``dict`` of format shown below:
``{'func': <callable>, 'args': <None | list | tuple>, 'kwargs': <None | dict>}``
``onSuccess``: The callable defined either directly or via dict should take a
argument (positional or keyword) with name ``value``.
``onError``: The callable defined either directly or via dict should take a
argument (positional or keyword) with name ``error``.
"""
subscriber = subject.subscribe(onSuccess=on_success, onError=on_error)
"""That is it. at any point the publisher publishes any thing to this subject
the ``onSuccess`` or ``onError`` will be called accordingly.
NOTE: if the subject has any value which is not None and not initial_value,
it will be passed to ``onSuccess`` right away.
"""
# to unsubscribe
subscriber.unsubscribe()

# **callable subject**
# there is no publisher code for this because the callable will be the publisher
from ps_py.publisher import Publisher
publisher = Publisher()
subject = publisher.add(lambda x,y,z=None: (x, y, z), 'x', 'y', 'z'='z_val', initial_value='init')
# syntax: subject = publisher.add(<function>, ``*args``, ``**kwargs``)
"""NOTE: The subject callable will not be called/executed until the first subscribe.
If no subscription the the subject callable will never be called/executed.
"""
subscriber = subject.subscribe(onSuccess=on_success, onError=on_error)

# **Creating subjects and using it directly**
from ps_py.subject import Subject
# for non-callable subject
# Subject constructor takes in one mandatory value that is subject unique identifier.
# ``initial_value`` is optional.
subject = Subject('example_subject', initial_value='example_value')
subject.next('example_value_1')

"""Assuming ``on_success``, ``on_error`` are already defined somewhere
before the below statement.
"""
subscriber = subject.subscribe(onSuccess=on_success, onError=on_error)

# for callable subject
subject = Subject(lambda x,y,z=None: (x, y, z), 'x', 'y', 'z'='z_val', initial_value='init')
"""NOTE: The function passed is called only at the fisrt subscription."""

"""Assuming ``on_success``, ``on_error`` are already defined somewhere
before the below statement.
"""
subscriber = subject.subscribe(onSuccess=on_success, onError=on_error)

Basic Usage

Below are some basic ussage for PS_PY package.

example:

>>>from ps_py.subject import Subject
>>>sbj = Subject('exp1', initial_value='val1')
>>>sbj.value
'val1'
>>>sbj2 = Subject('exp1', initial_value='val2')
>>>id(sbj) == id(sbj2)
True
>>>sbj.value  # the value of sbj is now the latest that is provided by sbj2
'val2'
>>>sbc = sbj.subscribe(onSuccess=lambda value: print(value), onError=None)  # onError is optional, sbj.value is passed to onSuccess; return Subscriber object.
val2
>>>sbc1 = sbj.subscribe(onSuccess=lambda value: print(value), onError=None)
val2
>>>sbj.next('val3')  # calls the onSuccess of both ``sbc`` anf ``sbc1``; returns
val3
val3
>>>sbc1.unsubscribe()  # ``sbc1`` is obsolete after this point.
>>>sbj.next([1, 2])  # calls the onSuccess of ``sbc``; any picklable value can be send.
[1, 2]
>>>sbj2 = Subject(lambda val, val2=None: (val, val2), 'args', val2='kwargs')
>>>sbc2 = sbj2.subscribe(onSuccess=lambda value: print(value))
('args', 'kwargs')

>>>from ps_py.contrib import map
>>>sbj = Subject("pip-map_ex", initial_value='val1')
>>>sbj1 = sbj.pipe(map(lambda value: "%s:%s"%("modified", value)))
>>>sbc = sbj1.subscribe(onSuccess=lambda value: print(value))
modified:val1  # the current value in the sbj is passed down to the pip.
>>>sbj.next('new_val')
modified:new_val

>>>from ps_py.contrib import Merge, map
>>>sbj1 = Subject("Merge_ex_1")
>>>sbj2 = Subject("Merge_ex_2")
>>>mrg = Merge(sbj1, sbj2)
>>>sbc = mrg.subscribe(onSuccess=lambda value: print(value))
>>>sbj1.next("first_val")
first_val
>>>sbj2.next("second_val")
second_val
>>>mrg_sbj = mrg.pipe(map(lambda value: "%s:%s"%("mergePipe", value)))
>>>sbc1 = mrg_sbj.subscribe(onSuccess=lambda value: print(value))
mergePipe:second_val  # latest value is passed to the pipe.
>>>sbj1.next("third_value")
third_value  # this is the value printed by ``sbc``'s ``onSuccess``
mergePipe:third_value  # this is the value printed by ``sbc1``'s ``onSuccess``
>>>sbj2.next("fourth_value")
fourth_value  # this is the value printed by ``sbc``'s ``onSuccess``
mergePipe:fourth_value  # this is the value printed by ``sbc1``'s ``onSuccess``
>>>sbc.unsubscribe()
>>>sbj1.next("fifth_value")
mergePipe:fifth_value  # this is the value printed by ``sbc1``'s ``onSuccess``

>>>from ps_py.contrib import Of, map
>>>of_obj = Of("val1", ["v", "a", "l", "2"], {"v": "a", "l":3}, timeout=2)
>>>of_sbj = of_obj.pipe(map(lambda value: "%s:%s"%("ofPipe", value)))
>>>sbc1 = of_sbj.subscribe(onSuccess=lambda value: print(value))
>>>sbc = of_obj.subscribe(onSuccess=lambda value: print("of_direct:%s" % value)
of_direct:val1
ofPipe:val1
of_direct:['v', 'a', 'l', '2']
ofPipe:['v', 'a', 'l', '2']
of_direct:{'l': 3, 'v': 'a'}
ofPipe:{'l': 3, 'v': 'a'}

Publisher APIs

Subject APIs

Subscriber APIs

Merge APIs

Of APIs