Coverage for /usr/local/lib/python3.12/site-packages/prefect/client/collections.py: 57%

1from typing import Any, Dict, Optional 1 ctx1a

2 

3from typing_extensions import Protocol 1 ctx1a

4 

5from prefect.client.cloud import get_cloud_client 1 ctx1a

6from prefect.client.orchestration import ServerType, get_client 1 ctx1a

7 

8 

9class CollectionsMetadataClient(Protocol): 1 ctx1a

10 async def read_worker_metadata(self) -> Dict[str, Any]: ... 10 ↛ exitline 10 didn't return from function 'read_worker_metadata' because 1 ctx1a

11 

12 async def __aenter__(self) -> "CollectionsMetadataClient": ... 12 ↛ exitline 12 didn't return from function '__aenter__' because 1 ctx1a

13 

14 async def __aexit__(self, *exc_info: Any) -> Any: ... 14 ↛ exitline 14 didn't return from function '__aexit__' because 1 ctx1a

15 

16 

17def get_collections_metadata_client( 1 ctx1a

18 httpx_settings: Optional[Dict[str, Any]] = None, 

19) -> "CollectionsMetadataClient": 

20 """ 

21 Creates a client that can be used to fetch metadata for 

22 Prefect collections. 

23 

24 Will return a `CloudClient` if profile is set to connect 

25 to Prefect Cloud, otherwise will return an `OrchestrationClient`. 

26 """ 

27 orchestration_client = get_client(httpx_settings=httpx_settings) 

28 if orchestration_client.server_type == ServerType.CLOUD: 

29 return get_cloud_client(httpx_settings=httpx_settings, infer_cloud_url=True) 

30 else: 

31 return orchestration_client