prawvents

PRAWvents, Events for PRAW

A simple wrapper to write event-based bots with PRAW.

Scope

You can register event handlers for everything thats based on the praw stream_generator Any other functionality is offered as-is, since this subclasses the main PRAW Reddit instance.

A async version of this should be possible, but is not yet planned.

Quickstart

This is a simple bot that will print out the subreddit and the submission title for all posts in the subreddits AskReddit and pics, while skipping the existing posts in AskReddit. This example assumes the presence of a praw.ini in your working directory.

from prawvents import EventReddit
from praw import reddit

r = EventReddit(user_agent=f"ExampleBot for prawvents version (0.0.1) by /u/laundmo") # change the description and username!

sub1 = r.subreddit("AskReddit")
sub2 = r.subreddit("pics")

def handle_exception(e): # very dumb exception handler
    print(e)

@r.register_event(sub1.stream.submissions, err_handler=handle_exception, skip_existing=True)
@r.register_event(sub2.stream.submissions, err_handler=handle_exception)
def handle(submission: reddit.Submission):
    print(submission.subreddit, submission.title)

r.run_loop()

Docs

RedditEventDecorator Objects

class RedditEventDecorator()

Decorator class for event handlers.

__init__

 | __init__(reddit: praw.Reddit, stream: RStream, err_handler: Callable)

Initialise RedditEventDecorator.

Arguments:

• reddit EventReddit - The EventReddit instance
• stream RStream - The stream to which the event responds.
• err_handler Callable - A function thats called with the exception as a argument.

__call__

 | __call__(f: Callable) -> Callable

Set the event handler.

Arguments:

• f Callable - The event handler function.

Returns:

• Callable - The function.

EventReddit Objects

class EventReddit(praw.Reddit)

Main Reddit instance, subclass of praw.Reddit.

Arguments:

• praw praw.Reddit - Praw Reddit superclass.

__init__

 | __init__(*args, **kwargs)

Initialise EventReddit. All arguments are passed through to praw.Reddit

register_event

 | register_event(stream: RStream, err_handler: Callable = None, **kwargs) -> RedditEventDecorator

Register a event, should generally be used as a decorator like this:

@r.register_event(subreddit.stream.submissions, err_handler=handle_exception)
def event_handler(submission):
    pass

Arguments:

• stream RStream - The stream to which the event responds.
• err_handler Callable, optional - The error handler for this event. Defaults to None.

Returns:

• RedditEventDecorator - The decorator instance.

handle_exception

 | handle_exception(f: Callable, e: Exception)

Handle a Exception happening in a function f

Arguments:

• f Callable - The function which threw the exception.
• e Exception - The exception which was thrown.

Raises:

• e - The Exception that was thrown.

run_stream_till_none

 | run_stream_till_none(stream: RStream, funcs: Iterable[Callable]) -> None

Runs a stream until none is returned

Arguments:

• stream RStream - The finalized stream to run.
• funcs Iterable[Callable] - The functions which handle this stream.

run_loop

 | run_loop(interweave=True) -> None

Run the event loop. If interweave is Truthy, events from multiple streams will be mixed to ensure a single high-traffic stream cant take up the entire event loop. This is highly recommended.

Arguments:

• interweave bool, optional - Whether to interweave streams to ensure fair distribution. Defaults to True.