Coverage for /Users/ajo/work/jumpstarter/jumpstarter/packages/jumpstarter/jumpstarter/utils/env.py: 43%

1import os 

2from contextlib import ExitStack, asynccontextmanager, contextmanager 

3 

4from anyio.from_thread import start_blocking_portal 

5 

6from jumpstarter.client import client_from_path 

7from jumpstarter.config.client import ClientConfigV1Alpha1Drivers 

8from jumpstarter.config.env import JUMPSTARTER_HOST 

9 

10 

11@asynccontextmanager 

12async def env_async(portal, stack): 

13 """Provide a client for an existing JUMPSTARTER_HOST environment variable. 

14 

15 Async version of env() 

16 

17 This is useful when interacting with an already established Jumpstarter shell, 

18 to either a local exporter or a remote one. 

19 """ 

20 host = os.environ.get(JUMPSTARTER_HOST, None) 

21 if host is None: 

22 raise RuntimeError(f"{JUMPSTARTER_HOST} not set") 

23 

24 drivers = ClientConfigV1Alpha1Drivers() 

25 

26 async with client_from_path( 

27 host, 

28 portal, 

29 stack, 

30 allow=drivers.allow, 

31 unsafe=drivers.unsafe, 

32 ) as client: 

33 try: 

34 yield client 

35 finally: 

36 if hasattr(client, "close"): 

37 client.close() 

38 

39 

40@contextmanager 

41def env(): 

42 """Provide a client for an existing JUMPSTARTER_HOST environment variable. 

43 

44 This is useful when interacting with an already established Jumpstarter shell, 

45 to either a local exporter or a remote one. 

46 """ 

47 with start_blocking_portal() as portal: 

48 with ExitStack() as stack: 

49 with portal.wrap_async_context_manager(env_async(portal, stack)) as client: 

50 yield client