Coverage for src/nats_contrib/connect_opts/connect.py: 100%

1from __future__ import annotations 

2 

3from typing import TypeVar, overload 

4 

5from nats.aio.client import Client 

6 

7from .connect_opts import ConnectOption, ConnectOpts 

8 

9CT = TypeVar("CT", bound=Client) 

10 

11 

12@overload 

13async def connect( 

14 *opt: ConnectOption, 

15 client: None = None, 

16) -> Client: ... 

17 

18 

19@overload 

20async def connect( 

21 *opt: ConnectOption, 

22 client: CT, 

23) -> CT: ... 

24 

25 

26async def connect( 

27 *opt: ConnectOption, 

28 client: Client | None = None, 

29) -> Client: 

30 """Connect to a NATS server using the provided options. 

31 

32 Args: 

33 *opt: Options to use when connecting to the NATS server. 

34 

35 Returns: 

36 Client: The connected client. 

37 

38 Example: 

39 

40 ```python 

41 from nats.aio.connect import connect, WithServers, WithUserPassword 

42 

43 client = await connect( 

44 WithServer("nats://localhost:4222""), 

45 ) 

46 ``` 

47 """ 

48 opts = ConnectOpts() 

49 for o in opt: 

50 o(opts) 

51 nc = client or Client() 

52 await nc.connect(**opts.to_dict()) 

53 return nc