API Reference

class lato.Application(name='lato.application', dependency_provider=None, **kwargs)

Bases: ApplicationModule

Core Application class.

This class represents the core functionality of an application and extends ApplicationModule.

Parameters:

dependency_provider_factory – a factory that returns DependencyProvider instance, defaults to BasicDependencyProvider.

__init__(name='lato.application', dependency_provider=None, **kwargs)

Initialize the application instance.

Parameters:
  • name – Name of the application

  • dependency_provider (Optional[DependencyProvider]) – dependency provider DependencyProvider instance. Defaults to BasicDependencyProvider instance populated with kwargs.

  • kwargs – Additional keyword arguments to be passed to the dependency provider.

call(func, *args, **kwargs)

Invokes a function with args and kwargs within the TransactionContext. If func is a string, then it is an alias, and the corresponding handler for the alias is retrieved. Any missing arguments are provided by the dependency provider of a transaction context, and args and kwargs parameters.

Parameters:
  • func (Union[Callable[..., Any], str]) – The function to invoke, or an alias.

  • args – Arguments to pass to the function.

  • kwargs – Keyword arguments to pass to the function.

Return type:

Any

Returns:

The result of the invoked function.

Raises:

ValueError – If an alias is provided, but no corresponding handler is found.

async call_async(func, *args, **kwargs)

Invokes an async function with args and kwargs within the TransactionContext. If func is a string, then it is an alias, and the corresponding handler for the alias is retrieved. Any missing arguments are provided by the dependency provider of a transaction context, and args and kwargs parameters.

Parameters:
  • func (Union[Callable[..., Awaitable[Any]], str]) – The async function to invoke, or an alias.

  • args – Arguments to pass to the function.

  • kwargs – Keyword arguments to pass to the function.

Return type:

Any

Returns:

The result of the invoked function.

Raises:

ValueError – If an alias is provided, but no corresponding handler is found.

compose(alias)

Decorator for composing results of handlers identified by an alias.

Example:

>>> from lato import Application, Command, TransactionContext
>>> class SomeCommand(Command):
...     pass
>>>
>>> app = Application()
>>> @app.compose(SomeCommand)
... def middleware1(**kwargs):
...     ...
dependency_provider_factory

alias of BasicDependencyProvider

emit(event)

Deprecated. Use publish() instead.

Return type:

dict[Callable, Any]

execute(message)

Executes a command within the TransactionContext. Use handler() decorator to register a handler for the command. If a command is handled by multiple handlers, then the final result is composed to a single return value using compose() decorator.

Parameters:

message (Message) – The message to be executed (usually, a Command or Query subclass).

Return type:

Any

Returns:

The result of the invoked message handler.

Raises:

ValueError: If no handlers are found for the message.

async execute_async(message)

Asynchronously executes a command within the TransactionContext. Use handler() decorator to register a handler for the command. If a command is handled by multiple handlers, then the final result is composed to a single return value using compose() decorator.

Parameters:

message (Message) – The message to be executed (usually, a Command or Query subclass).

Return type:

Any

Returns:

The result of the invoked message handler.

Raises:

ValueError: If no handlers are found for the message.

get_dependency(identifier)

Gets a dependency from the dependency provider. Dependencies can be resolved either by name or by type.

Parameters:

identifier (Union[type, str]) – A string or a type representing the dependency.

Return type:

Any

Returns:

The resolved dependency.

on_create_transaction_context(func)

Decorator for overriding default transaction context creation

Parameters:

func – callback to be called when creating a transaction context

Returns:

the decorated function

Example:

>>> from lato import Application, TransactionContext
>>> app = Application()
>>>
>>> class CustomTransactionContext(TransactionContext):
...     pass
>>>
>>> @app.on_create_transaction_context
... def create_transaction_context(**kwargs):
...    return CustomTransactionContext(**kwargs)
>>>
>>> print(app.transaction_context(foo="bar").__class__.__name__)
CustomTransactionContext
on_enter_transaction_context(func)

Decorator for registering a function to be called when entering a transaction context

Parameters:

func – callback to be called when entering a transaction context

Returns:

the decorated function

Example:

>>> from lato import Application, TransactionContext
>>> app = Application()
>>> @app.on_enter_transaction_context
... def on_enter_transaction_context(ctx: TransactionContext):
...     print('entering transaction context')
...     ctx.set_dependencies(foo="foo")
>>> app.call(lambda foo: print(foo))
entering transaction context
foo
on_exit_transaction_context(func)

Decorator for registering a function to be called when exiting a transaction context

Parameters:

func – callback to be called when exiting a transaction context

Returns:

the decorated function

Example:

>>> from lato import Application, TransactionContext
>>> app = Application()
>>>
>>> @app.on_exit_transaction_context
... def on_exit_transaction_context(ctx: TransactionContext, exception=None):
...    print("exiting context")
>>> app.call(lambda: print("calling"))
calling
exiting context
publish(event)

