Coverage for /usr/local/lib/python3.12/site-packages/prefect/infrastructure/provisioners/__init__.py: 45%
28 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 11:21 +0000
1from typing import TYPE_CHECKING, Any, Dict, Optional, Protocol, Type 1a
3import rich.console 1a
5from prefect._internal.lazy import LazyDict 1a
7if TYPE_CHECKING: 7 ↛ 8line 7 didn't jump to line 8 because the condition on line 7 was never true1a
8 from prefect.client.orchestration import PrefectClient
11def _load_provisioners() -> dict[str, type]: 1a
12 """Lazy load provisioners to avoid importing heavy cloud SDKs at module import time."""
13 from prefect.infrastructure.provisioners.coiled import CoiledPushProvisioner
14 from prefect.infrastructure.provisioners.modal import ModalPushProvisioner
16 from .cloud_run import CloudRunPushProvisioner
17 from .container_instance import ContainerInstancePushProvisioner
18 from .ecs import ElasticContainerServicePushProvisioner
20 return {
21 "cloud-run:push": CloudRunPushProvisioner,
22 "cloud-run-v2:push": CloudRunPushProvisioner,
23 "azure-container-instance:push": ContainerInstancePushProvisioner,
24 "ecs:push": ElasticContainerServicePushProvisioner,
25 "modal:push": ModalPushProvisioner,
26 "coiled:push": CoiledPushProvisioner,
27 }
30_provisioners_lazy: LazyDict[str, type] = LazyDict(_load_provisioners) 1a
33def __getattr__(name: str) -> LazyDict[str, type]: 1a
34 """Lazy load module attributes."""
35 if name == "_provisioners":
36 return _provisioners_lazy
37 raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
40class Provisioner(Protocol): 1a
41 @property 1a
42 def console(self) -> rich.console.Console: ... 42 ↛ exitline 42 didn't return from function 'console' because 1a
44 @console.setter 1a
45 def console(self, value: rich.console.Console) -> None: ... 45 ↛ exitline 45 didn't return from function 'console' because 1a
47 async def provision( 47 ↛ exitline 47 didn't return from function 'provision' because 1a
48 self,
49 work_pool_name: str,
50 base_job_template: Dict[str, Any],
51 client: Optional["PrefectClient"] = None,
52 ) -> Dict[str, Any]: ...
55def get_infrastructure_provisioner_for_work_pool_type( 1a
56 work_pool_type: str,
57) -> Type[Provisioner]:
58 """
59 Retrieve an instance of the infrastructure provisioner for the given work pool type.
61 Args:
62 work_pool_type: the work pool type
64 Returns:
65 an instance of the infrastructure provisioner for the given work pool type
67 Raises:
68 ValueError: if the work pool type is not supported
69 """
70 provisioner = _provisioners_lazy.get(work_pool_type)
71 if provisioner is None:
72 raise ValueError(f"Unsupported work pool type: {work_pool_type}")
73 return provisioner()