fastcc 5.2.2
Lightweight, efficient and developer-friendly framework for mqtt communication.
Requires Python: >=3.14
Keywords: mqtt, protobuf, aiomqtt, asyncio, iot

<p align="center">
    <img
        src="https://github.com/ReMi-HSBI/fastcc/blob/main/docs/src/static/images/logos/fastcc.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.RequestError 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.RequestError(repr(e), fastcc.Status.NOT_FOUND),
        )
        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.RequestError 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)
```

## stream_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.RequestError(repr(e), fastcc.Status.NOT_FOUND),
        )

        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)
```