Coverage for src/prisma/_registry.py: 79%

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING, Union, Callable 

4 

5from .errors import ClientNotRegisteredError, ClientAlreadyRegisteredError 

6 

7if TYPE_CHECKING: 

8 from .client import Prisma # noqa: TID251 

9 

10 

11RegisteredClient = Union['Prisma', Callable[[], 'Prisma']] 

12_registered_client: RegisteredClient | None = None 

13 

14 

15def register(client: RegisteredClient) -> None: 

16 """Register a client instance to be retrieved by `get_client()` 

17 

18 This function _must_ only be called once, preferrably as soon as possible 

19 to avoid any potentially confusing errors with threads or processes. 

20 """ 

21 from .client import Prisma # noqa: TID251 

22 

23 global _registered_client 

24 

25 if _registered_client is not None: 

26 raise ClientAlreadyRegisteredError() 

27 

28 if not isinstance(client, Prisma) and not callable(client): 28 ↛ 29line 28 didn't jump to line 29, because the condition on line 28 was never true

29 raise TypeError( 

30 f'Expected either a {Prisma} instance or a function that returns a {Prisma} but got {client} instead.' 

31 ) 

32 

33 _registered_client = client 

34 

35 

36def get_client() -> Prisma: 

37 """Get the registered client instance 

38 

39 Raises errors.ClientNotRegisteredError() if no client instance has been registered. 

40 """ 

41 from .client import Prisma # noqa: TID251 

42 

43 registered = _registered_client 

44 if registered is None: 

45 raise ClientNotRegisteredError() from None 

46 

47 if isinstance(registered, Prisma): 

48 return registered 

49 

50 client = registered() 

51 if not isinstance(client, Prisma): # pyright: ignore[reportUnnecessaryIsInstance] 

52 raise TypeError(f'Registered function returned {client} instead of a {Prisma} instance.') 

53 

54 return client