The App class

The main App class manages the asyncio loop, the servers and events.

It can be imported using:

from mara import App

API reference

class mara.app.app.App

Orchestrate servers, clients and tasks

add_server(server: AbstractServer) → AbstractServer

Add a new Server instance to the async loop

The server will start listening when the app is run(). If it is already running, it will start listening immediately.

add_timer(timer: AbstractTimer) → AbstractTimer

Add a new Timer instance to the async loop

The timer will start when the app is run(). If it is already running, it will start immediately.

Returns the timer instance. Because timer instances are callable decorators, you can use this as a decorator to define a timer and its function in one:

@app.add_timer(PeriodicTimer(every=1))
async def tick(timer):
    ...

which is shorthand for:

async def tick(timer):
    ...

timer = PeriodicTimer(every=60)
app.add_timer(timer)
timer(tick)

listen(event_class: type[Event], handler: event_manager.HandlerType | None = None, **filters: event_manager.FilterType)

Bind a handler callback to the specified event class, and to its subclasses

Parameters

• event_class (Type[Event]) – The Event class to listen for

• handler (Awaitable | None) – The handler, if not being decorated

• **filters – Key value pairs to match against inbound events

Can be called directly:

app.listen(Event, handler)

or can be called as a decorator with no handler argument:

@app.listen(Event)
async def callback(event):
    ...

remove_server(server: AbstractServer)

Stop and remove the specified Server instance

run(debug=True)

Start the main app async loop

This will start any Servers which have been added with add_server()

property status

Get the status of the app and its servers

Difference between this and self._status is when the app loop is RUNNING, this will return STARTING until all servers are listening

stop()

This will ask the main loop to stop, shutting down all servers, connections and other async tasks.