Skip to content

prefect.infrastructure.provisioners

get_infrastructure_provisioner_for_work_pool_type(work_pool_type)

Retrieve an instance of the infrastructure provisioner for the given work pool type.

Parameters:

Name Type Description Default
work_pool_type str

the work pool type

required

Returns:

Type Description
Type[Provisioner]

an instance of the infrastructure provisioner for the given work pool type

Raises:

Type Description
ValueError

if the work pool type is not supported

Source code in src/prefect/infrastructure/provisioners/__init__.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_infrastructure_provisioner_for_work_pool_type(
    work_pool_type: str,
) -> Type[Provisioner]:
    """
    Retrieve an instance of the infrastructure provisioner for the given work pool type.

    Args:
        work_pool_type: the work pool type

    Returns:
        an instance of the infrastructure provisioner for the given work pool type

    Raises:
        ValueError: if the work pool type is not supported
    """
    provisioner = _provisioners.get(work_pool_type)
    if provisioner is None:
        raise ValueError(f"Unsupported work pool type: {work_pool_type}")
    return provisioner()