fastcc 5.1.0

Lightweight, efficient and developer-friendly framework for mqtt communication.

FastCC

FastCC is a Python package that simplifies MQTT communication using decorators. With its intuitive @route system, developers can quickly define MQTT message handlers without boilerplate code. FastCC natively supports Protocol Buffers :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 which itself is built on top of paho-mqtt.

Examples

client.py

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

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

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

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)