Typer Entry Point

Description

An entry point for the python portion of the RLITT generator toolchain. The entry point exposes a command line interface utilizing the Typer library. Two interfaces are offered, one that starts the tool in producer mode and that starts the tool in worker mode.

Public Interfaces

Producer Mode

In producer mode the tool generates and enqueues new jobs.

State machine

stateDiagram-v2
    state "Load configuration." as lc
    state "Generate jobs" as gj

    state if_state <<choice>>

    [*]--> lc
    lc --> gj
    gj --> if_state
    if_state --> gj: No error detected
    if_state --> [*]: Error state

Worker Mode

The worker mode for the tool subscribes to and

State machine

stateDiagram-v2
    state "Load configuration." as lc
    state "Get job" as gj
    state "Proccess job" as pj

    state if_state <<choice>>

    [*]--> lc
    lc --> gj
    gj --> pj
    pj --> if_state
    if_state --> gj: No error detected
    if_state --> [*]: Error state

Unit test description

Implementation

runner.__main__

Main entry point for the RLITT generator toolchain.

app module-attribute

app = Typer()

client

client(config)

Client entry point.

Parameters:

Name	Type	Description	Default
config	Path	The path to a configuration file.	required

Source code in runner/__main__.py

@app.command()
def client(config: Path):
    """Client entry point.

    Args:
        config: The path to a configuration file.

    """
    cfg.load(config)
    fproducer.faktory_producer()

worker

worker(config)

Worker entry point.

Parameters:

Name	Type	Description	Default
config	Path	The path to a configuration file.	required

Source code in runner/__main__.py

@app.command()
def worker(config: Path):
    """Worker entry point.

    Args:
        config: The path to a configuration file.

    """
    cfg.load(config)
    with Client(
        faktory_url=f'tcp://{cfg.cfg_dict["faktory-connection-info"]["domain"]}:{cfg.cfg_dict["faktory-connection-info"]["port"]}',
        role='consumer',
    ) as client:
        consumer = Consumer(client=client, queues=['arborescent'], concurrency=1)
        consumer.register('arbor_job', fworker.faktory_job)
        consumer.run()