Metadata-Version: 2.5
Name: serverwamp
Version: 2.0.0
Summary: Components that add Web Application Messaging Protocol features to WebSocket servers.
Project-URL: Homepage, https://github.com/JustinTArthur/serverwamp
Project-URL: Documentation, https://serverwamp.readthedocs.io/
Author-email: Justin Turner Arthur <justinarthur@gmail.com>
License-Expression: MIT
License-File: LICENSE.md
Keywords: ASGI,RPC,WAMP,WebSockets,aiohttp,broker,dealer,pubsub
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Framework :: Trio
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Topic :: Internet
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Requires-Dist: msgpack<2,>=1.0.0
Provides-Extra: anyio
Requires-Dist: anyio>=4.0; extra == 'anyio'
Provides-Extra: docs
Requires-Dist: sphinx<10,>=5.3; extra == 'docs'
Provides-Extra: trio
Requires-Dist: trio>=0.25.0; extra == 'trio'
Description-Content-Type: text/markdown

# serverwamp
Adds [Web Application Messaging Protocol](https://wamp-proto.org/) features to
WebSocket servers. _serverwamp_ makes it easy to both serve requests from
clients and push events to clients over a single connection. Currently supports
ASGI servers using asyncio or Trio and aiohttp servers.

[Documentation](https://serverwamp.readthedocs.io/)

### Usage Example
```python
import asyncio

from aiohttp import web
import serverwamp

docs_api = serverwamp.RPCRouteSet()
docs_events = serverwamp.TopicRouteSet()

@docs_api.route('docs.getDoc')
async def get_doc(document_id):
    record = await my_db.retrieve(document_id)
    return {'status': 'SUCCESS', 'doc': record}

@docs_api.route('docs.deleteDoc')
async def delete_doc(document_id):
    succeeded = await my_db.delete(document_id)
    if succeeded:
        return {'status': 'SUCCESS'}

    return serverwamp.RPCErrorResult(
        'wamp.error.delete_failed',
        kwargs={'status': 'FAILURE'}
    )

@docs_events.route('docs.changes')
async def subscribe_to_doc_changes(session, changes_subscribers):
    changes_subscribers.add(session)
    yield
    changes_subscribers.remove(session)


async def server_context(app):

    async def watch_for_docs_changes():
        async for event in my_db.watch_changes():
            # A copy, as sessions can unsubscribe while events are sent.
            for session in list(app['changes_subscribers']):
                await session.send_event(
                    'docs.changes',
                    args=(f'Document changed: {event}',)
                )

    watcher = asyncio.create_task(watch_for_docs_changes())
    try:
        yield
    finally:
        watcher.cancel()


if __name__ == '__main__':
    changes_subscribers = set()

    app = serverwamp.Application()
    app.add_rpc_routes(docs_api)
    app.add_topic_routes(docs_events)
    app.set_default_arg('changes_subscribers', changes_subscribers)

    http_app = web.Application()
    http_app['changes_subscribers'] = changes_subscribers
    http_app.add_routes((
        web.get('/', app.aiohttp_websocket_handler()),
    ))
    http_app.cleanup_ctx.append(server_context)

    web.run_app(http_app)
```


## Development
Unit tests can be run with:

    pip install -e . --group ci
    pytest

Linting and type checking can be run with:

    ruff check serverwamp tests examples
    mypy serverwamp