Publish an event by calling all handlers for that event.

Parameters:

event (Event) – The event to publish, or an alias of an event handler to call.

Return type:

dict[Callable, Any]

Returns:

A dictionary mapping handlers to their results.

async publish_async(event)

Asynchronously publish an event by calling all handlers for that event.

Parameters:

event (Event) – The event to publish, or an alias of an event handler to call.

Return type:

dict[Callable, Any]

Returns:

A dictionary mapping handlers to their results.

transaction_context(**dependencies)

Creates a transaction context for the app.

The lifecycle of a transaction context is controlled by transaction_middleware(), on_enter_transaction_context(), on_exit_transaction_context() decorators.

Parameters:

dependencies – Keyword arguments to be passed as dependencies to the dependency provider of a transaction context, will override dependencies provided by the application.

Return type:

TransactionContext

Returns:

transaction context

transaction_middleware(middleware_func)

Decorator for registering a middleware function to be called when executing a function in a transaction context :type middleware_func: :param middleware_func: :return: the decorated function

Example:

>>> from typing import Callable
>>> from lato import Application, TransactionContext
>>>
>>> app = Application()
>>> @app.transaction_middleware
... def middleware1(ctx: TransactionContext, call_next: Callable):
...     ...
class lato.ApplicationModule(name)
__init__(name)

Initialize the application module instance.

Parameters:

name (str) – Name of the module

handler(alias)

Decorator for registering a handler. Handler can be aliased by a name or by a message type.

Parameters:

alias (Union[type[Message], str]) – lato.Message or a string.

Example #1:

>>> from lato import Application, ApplicationModule
>>> my_module = ApplicationModule("my_module")
>>>
>>> @my_module.handler("my_handler")
... def my_handler():
...     print("handler called")
>>>
>>> app = Application("example")
>>> app.include_submodule(my_module)
>>> app.call("my_handler")
handler called

Example #2:

>>> from lato import ApplicationModule, Command
>>> class MyCommand(Command):
...     pass
>>>
>>> my_module = ApplicationModule("my_module")
>>> @my_module.handler(MyCommand)
... def my_handler(command: MyCommand):
...     print("command handler called")
>>>
>>> app = Application("example")
>>> app.include_submodule(my_module)
>>> app.execute(MyCommand())
command handler called
include_submodule(a_module)

Adds a child submodule to this module.

Parameters:

a_module (ApplicationModule) – child module to add

class lato.DependencyProvider

A dependency provider interface that provides dependencies and helps in automatic dependency injection based on type or parameter name.

Parameters:
  • allow_namesTrue if dependency resolution by name is supported. Defaults to True

  • allow_typesTrue if dependency resolution by type is supported. Defaults to True

abstract copy(*args, **kwargs)

Creates a copy of self with updated dependencies.

Parameters:
  • args – dependencies to update, identified by type.

  • kwargs – dependencies to update, identified by name and type.

Return type:

DependencyProvider

Returns:

A copy of the dependency provider.

get_dependency(identifier)

Retrieve a dependency using its identifier (name or type).

Parameters:

identifier (Union[type, str]) – Identifier for the dependency

Return type:

Any

Returns:

The associated dependency

abstract has_dependency(identifier)

Check if a dependency with the given identifier exists.

Parameters:

identifier (Union[type, str]) – Identifier for the dependency

Return type:

bool

Returns:

True if the dependency exists, otherwise False

abstract register_dependency(identifier, dependency)

Register a dependency with a given identifier (name or type).

Parameters:
  • identifier (Union[type, str]) – The name or type to be used as an identifier for the dependency

  • dependency (Any) – The actual dependency

resolve_func_params(func, func_args=None, func_kwargs=None)

Resolves parameters of a function, by matching function parameters to dependencies.

Parameters:
  • func (Callable) – The function to get arguments for

  • func_args (Any) – Positional arguments to the function

  • func_kwargs (Any) – Keyword arguments to the function

Return type:

dict[str, Any]

Returns:

A dictionary of resolved dependencies, where the key is the name of the parameter and the value is the resolved dependency.

update(*args, **kwargs)

Updates the dependency provider with new dependencies.

Parameters:
  • args – dependencies to update, identified by type

  • kwargs – dependencies to update, identified by name and type

class lato.TransactionContext(dependency_provider=None, *args, **kwargs)

Transaction context is a context manager for handler execution.

Parameters:

dependency_provider_factory – a factory that returns DependencyProvider instance, defaults to BasicDependencyProvider.

Example: >>> from lato import TransactionContext >>> >>> def my_function(param1, param2): … print(param1, param2) >>> >>> with TransactionContext(param1=”foo”) as ctx: … ctx.call(my_function, param2=”bar”) foo bar

