# PyWebTransport
[](https://badge.fury.io/py/pywebtransport)
[](https://pypi.org/project/pywebtransport/)
[](https://opensource.org/licenses/MIT)
[](https://github.com/lemonsterfy/pywebtransport/actions)
[](https://codecov.io/gh/lemonsterfy/pywebtransport)
[](https://pywebtransport.readthedocs.io/en/latest/)
The canonical, async-native WebTransport stack for Python.
## Features
- **Full Async Support**: Built from the ground up on asyncio for high-performance, non-blocking I/O.
- **High-Level Frameworks**: Includes a ServerApp with routing and middleware, and a versatile WebTransportClient with helpers for pooling, auto-reconnection, and proxying.
- **Complete Protocol Implementation**: Full support for bidirectional and unidirectional streams, as well as unreliable datagrams.
- **Structured Messaging**: Pluggable JSON, MsgPack, and Protobuf serializers for sending and receiving structured data objects over streams and datagrams.
- **Lifecycle and Resource Management**: Robust, async context-managed components for handling connections, sessions, streams, and monitoring.
- **Event-Driven Architecture**: A powerful EventEmitter and EventBus system for decoupled, asynchronous communication between components.
- **Type-Safe and Tested**: A fully type-annotated API with extensive test coverage (unit, integration, E2E) to ensure reliability and maintainability.
## Installation
```bash
pip install pywebtransport
```
For more detailed instructions, including virtual environments and platform-specific notes, see the [Installation Guide](docs/installation.md).
## Quick Start
### Server
```python
# server.py
import asyncio
from pywebtransport import ServerApp, ServerConfig, WebTransportSession, WebTransportStream
from pywebtransport.exceptions import ConnectionError, SessionError
from pywebtransport.utils import generate_self_signed_cert
generate_self_signed_cert(hostname="localhost")
app = ServerApp(
config=ServerConfig.create(
certfile="localhost.crt",
keyfile="localhost.key",
)
)
async def handle_datagrams(session: WebTransportSession) -> None:
try:
datagrams = await session.datagrams
while True:
data = await datagrams.receive()
await datagrams.send(data=b"ECHO: " + data)
except (ConnectionError, SessionError, asyncio.CancelledError):
pass
async def handle_streams(session: WebTransportSession) -> None:
try:
async for stream in session.incoming_streams():
if isinstance(stream, WebTransportStream):
data = await stream.read_all()
await stream.write_all(data=b"ECHO: " + data)
except (ConnectionError, SessionError, asyncio.CancelledError):
pass
@app.route(path="/")
async def echo_handler(session: WebTransportSession) -> None:
datagram_task = asyncio.create_task(handle_datagrams(session))
stream_task = asyncio.create_task(handle_streams(session))
try:
await session.wait_closed()
finally:
datagram_task.cancel()
stream_task.cancel()
if __name__ == "__main__":
app.run(host="127.0.0.1", port=4433)
```
### Client
```python
# client.py
import asyncio
import ssl
from pywebtransport import ClientConfig, WebTransportClient
async def main() -> None:
config = ClientConfig.create(verify_mode=ssl.CERT_NONE)
async with WebTransportClient(config=config) as client:
session = await client.connect(url="https://127.0.0.1:4433/")
print("Connection established. Testing datagrams...")
datagrams = await session.datagrams
await datagrams.send(data=b"Hello, Datagram!")
response = await datagrams.receive()
print(f"Datagram echo: {response!r}\n")
print("Testing streams...")
stream = await session.create_bidirectional_stream()
await stream.write_all(data=b"Hello, Stream!")
response = await stream.read_all()
print(f"Stream echo: {response!r}")
await session.close()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
```
## Documentation
- **[Installation Guide](docs/installation.md)** - In-depth setup and installation guide.
- **[Quick Start](docs/quickstart.md)** - A 5-minute tutorial to get started.
- **[API Reference](docs/api-reference/)** - Complete API documentation.
## Requirements
- Python 3.11+
- asyncio support
- TLS 1.3
**Dependencies:**
- aioquic >= 1.2.0
- cryptography >= 45.0.4
## Contributing
Contributions are welcome! Please read our [Contributing Guide](CONTRIBUTING.md) for details on the development setup, testing, and pull request process.
**Development Setup:**
```bash
git clone https://github.com/lemonsterfy/pywebtransport.git
cd pywebtransport
pip install -r dev-requirements.txt
pip install -e .
tox
```
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Acknowledgments
- [aioquic](https://github.com/aiortc/aioquic) for the underlying QUIC protocol implementation.
- [WebTransport Working Group](https://datatracker.ietf.org/wg/webtrans/) for standardizing the protocol.
## Support
- **Issues**: [GitHub Issues](https://github.com/lemonsterfy/pywebtransport/issues)
- **Discussions**: [GitHub Discussions](https://github.com/lemonsterfy/pywebtransport/discussions)
- **Email**: lemonsterfy@gmail.com
Raw data
{
"_id": null,
"home_page": null,
"name": "pywebtransport",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "async, http3, networking, protocol, quic, webtransport",
"author": null,
"author_email": "lemonsterfy <lemonsterfy@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/7f/be/8755c4b5df388d4e19f66c17caed58bab229ef6165b926818fafad2d83fd/pywebtransport-0.5.1.tar.gz",
"platform": null,
"description": "# PyWebTransport\n\n[](https://badge.fury.io/py/pywebtransport)\n[](https://pypi.org/project/pywebtransport/)\n[](https://opensource.org/licenses/MIT)\n[](https://github.com/lemonsterfy/pywebtransport/actions)\n[](https://codecov.io/gh/lemonsterfy/pywebtransport)\n[](https://pywebtransport.readthedocs.io/en/latest/)\n\nThe canonical, async-native WebTransport stack for Python.\n\n## Features\n\n- **Full Async Support**: Built from the ground up on asyncio for high-performance, non-blocking I/O.\n- **High-Level Frameworks**: Includes a ServerApp with routing and middleware, and a versatile WebTransportClient with helpers for pooling, auto-reconnection, and proxying.\n- **Complete Protocol Implementation**: Full support for bidirectional and unidirectional streams, as well as unreliable datagrams.\n- **Structured Messaging**: Pluggable JSON, MsgPack, and Protobuf serializers for sending and receiving structured data objects over streams and datagrams.\n- **Lifecycle and Resource Management**: Robust, async context-managed components for handling connections, sessions, streams, and monitoring.\n- **Event-Driven Architecture**: A powerful EventEmitter and EventBus system for decoupled, asynchronous communication between components.\n- **Type-Safe and Tested**: A fully type-annotated API with extensive test coverage (unit, integration, E2E) to ensure reliability and maintainability.\n\n## Installation\n\n```bash\npip install pywebtransport\n```\n\nFor more detailed instructions, including virtual environments and platform-specific notes, see the [Installation Guide](docs/installation.md).\n\n## Quick Start\n\n### Server\n\n```python\n# server.py\nimport asyncio\n\nfrom pywebtransport import ServerApp, ServerConfig, WebTransportSession, WebTransportStream\nfrom pywebtransport.exceptions import ConnectionError, SessionError\nfrom pywebtransport.utils import generate_self_signed_cert\n\ngenerate_self_signed_cert(hostname=\"localhost\")\n\napp = ServerApp(\n config=ServerConfig.create(\n certfile=\"localhost.crt\",\n keyfile=\"localhost.key\",\n )\n)\n\n\nasync def handle_datagrams(session: WebTransportSession) -> None:\n try:\n datagrams = await session.datagrams\n while True:\n data = await datagrams.receive()\n await datagrams.send(data=b\"ECHO: \" + data)\n except (ConnectionError, SessionError, asyncio.CancelledError):\n pass\n\n\nasync def handle_streams(session: WebTransportSession) -> None:\n try:\n async for stream in session.incoming_streams():\n if isinstance(stream, WebTransportStream):\n data = await stream.read_all()\n await stream.write_all(data=b\"ECHO: \" + data)\n except (ConnectionError, SessionError, asyncio.CancelledError):\n pass\n\n\n@app.route(path=\"/\")\nasync def echo_handler(session: WebTransportSession) -> None:\n datagram_task = asyncio.create_task(handle_datagrams(session))\n stream_task = asyncio.create_task(handle_streams(session))\n try:\n await session.wait_closed()\n finally:\n datagram_task.cancel()\n stream_task.cancel()\n\n\nif __name__ == \"__main__\":\n app.run(host=\"127.0.0.1\", port=4433)\n\n```\n\n### Client\n\n```python\n# client.py\nimport asyncio\nimport ssl\n\nfrom pywebtransport import ClientConfig, WebTransportClient\n\n\nasync def main() -> None:\n config = ClientConfig.create(verify_mode=ssl.CERT_NONE)\n\n async with WebTransportClient(config=config) as client:\n session = await client.connect(url=\"https://127.0.0.1:4433/\")\n\n print(\"Connection established. Testing datagrams...\")\n datagrams = await session.datagrams\n await datagrams.send(data=b\"Hello, Datagram!\")\n response = await datagrams.receive()\n print(f\"Datagram echo: {response!r}\\n\")\n\n print(\"Testing streams...\")\n stream = await session.create_bidirectional_stream()\n await stream.write_all(data=b\"Hello, Stream!\")\n response = await stream.read_all()\n print(f\"Stream echo: {response!r}\")\n\n await session.close()\n\n\nif __name__ == \"__main__\":\n try:\n asyncio.run(main())\n except KeyboardInterrupt:\n pass\n\n```\n\n## Documentation\n\n- **[Installation Guide](docs/installation.md)** - In-depth setup and installation guide.\n- **[Quick Start](docs/quickstart.md)** - A 5-minute tutorial to get started.\n- **[API Reference](docs/api-reference/)** - Complete API documentation.\n\n## Requirements\n\n- Python 3.11+\n- asyncio support\n- TLS 1.3\n\n**Dependencies:**\n\n- aioquic >= 1.2.0\n- cryptography >= 45.0.4\n\n## Contributing\n\nContributions are welcome! Please read our [Contributing Guide](CONTRIBUTING.md) for details on the development setup, testing, and pull request process.\n\n**Development Setup:**\n\n```bash\ngit clone https://github.com/lemonsterfy/pywebtransport.git\ncd pywebtransport\npip install -r dev-requirements.txt\npip install -e .\ntox\n```\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## Acknowledgments\n\n- [aioquic](https://github.com/aiortc/aioquic) for the underlying QUIC protocol implementation.\n- [WebTransport Working Group](https://datatracker.ietf.org/wg/webtrans/) for standardizing the protocol.\n\n## Support\n\n- **Issues**: [GitHub Issues](https://github.com/lemonsterfy/pywebtransport/issues)\n- **Discussions**: [GitHub Discussions](https://github.com/lemonsterfy/pywebtransport/discussions)\n- **Email**: lemonsterfy@gmail.com\n",
"bugtrack_url": null,
"license": null,
"summary": "The canonical, async-native WebTransport stack for Python.",
"version": "0.5.1",
"project_urls": {
"Changelog": "https://github.com/lemonsterfy/pywebtransport/blob/main/CHANGELOG.md",
"Documentation": "https://pywebtransport.readthedocs.io",
"Homepage": "https://github.com/lemonsterfy/pywebtransport",
"Issues": "https://github.com/lemonsterfy/pywebtransport/issues",
"Repository": "https://github.com/lemonsterfy/pywebtransport.git"
},
"split_keywords": [
"async",
" http3",
" networking",
" protocol",
" quic",
" webtransport"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "c9c06c3cb91d64b5d769e0af8310cac9d01eef51c525e7a8baf265d9ae54965d",
"md5": "30a6cc905ec9caafa99697905420dac4",
"sha256": "c8cbeffbbff01d5edb92ef0112b0182a6d434c75ffb2a2f5539fe841cfe46358"
},
"downloads": -1,
"filename": "pywebtransport-0.5.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "30a6cc905ec9caafa99697905420dac4",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 134519,
"upload_time": "2025-09-10T20:48:37",
"upload_time_iso_8601": "2025-09-10T20:48:37.456251Z",
"url": "https://files.pythonhosted.org/packages/c9/c0/6c3cb91d64b5d769e0af8310cac9d01eef51c525e7a8baf265d9ae54965d/pywebtransport-0.5.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "7fbe8755c4b5df388d4e19f66c17caed58bab229ef6165b926818fafad2d83fd",
"md5": "380acffed8037a6d31d30411a0543389",
"sha256": "5c3a6b569853ea73d63d0acc709457385240f4bd6e1d158a633f3686ab661554"
},
"downloads": -1,
"filename": "pywebtransport-0.5.1.tar.gz",
"has_sig": false,
"md5_digest": "380acffed8037a6d31d30411a0543389",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 107616,
"upload_time": "2025-09-10T20:48:35",
"upload_time_iso_8601": "2025-09-10T20:48:35.724467Z",
"url": "https://files.pythonhosted.org/packages/7f/be/8755c4b5df388d4e19f66c17caed58bab229ef6165b926818fafad2d83fd/pywebtransport-0.5.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-10 20:48:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "lemonsterfy",
"github_project": "pywebtransport",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pywebtransport"
}
JSON
