CarbonAware Prefect API

CarbonAware Delay Decorator

Decorator to delay execution until a CO2-optimal time.

If region and provider are not specified, and they can't be detected automatically, a warning will be logged and no delay will be applied.

Parameters:

Name	Type	Description	Default
window	timedelta	The maximum delay to wait for an optimal time.	timedelta(hours=6)
duration	timedelta	The duration of the job.	timedelta(minutes=30)
region	str | None	The region of the cloud zone. If not specified, it will be detected automatically.	None
provider	str | None	The provider of the cloud zone. If not specified, it will be detected automatically.	None

Returns:

Type	Description
	A decorator function that can be applied to any function to delay its execution.

Source code in .venv/lib/python3.13/site-packages/carbonaware_prefect/decorator.py

def carbonaware_delay_decorator(
    window: timedelta = timedelta(hours=6),
    duration: timedelta = timedelta(minutes=30),
    region: str | None = None,
    provider: str | None = None,
):
    """Decorator to delay execution until a CO2-optimal time.

    If region and provider are not specified, and they can't be detected automatically,
    a warning will be logged and no delay will be applied.

    Args:
        window: The maximum delay to wait for an optimal time.
        duration: The duration of the job.
        region: The region of the cloud zone. If not specified, it will be detected automatically.
        provider: The provider of the cloud zone. If not specified, it will be detected automatically.

    Returns:
        A decorator function that can be applied to any function to delay its execution.
    """

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Apply the carbon-aware delay
            carbonaware_delay(
                window=window,
                duration=duration,
                region=region,
                provider=provider,
            )

            # Execute the wrapped function
            return func(*args, **kwargs)

        return wrapper

    return decorator

CarbonAware Delay Task

Creates a Prefect task that delays execution until a CO2-optimal time.

If region and provider are not specified, and they can't be detected automatically, a warning will be logged and no delay will be applied.

Parameters:

Name	Type	Description	Default
window	timedelta	The maximum delay to wait for an optimal time.	timedelta(hours=6)
duration	timedelta	The duration of the job.	timedelta(minutes=30)
region	str | None	The region of the cloud zone. If not specified, it will be detected automatically.	None
provider	str | None	The provider of the cloud zone. If not specified, it will be detected automatically.	None
**task_kwargs		Additional keyword arguments to pass to the Prefect task.	{}

Returns:

Type	Description
	A Prefect task that delays execution until a CO2-optimal time.

Source code in .venv/lib/python3.13/site-packages/carbonaware_prefect/task.py

def carbonaware_delay_task(
    window: timedelta = timedelta(hours=6),
    duration: timedelta = timedelta(minutes=30),
    region: str | None = None,
    provider: str | None = None,
    **task_kwargs,
):
    """
    Creates a Prefect task that delays execution until a CO2-optimal time.

    If region and provider are not specified, and they can't be detected automatically,
    a warning will be logged and no delay will be applied.

    Args:
        window: The maximum delay to wait for an optimal time.
        duration: The duration of the job.
        region: The region of the cloud zone. If not specified, it will be detected automatically.
        provider: The provider of the cloud zone. If not specified, it will be detected automatically.
        **task_kwargs: Additional keyword arguments to pass to the Prefect task.

    Returns:
        A Prefect task that delays execution until a CO2-optimal time.
    """

    # Create a task that performs the carbon-aware delay
    @task(**task_kwargs)
    def _carbonaware_delay_task():
        carbonaware_delay(
            window=window,
            duration=duration,
            region=region,
            provider=provider,
        )

    return _carbonaware_delay_task