Example Project Structure

Startup File

from arconq_sdk import AppBuilder, Dependency, arconq_app
from containers.container import Container
from message_handlers import example_message_handler
from routers import examples_router
from scheduled_job import example_scheduled_job
from background_jobs import example_background_job


@arconq_app
def main(
    container: Container,
    app_builder: AppBuilder = Dependency(Container.app_builder),
) -> None:
    app_builder.configure_arconq(container)

    # registrations of a router, message handler, a scheduled job and a background job (parts of the example app)
    app_builder.add_web_api_router(container, examples_router, examples_router.router, "/router")
    app_builder.add_scheduled_job(container, example_scheduled_job, example_scheduled_job.job)
    app_builder.add_background_job(container, example_background_job, example_background_job.job)
    app_builder.add_message_handler(container, example_message_handler, example_message_handler.handler)

    app = app_builder.build_app()

    app.run()


if __name__ == "__main__":
    # usage of the Container class (more information below)
    container = Container()
    container.wire(modules=[__name__])
    main(container)

Container Class

Registrations of services and other classes that can be injected to message handlers, scheduled jobs and routers should be put here. It's going to make possible injecting them via dependency injection mechanism. Each registered service need to get passed the parameters defined in their constructors.

from dependency_injector import providers
from arconq_sdk.dependency_injection.base_container import BaseContainer
from services.hello_world_service import HelloWorldService
from services.random_int_service import RandomIntService
from services.messaging_service import MessagingService

class Container(BaseContainer):
    random_int_service = providers.Factory(RandomIntService, BaseContainer.logger)
    hello_world_service = providers.Factory(HelloWorldService, random_int_service)
    messaging_service = providers.Factory(MessagingService, BaseContainer.logger, BaseContainer.message_bus)

Example Router

You need to create a Router object and use it to decorate a function. In the highlighted lines two injections are happening:

Logger - part of the ArconQ SDK
HelloWorldService - part of the example app

Since it's a function and not a class receiving the injections, there's a need to use the Dependency function and refer to the injected class from the Container.

from arconq_sdk import Dependency, Logger
from arconq_sdk.web_api.router import Router
from containers.container import Container
from services.hello_world_service import HelloWorldService

router = Router()
tag = "Examples"

@router.http_get("", response_model=dict[str, str], tags=[tag])
async def get_examples_async(
    logger: Logger = Dependency(Container.logger),
    hello_world_service: HelloWorldService = Dependency(Container.hello_world_service),
):
    greeting_phrase = hello_world_service.generate_greeting_phrase()
    logger.info(greeting_phrase)
    return {"greeting_phrase": greeting_phrase}

Example Services

Here is the HelloWorldService definition. It also is getting injected another service called RandomIntService. Notice a different syntax. When injecting to a class constructor, there's no need to use the Dependency function. Putting the parameter in constructor is enough. Remember the service registration in the Container class.

from services.random_int_service import RandomIntService

class HelloWorldService:
    def __init__(self, random_int_service: RandomIntService) -> None:
        self.random_int_service = random_int_service

    def generate_greeting_phrase(self) -> str:
        return f"Hello World! {self.random_int_service.generate_random_int()}"

RandomIntService definition:

import random
from arconq_sdk import Logger

class RandomIntService:
    def __init__(self, logger: Logger) -> None:
        self.logger = logger

    def generate_random_int(self) -> int:
        self.logger.info("Generating new random integer")
        return random.choice([1, 2, 3])

Example Message

This class represents payload used in the examples below the Scheduled Job sends it and the Message Handler receives it.

from arconq_sdk.contracts.message import Message


class ExampleMessage(Message):
    Id: int
    Description: str
    Number: int

Example Scheduled Job

You need to define a ScheduledJob object and use it to decorate a function. In the highlighted lines two injections are happening:

Logger - part of the ArconQ SDK
MessagingService - part of the example app

For more information about scheduling check this article.

from arconq_sdk import Dependency, Logger, ScheduledJob
from containers.container import Container
from services.messaging_service import MessagingService

job = ScheduledJob()

@job.schedule("/example-job", cron_schedule="0/10 * * ? * * *")
async def example_scheduled_job(
    logger: Logger = Dependency(Container.logger), 
    messaging_service: MessagingService = Dependency(Container.messaging_service)
):
    logger.info("Scheduled job was executed")
    await messaging_service.send_message()

Below you can find the MessagingService definition. It uses the MessageBus (part of ArconQ SDK) to publish a message.
Remember the service registration in the Container class.

For more information about sending messages please check this article.

from arconq_sdk import Logger, MessageBus
import random
from contracts.example_message import ExampleMessage

class MessagingService:
    def __init__(self, logger: Logger, message_bus: MessageBus) -> None:
        self.logger = logger
        self.message_bus = message_bus

    async def send_message(self) -> None:
        message = ExampleMessage(
            Id=1,
            Description="message sent by a Python app",
            Number=random.randint(1, 100),
        )
        self.logger.info("Sending a message: {Message}", Message=message)
        await self.message_bus.publish_async(message)

Example Background Job

You need to define a BackgroundJob object and use it to decorate a function. In the highlighted lines two injections are happening:

Logger - part of the ArconQ SDK
MessagingService - part of the example app

For more information about background jobs check this article.

import asyncio

from arconq_sdk import Dependency, Logger, BackgroundJob
from containers.container import Container
from services.messaging_service import MessagingService

job = BackgroundJob()


@job.register(name="example background job")
async def example_background_job(
    logger: Logger = Dependency(Container.logger), 
    messaging_service: MessagingService = Dependency(Container.messaging_service)
):
    while True:
        await messaging_service.send_message()
        logger.info("Message has been sent")
        await asyncio.sleep(10.0)

Example Message Handler

You need to create a MessageHandler object and use it to decorate a function. Type of message that will be handled needs to be passed in the decorator parameters message_type=ExampleMessage. The method parameters:

message - the message that the handler will receive
logger - part of the ArconQ SDK, injected via dependency injection.

For more information about receiving messages please check this article

from arconq_sdk import Dependency, Logger, MessageHandler
from containers.container import Container
from contracts.example_message import ExampleMessage

handler = MessageHandler()

@handler.handle("/example-handler", message_type=ExampleMessage)
async def example_message_handler(
    message: ExampleMessage,
    logger: Logger = Dependency(Container.logger),
):
    logger.info("Handler received a message of content: {Message}", Message=message)

Example Project Structure

Startup File​

Container Class​

Example Router​

Example Services​

Example Message​

Example Scheduled Job​

Example Background Job​

Example Message Handler​