Metadata-Version: 2.4
Name: fastcc
Version: 5.1.0
Summary: Lightweight, efficient and developer-friendly framework for mqtt communication.
Author-email: "J. Baudisch" <justin.baudisch@hsbi.de>
Maintainer-email: "J. Baudisch" <justin.baudisch@hsbi.de>
Project-URL: Repository, https://github.com/ReMi-HSBI/fastcc
Keywords: mqtt,protobuf,aiomqtt,asyncio,iot
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development
Classifier: Topic :: Communications
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Typing :: Typed
Requires-Python: >=3.14
Description-Content-Type: text/markdown
Requires-Dist: paho-mqtt
Requires-Dist: aiomqtt
Requires-Dist: protobuf
Provides-Extra: dev
Requires-Dist: ruff; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: Sphinx; extra == "dev"
Requires-Dist: furo; extra == "dev"
Requires-Dist: types-protobuf; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"

<p align="center">
    <img
        src="https://github.com/ReMi-HSBI/fastcc/blob/main/docs/src/static/images/fastcc_logo.svg?raw=true"
        alt="FastCC Logo"
        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>

FastCC is a [Python](https://www.python.org) package that simplifies
[MQTT](https://mqtt.org) communication using decorators. With its
intuitive `@route` system, developers can quickly define MQTT message
handlers without boilerplate code. FastCC natively supports
[Protocol Buffers](https://protobuf.dev) :boom:, automatically handling
serialization to byte format for efficient and structured data exchange.

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

This project is built on top of [aiomqtt](https://github.com/empicano/aiomqtt)
which itself is built on top of [paho-mqtt](https://eclipse.dev/paho).

# Examples

## client.py

```python
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.DEBUG)

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

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


loop_factory: type[asyncio.AbstractEventLoop] | None = None

# See: https://github.com/empicano/aiomqtt#note-for-windows-users
if sys.platform.lower() == "win32" or os.name.lower() == "nt":
    loop_factory = asyncio.SelectorEventLoop

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

## app.py

```python
import asyncio
import contextlib
import logging
import os
import sys

import fastcc

router = fastcc.Router()


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

    Parameters
    ----------
    packet
        The name of the user.
        Autofilled by fastcc.
    family
        The family 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[packet] += 1
    occurrence = database[packet]
    return (
        f"Hello, {packet} from the {family} family! Saw you {occurrence} times!"
    )


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

    database: dict[str, int] = {"Alice": 0, "Bob": 0}

    async with fastcc.Application("localhost") as app:
        app.add_router(router)
        app.add_injector(database=database)
        app.add_exception_handler(
            KeyError,
            lambda e: fastcc.RouteExecutionError(repr(e), 404),
        )
        await app.run()


loop_factory: type[asyncio.AbstractEventLoop] | None = None

# See: https://github.com/empicano/aiomqtt#note-for-windows-users
if sys.platform.lower() == "win32" or os.name.lower() == "nt":
    loop_factory = asyncio.SelectorEventLoop

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

## stream_client.py

```python
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.DEBUG)

    async with fastcc.Client("localhost") as client:
        try:
            async for response in client.stream(
                "greet",
                "Charlie",
                response_type=str,
            ):
                _logger.info("response: %r", response)
        except fastcc.RouteExecutionError as e:
            details = f"An error occurred on the server: {e}"
            _logger.error(details)

        async for response in client.stream(
            "greet",
            "Alice",
            response_type=str,
        ):
            _logger.info("response: %r", response)


loop_factory: type[asyncio.AbstractEventLoop] | None = None

# See: https://github.com/empicano/aiomqtt#note-for-windows-users
if sys.platform.lower() == "win32" or os.name.lower() == "nt":
    loop_factory = asyncio.SelectorEventLoop

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

## app.py

```python
import asyncio
import contextlib
import logging
import os
import sys
from collections.abc import AsyncIterator  # noqa: TC003

import fastcc

router = fastcc.Router()


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

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

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

    for _ in range(2):
        database[packet] += 1
        occurrence = database[packet]
        yield f"Hello, {packet}! Saw you {occurrence} times!"


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

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

        await app.run()


loop_factory: type[asyncio.AbstractEventLoop] | None = None

# See: https://github.com/empicano/aiomqtt#note-for-windows-users
if sys.platform.lower() == "win32" or os.name.lower() == "nt":
    loop_factory = asyncio.SelectorEventLoop

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