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[MessageHandler, 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[MessageHandler, 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[MessageHandler, 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.

If func is coroutine, or any of the middleware functions is coroutine, TypeError will be raised.

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.

Edge cases: - middlewares and func are sync - this will behave like call() - middleware is sync, and call_next is async - will raise TypeError, as middleware will not be able to wait for call_next()

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

is_async_context_manager()¶

Determine if the transaction context requires async with context manager.

This method checks if either the on_enter or on_exit callbacks associated with the transaction context are asynchronous.

Return type:: bool
Returns:: True if on_enter or on_exit are asynchronous callbacks, False otherwise.

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