Coverage for src/prisma/generator/jsonrpc.py: 93%

1from __future__ import annotations 

2 

3import sys 

4import json 

5import logging 

6from typing import Any, Dict, List, Type, Union, Optional 

7from pathlib import Path 

8from typing_extensions import Literal, TypedDict 

9 

10from pydantic import Field 

11 

12from .models import BaseModel 

13from .._compat import model_json 

14 

15log: logging.Logger = logging.getLogger(__name__) 

16 

17 

18__all__ = ('Manifest',) 

19 

20 

21class Request(BaseModel): 

22 # JSON RPC protocol version 

23 jsonrpc: str = '2.0' 

24 

25 # identifies a request 

26 id: int 

27 

28 # request intention 

29 method: str 

30 

31 # request payload 

32 params: Optional[Dict[str, Any]] = None 

33 

34 

35class SuccessResponse(BaseModel): 

36 id: int 

37 jsonrpc: str = '2.0' 

38 result: Optional[Dict[str, Any]] = None 

39 

40 

41class ErrorData(TypedDict): 

42 code: int 

43 message: str 

44 data: object 

45 

46 

47class ErrorResponse(BaseModel): 

48 id: int 

49 error: ErrorData 

50 jsonrpc: str = '2.0' 

51 

52 

53Response = Union[SuccessResponse, ErrorResponse] 

54 

55EngineType = Literal[ 

56 'prismaFmt', 

57 'queryEngine', 

58 'libqueryEngine', 

59 'migrationEngine', 

60 'introspectionEngine', 

61] 

62 

63 

64class Manifest(BaseModel): 

65 """Generator metadata""" 

66 

67 prettyName: str = Field(alias='name') 

68 defaultOutput: Union[str, Path] = Field(alias='default_output') 

69 denylist: Optional[List[str]] = None 

70 requiresEngines: Optional[List[EngineType]] = Field(alias='requires_engines', default=None) 

71 requiresGenerators: Optional[List[str]] = Field(alias='requires_generators', default=None) 

72 

73 

74# TODO: proper types 

75method_mapping: Dict[str, Type[Request]] = { 

76 'getManifest': Request, 

77 'generate': Request, 

78} 

79 

80 

81def readline() -> Optional[str]: 

82 try: 

83 line = input() 

84 except EOFError: 

85 log.debug('Ignoring EOFError') 

86 return None 

87 

88 return line 

89 

90 

91def parse(line: str) -> Request: 

92 data = json.loads(line) 

93 try: 

94 method = data['method'] 

95 except (KeyError, TypeError): 

96 # TODO 

97 raise 

98 else: 

99 request_type = method_mapping.get(method) 

100 if request_type is None: 100 ↛ 101line 100 didn't jump to line 101, because the condition on line 100 was never true

101 raise RuntimeError(f'Unknown method: {method}') 

102 

103 return request_type(**data) 

104 

105 

106def reply(response: Response) -> None: 

107 dumped = model_json(response) + '\n' 

108 print(dumped, file=sys.stderr, flush=True) # noqa: T201 

109 log.debug('Replied with %s', dumped)