PHX Events

PHX Events is an AsyncIO library to set up a websocket connection with Phoenix Channels in Python 3.9+.

Check out the phx_event documentation

Installing PHX Events

From Pip

pip install phx-events

From Source

Clone the Git repo and then install the dependencies

pip install -r requirements/core.txt

Use the client in your code:

import asyncio
from concurrent.futures import ThreadPoolExecutor

from phx_events.client import PHXChannelsClient
from phx_events.phx_messages import ChannelMessage, Event, Topic


def print_handler(message: ChannelMessage, client: PHXChannelsClient) -> None:
    client.logger.info(f'DEFAULT: {message=}')


async def async_print_handler(message: ChannelMessage, client: PHXChannelsClient) -> None:
    client.logger.info(f'ASYNC: {message=}')


async def main() -> None:
    token = 'auth_token'
    client: PHXChannelsClient

    with ThreadPoolExecutor() as pool:
        async with PHXChannelsClient(token) as client:
            client.register_event_handler(
                event=Event('event_name'),
                handlers=[
                    print_handler,
                    async_print_handler,
                ],
            )
            client.register_topic_subscription(Topic('topic:subtopic'))

            await client.start_processing(pool)


if __name__ == '__main__':
    asyncio.run(main(), debug=True)

Developing

This project uses pip-tools to manage dependencies.

Before contributing ensure you agree to the DCO. Commits contributed to this project need to be signed in git.

This can be done as follows (Note -s is a shortcut for --signoff) :

git commit --signoff

1. Create a virtualenv

Note: Creating the virtualenv can be done however you want. We will assume you've done created a new virtualenv and activated it from this point.

2. Install pip-tools:

pip install pip-tools

3. Install Dependencies

pip-sync requirements/core.txt requirements/dev.txt