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

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 log.debug('Parsing JSONRPC request line %s', line) 

93 

94 data = json.loads(line) 

95 try: 

96 method = data['method'] 

97 except (KeyError, TypeError): 

98 # TODO 

99 raise 

100 else: 

101 request_type = method_mapping.get(method) 

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

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

104 

105 return request_type(**data) 

106 

107 

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

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

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

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