⚡Mail Server implementation#

Should we reconsider our implementation method to align with the webhooks approach detailed in the documentation at mail.tm

Given the documentation at mail.tm, I've explored the possibility of implementing webhooks for our project. However,

It appears that the webhooks utilize SSE (Server-Sent Events) instead of conventional webhook captures for real-time event updates.
Unfortunately, due to the absence of reliable SSE client implementations in Python, the current implementation stands as the most suitable approach for now.

⚡Mail Server Base Implementation#

Note

Just below this text, lies the Base implementation of MailServer. You should never use this to create a MailServer instance directly. Instead you should choose MailServer

MailServerBase #

MailServerBase(
    server_auth: ServerAuth,
    pooling_rate: t.Optional[int],
    banner: t.Optional[bool] = True,
    banner_path: t.Optional[
        t.Union[pathlib.Path, str]
    ] = default_banner,
    suppress_errors: t.Optional[bool] = False,
    enable_logging: t.Optional[bool] = False,
)

Stellar Addition - Custom MailServer.

This script sets up a pooling-based server that checks the API every second for new events. When a difference is detected, the corresponding event is dispatched, allowing you to respond dynamically to incoming messages. In addition to the core SDK functionalities, this package offers an additional layer of scripts designed to handle clients in an event-driven manner, reminiscent of frameworks like discord.py or hikari. With this SDK, you gain access to a client that dispatches events seamlessly.

PARAMETER	DESCRIPTION
`server_auth`	The server authentication details. TYPE: `ServerAuth`
`pooling_rate`	The pooling rate in seconds. If not provided, the default pooling rate will be used. TYPE: `Optional[int]`
`banner`	Whether to display a banner upon initialization. Defaults to True. TYPE: `Optional[bool]` DEFAULT: `True`
`banner_path`	The path to the banner file. Defaults to the default banner file. TYPE: `Optional[Union[Path, str]]` DEFAULT: `default_banner`
`suppress_errors`	Whether to suppress errors. Defaults to False. TYPE: `Optional[bool]` DEFAULT: `False`
`enable_logging`	Whether to enable logging. Defaults to False. TYPE: `Optional[bool]` DEFAULT: `False`

Example

import mailtm
server = mailtm.MailServer(
    server_auth=ServerAuth(
        account_id="...",  # Your account ID.
        account_token="...",  # Your account token.
    )
)
# Define an event handler for new messages
# The stand-alone decorators should be used
# without the brackets since we provide no
# function to handle, and append with the
# handler.
@server.on_new_message
async def event(event: NewMessage):
    print(event.new_message.text)
# Start the event loop
server.run()

log #

log(
    message: str,
    severity: t.Literal[
        "INFO", "WARNING", "ERROR"
    ] = "INFO",
) -> None

Logs the message to the console with the corresponding severity.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`severity`	The severity of the message. Defaults to INFO. TYPE: `Literal[INFO, WARNING, ERROR]` DEFAULT: `'INFO'`

RETURNS	DESCRIPTION
`None`

subscribe #

subscribe(
    event_type: t.Type[BaseEvent],
) -> t.Callable[
    [t.Callable[[ServerSideEvents], t.Awaitable[None]]],
    t.Callable[[ServerSideEvents], t.Awaitable[None]],
]

Decorator to subscribe a function to handle server events.

PARAMETER	DESCRIPTION
`event_type`	The type of event to subscribe to. TYPE: `Type[BaseEvent]`

RETURNS	DESCRIPTION
`Callable`	The decorated function.

Example

@server.subscribe(ServerSideEvents.NewMessage)
async def event(event: ServerSideEvents.NewMessage):
    print(event.new_message.text)

on_new_message #

on_new_message(
    func: t.Callable[[NewMessage], t.Awaitable[None]]
)

Registers a callback function to handle new messages.

PARAMETER	DESCRIPTION
`func`	The callback function to handle new messages. TYPE: `Callable[[NewMessage], Awaitable[None]]`

RETURNS	DESCRIPTION
`The result of the extracted function call.`

on_new_domain #

on_new_domain(
    func: t.Callable[[DomainChange], t.Awaitable[None]]
)

Registers a callback function to handle new domains.

PARAMETER	DESCRIPTION
`func`	The callback function to handle new domains. TYPE: `Callable[[DomainChange], Awaitable[None]]`

RETURNS	DESCRIPTION
`The result of the extracted function call.`

dispatch `async` #

dispatch(event: BaseEvent) -> None

Asynchronously dispatches the given event to the appropriate handlers.

Args: self: The MailServerBase instance. event (BaseEvent): The event to dispatch.

Returns: None

shutdown `async` #

shutdown() -> None

Calls the shutdown method on the MailClient instance and dispatches the ServerCalledOff event.

Returns: None

runner `async` #

runner() -> None

This function is responsible for executing the main server logic. It starts a session and continuously polls the API for new events. When a difference is detected, the corresponding event is dispatched. The function runs in an infinite loop until it is interrupted by a keyboard interrupt.

