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.
Return type:: Callable

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_names – True if dependency resolution by name is supported. Defaults to True
allow_types – True 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