__init__(dependency_provider=None, *args, **kwargs)

Initialize the transaction context instance.

Parameters:
  • dependency_provider (Optional[DependencyProvider]) – dependency provider DependencyProvider instance. Defaults to BasicDependencyProvider instance populated with args and kwargs.

  • args – Additional positional arguments to be passed to the dependency provider.

  • kwargs – Additional keyword arguments to be passed to the dependency provider.

begin()

Starts a transaction by calling on_enter_transaction_context callback.

The callback could be used to set up the transaction-level dependencies (i.e. current time, transaction id), or to start the database transaction.

async begin_async()

Asynchronously starts a transaction by calling async on_enter_transaction_context callback.

The callback could be used to set up the transaction-level dependencies (i.e. current time, transaction id), or to start the database transaction.

call(func, *func_args, **func_kwargs)

Call a function with the arguments and keyword arguments. Missing arguments will be resolved with the dependency provider.

Parameters:
  • func (Callable) – The function to call.

  • func_args (Any) – Positional arguments to pass to the function.

  • func_kwargs (Any) – Keyword arguments to pass to the function.

Return type:

Any

Returns:

The result of the function call.

async call_async(func, *func_args, **func_kwargs)

Call an async function with the arguments and keyword arguments. Missing arguments will be resolved with the dependency provider.

Parameters:
  • func (Callable[..., Awaitable[Any]]) – The function to call.

  • func_args (Any) – Positional arguments to pass to the function.

  • func_kwargs (Any) – Keyword arguments to pass to the function.

Return type:

Any

Returns:

The result of the function call.

configure(on_enter_transaction_context=None, on_exit_transaction_context=None, middlewares=None, composers=None, handlers_iterator=None)

Customize the behavior of the transaction context with callbacks, middlewares, and composers.

Parameters:
  • on_enter_transaction_context (Optional[Callable[[TransactionContext], Awaitable[None]]]) – Optional; Function to be called when entering a transaction context.

  • on_exit_transaction_context (Optional[Callable[[TransactionContext, Optional[Exception]], Awaitable[None]]]) – Optional; Function to be called when exiting a transaction context.

  • middlewares (Optional[list[Callable[[TransactionContext, Callable], Awaitable[Any]]]]) – Optional; List of middleware functions to be applied.

  • composers (Optional[dict[Union[type[Message], str], Callable[..., Callable]]]) – Optional; List of composers functions to be applied.

  • handlers_iterator (Optional[Callable[[Union[type[Message], str]], Iterator[MessageHandler]]]) – Optional; Function to iterate over handlers.

property current_action: tuple[Message, Callable]

Returns current message and handler being executed

dependency_provider_factory

alias of BasicDependencyProvider

end(exception=None)

Ends the transaction context by calling on_exit_transaction_context callback, optionally passing an exception.

The callback could be used to commit/end a database transaction.

Parameters:

exception (Optional[Exception]) – Optional; The exception to handle at the end of the transaction, if any.

async end_async(exception=None)

Ends the transaction context by calling on_exit_transaction_context callback, optionally passing an exception.

The callback could be used to commit/end a database transaction.

Parameters:

exception (Optional[Exception]) – Optional; The exception to handle at the end of the transaction, if any.

execute(message)

Executes all handlers bound to the message. Returns a tuple of handlers’ return values.

Parameters:

message (Message) – The message to be executed.

Return type:

tuple[Any, ...]

Returns:

a tuple of return values from executed handlers

Raises:

ValueError: If no handlers are found for the message.

async execute_async(message)

Executes all async handlers bound to the message. Returns a tuple of handlers’ return values.

Parameters:

message (Message) – The message to be executed.

Return type:

tuple[Any, ...]

Returns:

a tuple of return values from executed handlers

Raises:

ValueError: If no handlers are found for the message.

get_dependency(identifier)

Gets a dependency from the dependency provider

Return type:

Any

publish(message, *args, **kwargs)

Publish a message by calling all handlers for that message.

Parameters:
  • message (Union[str, Message]) – The message object to publish, or an alias of a handler to call.

  • args – Positional arguments to pass to the handlers.

  • kwargs – Keyword arguments to pass to the handlers.

Return type:

dict[MessageHandler, Any]

Returns:

A dictionary mapping handlers to their results.

async publish_async(message, *args, **kwargs)

Asynchronously publish a message by calling all handlers for that message.

Parameters:
  • message (Union[str, Message]) – The message object to publish, or an alias of a handler to call.

  • args – Positional arguments to pass to the handlers.

  • kwargs – Keyword arguments to pass to the handlers.

Return type:

dict[MessageHandler, Awaitable[Any]]

Returns:

A dictionary mapping handlers to their results.

set_dependencies(**kwargs)

Sets multiple dependencies at once

set_dependency(identifier, dependency)

Sets a dependency in the dependency provider

Return type:

None