RAISES	DESCRIPTION
`Exception`	If no events have been subscribed to.

run #

run() -> None

Executes the main server logic by running the event runner within asyncio event loop.

⚡The Mail Server#

MailServer #

MailServer(
    server_auth: ServerAuth,
    pooling_rate: int | None,
    banner: bool | None = True,
    banner_path: pathlib.Path | str | None = default_banner,
    suppress_errors: bool | None = False,
    enable_logging: bool | None = False,
)

Bases: MailServerBase

log #

log(
    message: str,
    severity: t.Literal[
        "INFO", "WARNING", "ERROR"
    ] = "INFO",
) -> None

Logs the message to the console with the corresponding severity.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`severity`	The severity of the message. Defaults to INFO. TYPE: `Literal[INFO, WARNING, ERROR]` DEFAULT: `'INFO'`

RETURNS	DESCRIPTION
`None`

subscribe #

subscribe(
    event_type: t.Type[BaseEvent],
) -> t.Callable[
    [t.Callable[[ServerSideEvents], t.Awaitable[None]]],
    t.Callable[[ServerSideEvents], t.Awaitable[None]],
]

Decorator to subscribe a function to handle server events.

PARAMETER	DESCRIPTION
`event_type`	The type of event to subscribe to. TYPE: `Type[BaseEvent]`

RETURNS	DESCRIPTION
`Callable`	The decorated function.

Example

@server.subscribe(ServerSideEvents.NewMessage)
async def event(event: ServerSideEvents.NewMessage):
    print(event.new_message.text)

on_new_message #

on_new_message(
    func: t.Callable[[NewMessage], t.Awaitable[None]]
)

Registers a callback function to handle new messages.

PARAMETER	DESCRIPTION
`func`	The callback function to handle new messages. TYPE: `Callable[[NewMessage], Awaitable[None]]`

RETURNS	DESCRIPTION
`The result of the extracted function call.`

on_new_domain #

on_new_domain(
    func: t.Callable[[DomainChange], t.Awaitable[None]]
)

Registers a callback function to handle new domains.

PARAMETER	DESCRIPTION
`func`	The callback function to handle new domains. TYPE: `Callable[[DomainChange], Awaitable[None]]`

RETURNS	DESCRIPTION
`The result of the extracted function call.`

dispatch `async` #

dispatch(event: BaseEvent) -> None

Asynchronously dispatches the given event to the appropriate handlers.

Args: self: The MailServerBase instance. event (BaseEvent): The event to dispatch.

Returns: None

run #

run() -> None

Executes the main server logic by running the event runner within asyncio event loop.

create_account `async` #

create_account(
    account_address: str, account_password: str
) -> t.Optional[Account]

Creates a new account with the given email address and password.

PARAMETER	DESCRIPTION
`account_address`	The email address of the new account. TYPE: `str`
`account_password`	The password for the new account. TYPE: `str`

RETURNS	DESCRIPTION
`Optional[Account]`	The newly created account object if successful, None otherwise.

delete_message `async` #

delete_message(message_id: str) -> None

Deletes a message from the Mail Box.

PARAMETER	DESCRIPTION
`message_id`	The ID of the message to be deleted. TYPE: `str`

RETURNS	DESCRIPTION
`None`

switch_account `async` #

switch_account(
    new_account_token: t.Union[Token, str]
) -> None

Switches to a new account.

PARAMETER	DESCRIPTION
`new_account_token`	The new account token to switch to. TYPE: `t.Union[Token, str]`

RETURNS	DESCRIPTION
`None`

delete_account `async` #

delete_account(account_id: str) -> None

Deletes an account.

PARAMETER	DESCRIPTION
`account_id`	The ID of the account to be deleted. TYPE: `str`

RETURNS	DESCRIPTION
`None`

shutdown `async` #

shutdown() -> None

Calls the shutdown method on the MailClient instance and dispatches the ServerCalledOff event.

Returns: None

runner `async` #

runner() -> None

This function is responsible for executing the main server logic. It starts a session and continuously polls the API for new events. When a difference is detected, the corresponding event is dispatched. The function runs in an infinite loop until it is interrupted by a keyboard interrupt.

RAISES	DESCRIPTION
`Exception`	If no events have been subscribed to.

⚡Mail Server implementation#

⚡Mail Server Base Implementation#

MailServerBase #

log #

subscribe #

on_new_message #

on_new_domain #

dispatch async #

shutdown async #

runner async #

run #

⚡The Mail Server#

MailServer #

log #

subscribe #

on_new_message #

on_new_domain #

dispatch async #

run #

create_account async #

delete_message async #

switch_account async #

delete_account async #

shutdown async #

runner async #

dispatch `async` #

shutdown `async` #

runner `async` #

dispatch `async` #

create_account `async` #

delete_message `async` #

switch_account `async` #

delete_account `async` #

shutdown `async` #

runner `async` #