Skip to content

Utilities

Helper functions for logging, context access, and URL handling.

Logging Initialisation

logging_init(cfg_path=None)

Source code in ivcap_lambda/logger.py
def logging_init(cfg_path: str | None = None):
    if not cfg_path:
        script_dir = os.path.dirname(__file__)
        cfg_path = os.path.join(script_dir, "logging.json")

    with open(cfg_path) as file:
        config = json.load(file)
        set_service_log_config(config)

Context Helpers

These helpers let you access the current job's context from anywhere in your code — useful in helper functions that don't have direct access to JobContext.

get_event_reporter()

Get the current event reporter from the job context.

Source code in ivcap_lambda/executor.py
def get_event_reporter() -> EventReporter | None:
    """Get the current event reporter from the job context."""
    jctxt = job_context.get()
    return jctxt.report if jctxt is not None else None

get_job_id()

Get the current job ID from the job context.

Source code in ivcap_lambda/executor.py
def get_job_id() -> str | None:
    """Get the current job ID from the job context."""
    jctxt = job_context.get()
    return jctxt.job_id if jctxt is not None else None

URL Utilities

get_public_url_prefix(req)

Return the public url prefix for req.

First checks for a Forwarded http header and if absent uses the request's base_ur; variable.

Parameters:

Name Type Description Default
req Request

A FastAPI request instance

required

Returns:

Name Type Description
str str

A url as string

Source code in ivcap_lambda/utils.py
def get_public_url_prefix(req: Request) -> str:
    """Return the public url prefix for `req`.

    First checks for a `Forwarded` http header and if
    absent uses the request's `base_ur;` variable.

    Args:
        req (Request): A FastAPI request instance

    Returns:
        str: A url as string
    """
    fw = get_forwarded_header(req)
    if fw is not None:
        prefix = f"{fw.get('proto', 'http')}:://{fw.get('for')}"
    else:
        prefix = str(req.base_url).rstrip("/")
    return prefix