fastcc


Namefastcc JSON
Version 4.0.4 PyPI version JSON
download
home_pageNone
SummaryLightweight, efficient and developer-friendly framework for component communication
upload_time2025-02-14 08:34:56
maintainerNone
docs_urlNone
authorNone
requires_python>=3.13
licenseNone
keywords python framework asyncio aiomqtt mqtt iot
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            <p align="center">
    <img src="https://github.com/ReMi-HSBI/fastcc/blob/main/docs/src/static/images/fastci_logo.svg?raw=true" alt="drawing" width="33%"/>
</p>

# FastCC

<a href="https://docs.astral.sh/ruff">
  <img
    src="https://img.shields.io/badge/ruff-⚡-261230.svg?style=flat-square"
    alt="Ruff"
  />
</a>
<a href="https://mypy-lang.org">
  <img
    src="https://img.shields.io/badge/mypy-📝-2a6db2.svg?style=flat-square"
    alt="Mypy"
  />
</a>
<a href="https://gitmoji.dev">
  <img
    src="https://img.shields.io/badge/gitmoji-😜%20😍-FFDD67.svg?style=flat-square"
    alt="Gitmoji"
  />
</a>

Framework for component communication with [MQTT](https://mqtt.org) and [Protocol Buffers](https://protobuf.dev) :boom:.

- Lightweight :zap:
- Efficient :rocket:
- Developer-friendly :technologist:

This framework is built on top of [empicano](https://github.com/empicano)'s [aiomqtt](https://github.com/empicano/aiomqtt).

## Example

```python
# app.py
from __future__ import annotations

import asyncio
import contextlib
import logging
import os
import sys

import fastcc

router = fastcc.Router()


@router.route("greet")
async def greet(name: str, *, database: dict[str, int]) -> str:
    """Greet a user.

    Parameters
    ----------
    name
        The name of the user.
        Autofilled by fastcc.
    database
        The database.
        Autofilled by fastcc.

    Returns
    -------
    str
        The greeting message.
    """
    # ... do some async work
    await asyncio.sleep(0.1)

    database[name] += 1
    occurrence = database[name]
    return f"Hello, {name}! Saw you {occurrence} times!"


async def main() -> None:
    """Run the app."""
    logging.basicConfig(level=logging.INFO)

    database: dict[str, int] = {"Alice": 0, "Bob": 0}
    app = fastcc.FastCC("localhost")
    app.add_router(router)
    app.add_injector(database=database)
    app.add_exception_handler(
        KeyError,
        lambda e: fastcc.MQTTError(repr(e), 404),
    )

    await app.run()


# https://github.com/empicano/aiomqtt?tab=readme-ov-file#note-for-windows-users
if sys.platform.lower() == "win32" or os.name.lower() == "nt":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

with contextlib.suppress(KeyboardInterrupt):
    asyncio.run(main())
```

```python
# client.py
from __future__ import annotations

import asyncio
import contextlib
import logging
import os
import sys

import fastcc

_logger = logging.getLogger(__name__)


async def main() -> None:
    """Run the app."""
    logging.basicConfig(level=logging.INFO)

    async with fastcc.Client("localhost") as client:
        try:
            response = await client.request(
                "greet",
                "Charlie",
                response_type=str,
            )
        except fastcc.MQTTError as e:
            details = f"An error occurred on the server: {e.message}"
            _logger.error(details)

        response = await client.request("greet", "Alice", response_type=str)
        _logger.info("response: %r", response)


# https://github.com/empicano/aiomqtt?tab=readme-ov-file#note-for-windows-users
if sys.platform.lower() == "win32" or os.name.lower() == "nt":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

with contextlib.suppress(KeyboardInterrupt):
    asyncio.run(main())
```

            

Raw data

            {
    "_id": null,
    "home_page": null,
    "name": "fastcc",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.13",
    "maintainer_email": "\"J. Baudisch\" <justin.baudisch@hsbi.de>",
    "keywords": "python, framework, asyncio, aiomqtt, mqtt, iot",
    "author": null,
    "author_email": "\"J. Baudisch\" <justin.baudisch@hsbi.de>",
    "download_url": "https://files.pythonhosted.org/packages/7b/8f/3b15209a56ed2edb82248ce38a5c279a65073a372f329ee00b0aaccb3183/fastcc-4.0.4.tar.gz",
    "platform": null,
    "description": "<p align=\"center\">\r\n    <img src=\"https://github.com/ReMi-HSBI/fastcc/blob/main/docs/src/static/images/fastci_logo.svg?raw=true\" alt=\"drawing\" width=\"33%\"/>\r\n</p>\r\n\r\n# FastCC\r\n\r\n<a href=\"https://docs.astral.sh/ruff\">\r\n  <img\r\n    src=\"https://img.shields.io/badge/ruff-\u26a1-261230.svg?style=flat-square\"\r\n    alt=\"Ruff\"\r\n  />\r\n</a>\r\n<a href=\"https://mypy-lang.org\">\r\n  <img\r\n    src=\"https://img.shields.io/badge/mypy-\ud83d\udcdd-2a6db2.svg?style=flat-square\"\r\n    alt=\"Mypy\"\r\n  />\r\n</a>\r\n<a href=\"https://gitmoji.dev\">\r\n  <img\r\n    src=\"https://img.shields.io/badge/gitmoji-\ud83d\ude1c%20\ud83d\ude0d-FFDD67.svg?style=flat-square\"\r\n    alt=\"Gitmoji\"\r\n  />\r\n</a>\r\n\r\nFramework for component communication with [MQTT](https://mqtt.org) and [Protocol Buffers](https://protobuf.dev) :boom:.\r\n\r\n- Lightweight :zap:\r\n- Efficient :rocket:\r\n- Developer-friendly :technologist:\r\n\r\nThis framework is built on top of [empicano](https://github.com/empicano)'s [aiomqtt](https://github.com/empicano/aiomqtt).\r\n\r\n## Example\r\n\r\n```python\r\n# app.py\r\nfrom __future__ import annotations\r\n\r\nimport asyncio\r\nimport contextlib\r\nimport logging\r\nimport os\r\nimport sys\r\n\r\nimport fastcc\r\n\r\nrouter = fastcc.Router()\r\n\r\n\r\n@router.route(\"greet\")\r\nasync def greet(name: str, *, database: dict[str, int]) -> str:\r\n    \"\"\"Greet a user.\r\n\r\n    Parameters\r\n    ----------\r\n    name\r\n        The name of the user.\r\n        Autofilled by fastcc.\r\n    database\r\n        The database.\r\n        Autofilled by fastcc.\r\n\r\n    Returns\r\n    -------\r\n    str\r\n        The greeting message.\r\n    \"\"\"\r\n    # ... do some async work\r\n    await asyncio.sleep(0.1)\r\n\r\n    database[name] += 1\r\n    occurrence = database[name]\r\n    return f\"Hello, {name}! Saw you {occurrence} times!\"\r\n\r\n\r\nasync def main() -> None:\r\n    \"\"\"Run the app.\"\"\"\r\n    logging.basicConfig(level=logging.INFO)\r\n\r\n    database: dict[str, int] = {\"Alice\": 0, \"Bob\": 0}\r\n    app = fastcc.FastCC(\"localhost\")\r\n    app.add_router(router)\r\n    app.add_injector(database=database)\r\n    app.add_exception_handler(\r\n        KeyError,\r\n        lambda e: fastcc.MQTTError(repr(e), 404),\r\n    )\r\n\r\n    await app.run()\r\n\r\n\r\n# https://github.com/empicano/aiomqtt?tab=readme-ov-file#note-for-windows-users\r\nif sys.platform.lower() == \"win32\" or os.name.lower() == \"nt\":\r\n    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())\r\n\r\nwith contextlib.suppress(KeyboardInterrupt):\r\n    asyncio.run(main())\r\n```\r\n\r\n```python\r\n# client.py\r\nfrom __future__ import annotations\r\n\r\nimport asyncio\r\nimport contextlib\r\nimport logging\r\nimport os\r\nimport sys\r\n\r\nimport fastcc\r\n\r\n_logger = logging.getLogger(__name__)\r\n\r\n\r\nasync def main() -> None:\r\n    \"\"\"Run the app.\"\"\"\r\n    logging.basicConfig(level=logging.INFO)\r\n\r\n    async with fastcc.Client(\"localhost\") as client:\r\n        try:\r\n            response = await client.request(\r\n                \"greet\",\r\n                \"Charlie\",\r\n                response_type=str,\r\n            )\r\n        except fastcc.MQTTError as e:\r\n            details = f\"An error occurred on the server: {e.message}\"\r\n            _logger.error(details)\r\n\r\n        response = await client.request(\"greet\", \"Alice\", response_type=str)\r\n        _logger.info(\"response: %r\", response)\r\n\r\n\r\n# https://github.com/empicano/aiomqtt?tab=readme-ov-file#note-for-windows-users\r\nif sys.platform.lower() == \"win32\" or os.name.lower() == \"nt\":\r\n    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())\r\n\r\nwith contextlib.suppress(KeyboardInterrupt):\r\n    asyncio.run(main())\r\n```\r\n",
    "bugtrack_url": null,
    "license": null,
    "summary": "Lightweight, efficient and developer-friendly framework for component communication",
    "version": "4.0.4",
    "project_urls": {
        "Issues": "https://github.com/ReMi-HSBI/fastcc/issues",
        "Repository": "https://github.com/ReMi-HSBI/fastcc"
    },
    "split_keywords": [
        "python",
        " framework",
        " asyncio",
        " aiomqtt",
        " mqtt",
        " iot"
    ],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "972c38150bcddf5e08d70cb27f82cdb6c9fa442c4bce76e9c9ede43f6bcb2881",
                "md5": "270fe1b4c90fa722af5eaa66fa380b68",
                "sha256": "0d002cb00ce5537cf520b0f19e7f7deb49d26b3086ca1224da2b3dc1d7e22866"
            },
            "downloads": -1,
            "filename": "fastcc-4.0.4-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "270fe1b4c90fa722af5eaa66fa380b68",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.13",
            "size": 15437,
            "upload_time": "2025-02-14T08:34:54",
            "upload_time_iso_8601": "2025-02-14T08:34:54.164366Z",
            "url": "https://files.pythonhosted.org/packages/97/2c/38150bcddf5e08d70cb27f82cdb6c9fa442c4bce76e9c9ede43f6bcb2881/fastcc-4.0.4-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "7b8f3b15209a56ed2edb82248ce38a5c279a65073a372f329ee00b0aaccb3183",
                "md5": "f9b73e72c691d5a72fc85acc161cbf77",
                "sha256": "d7adbe634e0f72cd89ae03e8c7389ab10fdbb4163552a9693b0578a3784e09ec"
            },
            "downloads": -1,
            "filename": "fastcc-4.0.4.tar.gz",
            "has_sig": false,
            "md5_digest": "f9b73e72c691d5a72fc85acc161cbf77",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.13",
            "size": 14471,
            "upload_time": "2025-02-14T08:34:56",
            "upload_time_iso_8601": "2025-02-14T08:34:56.093566Z",
            "url": "https://files.pythonhosted.org/packages/7b/8f/3b15209a56ed2edb82248ce38a5c279a65073a372f329ee00b0aaccb3183/fastcc-4.0.4.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-02-14 08:34:56",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "ReMi-HSBI",
    "github_project": "fastcc",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": false,
    "lcname": "fastcc"
}
        
Elapsed time: 0.40504s