Skip to main content

utils

get_cache_scope​

def get_cache_scope(cache_type: Optional[CacheType]) -> CacheScope

Helper to resolve the cache scope

Arguments:

  • cache_type: whether to pull the value from the specified cache specific store or the global one, defaults to the global one

run_user_handler​

async def run_user_handler(handler: Callable,
                           args: Sequence = [],
                           kwargs: dict = {})

Run a user-defined handler function. Runs sync functions in a threadpool.

Handles SystemExits cleanly.

Arguments:

  • handler: user-defined handler function
  • args: list of arguments to pass to the function
  • kwargs: dict of kwargs to past to the function

call_async​

def call_async(handler: Callable[..., Coroutine], *args)

Run an async function from a sync context.

Arguments:

  • handler: async function to run
  • args: arguments to pass to the function

import_config​

def import_config(config_path: str) -> Tuple[ModuleType, ConfigurationBuilder]

Import Dara from specified config in format "my_package.my_module:variable_name"

find_module_path​

def find_module_path(config_path: str)

Find the parent path to the module containing the config file wihout importing it.

enforce_sso​

def enforce_sso(conf: ConfigurationBuilder)

Checks if given configuration has SSO enabled.

Raises if SSO is not used

async_dedupe​

def async_dedupe(fn: Callable[..., Awaitable])

Decorator to deduplicate concurrent calls to asynchronous functions based on their arguments.

If multiple concurrent calls are made with the same arguments, the function is executed only once. Subsequent calls will wait for the first one to complete and then return its result.

This decorator is useful for operations that might be triggered multiple times in parallel but should be executed only once to prevent redundant work or data fetches.

resolve_exception_group​

def resolve_exception_group(error: Any)

Simplify an ExceptionGroup to a single exception if possible

Arguments:

  • error: The error to resolve