nats-app


Namenats-app JSON
Version 0.6.2 PyPI version JSON
download
home_pageNone
SummaryNATS App is wrapper client on NATS Connection
upload_time2025-07-25 14:24:55
maintainerNone
docs_urlNone
authorVadym Statishyn
requires_python<4.0,>=3.12
licenseApache-2.0
keywords nats
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # NATS App

[![pypi](https://img.shields.io/pypi/v/nats_app.svg)](https://pypi.python.org/pypi/nats_app)
[![versions](https://img.shields.io/pypi/pyversions/nats_app.svg)](https://github.com/centum/nats_app)
[![test](https://github.com/centum/nats_app/actions/workflows/test.yml/badge.svg)](https://github.com/centum/nats_app/actions/workflows/test.yml)
[![Apache 2 licensed](https://img.shields.io/badge/license-Apache2-blue.svg)](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[![pypi](https://img.shields.io/pypi/v/nats_app.svg)](https://pypi.python.org/pypi/nats_app)\n[![versions](https://img.shields.io/pypi/pyversions/nats_app.svg)](https://github.com/centum/nats_app)\n[![test](https://github.com/centum/nats_app/actions/workflows/test.yml/badge.svg)](https://github.com/centum/nats_app/actions/workflows/test.yml)\n[![Apache 2 licensed](https://img.shields.io/badge/license-Apache2-blue.svg)](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"
}
        
Elapsed time: 0.59062s