Name | nats-app JSON |
Version |
0.6.2
JSON |
| download |
home_page | None |
Summary | NATS App is wrapper client on NATS Connection |
upload_time | 2025-07-25 14:24:55 |
maintainer | None |
docs_url | None |
author | Vadym Statishyn |
requires_python | <4.0,>=3.12 |
license | Apache-2.0 |
keywords |
nats
|
VCS |
|
bugtrack_url |
|
requirements |
No requirements were recorded.
|
Travis-CI |
No Travis.
|
coveralls test coverage |
No coveralls.
|
# NATS App
[](https://pypi.python.org/pypi/nats_app)
[](https://github.com/centum/nats_app)
[](https://github.com/centum/nats_app/actions/workflows/test.yml)
[](https://raw.githubusercontent.com/centum/nats_app/refs/heads/master/LICENSE)
NATS App is wrapper application on NATS Connection
### Create NATS Application
```python
NATS_URL = "nats://localhost:4222"
nc = NATSApp(NATS_URL)
await nc.connect()
```
### Add RPC Handler
```python
@nc.push_subscribe("app.test.echo", queue="worker")
async def rpc_func_echo(msg: Any) -> str:
return f"Response with Msg.data: {msg.data}"
```
### Call RPC
```python
res = await nc.request("app.test.echo", {"args": ["test"]})
print(res)
```
### JetStream push subscription
```python
@nc.js_push_subscribe("app.test.js.subs", queue="worker")
async def handler(msg: Msg):
print(msg.data)
await msg.ack()
```
### JetStream pull subscription
```python
@nc.js_pull_subscribe("app.subject.js.subs", batch=1)
async def handler(msgs: list[Msg]):
for m in msgs:
print(m)
await m.ack()
```
### JetStream publish
```python
await nc.js.publish("app.subject.js.subs", b"TEST123")
```
Raw data
{
"_id": null,
"home_page": null,
"name": "nats-app",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.12",
"maintainer_email": null,
"keywords": "nats",
"author": "Vadym Statishyn",
"author_email": "statishin@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/cc/5e/d63730827a54f8410c100284b577161abce068d1a144cbbd0471d1473721/nats_app-0.6.2.tar.gz",
"platform": null,
"description": "# NATS App\n\n[](https://pypi.python.org/pypi/nats_app)\n[](https://github.com/centum/nats_app)\n[](https://github.com/centum/nats_app/actions/workflows/test.yml)\n[](https://raw.githubusercontent.com/centum/nats_app/refs/heads/master/LICENSE)\n\nNATS App is wrapper application on NATS Connection\n\n### Create NATS Application\n\n```python\nNATS_URL = \"nats://localhost:4222\"\n\nnc = NATSApp(NATS_URL)\nawait nc.connect()\n```\n\n### Add RPC Handler\n\n```python\n@nc.push_subscribe(\"app.test.echo\", queue=\"worker\")\nasync def rpc_func_echo(msg: Any) -> str:\n return f\"Response with Msg.data: {msg.data}\"\n```\n\n### Call RPC\n```python\nres = await nc.request(\"app.test.echo\", {\"args\": [\"test\"]})\nprint(res)\n```\n\n### JetStream push subscription\n\n```python\n@nc.js_push_subscribe(\"app.test.js.subs\", queue=\"worker\")\nasync def handler(msg: Msg):\n print(msg.data)\n await msg.ack()\n```\n\n### JetStream pull subscription\n\n```python\n@nc.js_pull_subscribe(\"app.subject.js.subs\", batch=1)\nasync def handler(msgs: list[Msg]):\n for m in msgs:\n print(m)\n await m.ack()\n```\n\n### JetStream publish\n\n```python\nawait nc.js.publish(\"app.subject.js.subs\", b\"TEST123\")\n```\n\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "NATS App is wrapper client on NATS Connection",
"version": "0.6.2",
"project_urls": null,
"split_keywords": [
"nats"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "fecd1e3f5ec865238105b3fe9805a317771b55099098fbe0db0e64c9d3990fa8",
"md5": "256a75e156cb7c2d617e0cb4560ab90f",
"sha256": "658af4dee4e51f0ef2c236860b56d33f02f61518eab216d0f4a0f03b57d5fffb"
},
"downloads": -1,
"filename": "nats_app-0.6.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "256a75e156cb7c2d617e0cb4560ab90f",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.12",
"size": 26838,
"upload_time": "2025-07-25T14:24:53",
"upload_time_iso_8601": "2025-07-25T14:24:53.718604Z",
"url": "https://files.pythonhosted.org/packages/fe/cd/1e3f5ec865238105b3fe9805a317771b55099098fbe0db0e64c9d3990fa8/nats_app-0.6.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "cc5ed63730827a54f8410c100284b577161abce068d1a144cbbd0471d1473721",
"md5": "415c46a363a2d313740d774c2dc8e7e5",
"sha256": "80b161b97f0f2569a19522b2d661011c198599589d0c5c5d81cf14ea399b9bbf"
},
"downloads": -1,
"filename": "nats_app-0.6.2.tar.gz",
"has_sig": false,
"md5_digest": "415c46a363a2d313740d774c2dc8e7e5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.12",
"size": 20705,
"upload_time": "2025-07-25T14:24:55",
"upload_time_iso_8601": "2025-07-25T14:24:55.427779Z",
"url": "https://files.pythonhosted.org/packages/cc/5e/d63730827a54f8410c100284b577161abce068d1a144cbbd0471d1473721/nats_app-0.6.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-25 14:24:55",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "nats-app"
}