| Name | fastcc JSON |
| Version |
4.0.4
JSON |
| download |
| home_page | None |
| Summary | Lightweight, efficient and developer-friendly framework for component communication |
| upload_time | 2025-02-14 08:34:56 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.13 |
| license | None |
| 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"
}
