Coverage for src / mysingle / cli / protos / commands / generate.py: 0%

1""" 

2Generate 명령 - Buf를 사용하여 Python gRPC 스텁 생성. 

3""" 

4 

5from __future__ import annotations 

6 

7import argparse 

8import re 

9import subprocess 

10from pathlib import Path 

11 

12from ...utils import ask_confirm, print_success 

13from ..models import ProtoConfig 

14from ..utils import Color, LogLevel, colorize, log, log_header 

15 

16 

17def ensure_file_exists(path: Path, description: str) -> None: 

18 """필수 파일 존재 확인""" 

19 if not path.exists(): 

20 raise SystemExit(f"필수 파일 누락: {description} ({path})") 

21 

22 

23def buf_generate(config: ProtoConfig) -> None: 

24 """Buf를 사용하여 코드 생성""" 

25 ensure_file_exists(config.buf_template, "buf.gen.yaml 템플릿") 

26 

27 log("Buf를 사용하여 코드 생성 중...", LogLevel.STEP) 

28 

29 try: 

30 # repo_root에서 실행하고 protos/buf.gen.yaml을 템플릿으로 사용 

31 subprocess.run( 

32 [ 

33 "buf", 

34 "generate", 

35 "protos", 

36 "--template", 

37 "protos/buf.gen.yaml", 

38 ], 

39 cwd=config.repo_root, 

40 check=True, 

41 ) 

42 log("코드 생성 완료", LogLevel.SUCCESS) 

43 except subprocess.CalledProcessError as e: 

44 log(f"코드 생성 실패: {e}", LogLevel.ERROR) 

45 raise SystemExit(1) from e 

46 except FileNotFoundError: 

47 log("Buf가 설치되어 있지 않습니다.", LogLevel.ERROR) 

48 log("설치 방법: https://buf.build/docs/installation", LogLevel.INFO) 

49 raise SystemExit(1) 

50 

51 

52def rewrite_generated_imports( 

53 generated_dir: Path, package_name: str = "mysingle" 

54) -> list[Path]: 

55 """생성된 파일의 import 경로 수정""" 

56 if not generated_dir.exists(): 

57 return [] 

58 

59 log("생성된 파일의 import 경로 수정 중...", LogLevel.STEP) 

60 

61 # .py 파일과 .pyi 타입 스텁 파일 모두 처리 

62 patterns = ("*_pb2.py", "*_pb2_grpc.py", "*_pb2.pyi", "*_pb2_grpc.pyi") 

63 replacements = [ 

64 # Pattern 1: from protos.xxx -> from mysingle.protos.xxx 

65 (re.compile(r"from protos\."), f"from {package_name}.protos."), 

66 # Pattern 2: import protos.xxx -> import mysingle.protos.xxx 

67 (re.compile(r"import protos\."), f"import {package_name}.protos."), 

68 # Pattern 3: from common import xxx -> from mysingle.protos.common import xxx 

69 ( 

70 re.compile(r"^from common import ", re.MULTILINE), 

71 f"from {package_name}.protos.common import ", 

72 ), 

73 # Pattern 4: from services.xxx import -> from mysingle.protos.services.xxx import 

74 ( 

75 re.compile(r"^from services\.", re.MULTILINE), 

76 f"from {package_name}.protos.services.", 

77 ), 

78 ] 

79 

80 modified: list[Path] = [] 

81 

82 for pattern in patterns: 

83 for file_path in generated_dir.rglob(pattern): 

84 original = file_path.read_text(encoding="utf-8") 

85 updated = original 

86 

87 for regex, repl in replacements: 

88 updated = regex.sub(repl, updated) 

89 

90 if updated != original: 

91 file_path.write_text(updated, encoding="utf-8") 

92 modified.append(file_path) 

93 log( 

94 f"수정: {colorize(str(file_path.relative_to(generated_dir)), Color.CYAN)}", 

95 LogLevel.DEBUG, 

96 ) 

97 

98 if modified: 

99 log( 

100 f"총 {colorize(str(len(modified)), Color.GREEN, bold=True)}개 파일 import 수정 완료", 

101 LogLevel.SUCCESS, 

102 ) 

103 else: 

104 log("import 수정이 필요한 파일 없음", LogLevel.INFO) 

105 

106 return modified 

107 

108 

109def ensure_init_files(generated_dir: Path) -> list[Path]: 

110 """생성된 디렉토리에 __init__.py 파일 생성""" 

111 if not generated_dir.exists(): 

112 return [] 

113 

114 log("__init__.py 파일 생성 중...", LogLevel.STEP) 

115 

116 created: list[Path] = [] 

117 

118 # protos 디렉토리의 모든 하위 디렉토리에 __init__.py 생성 

119 for dirpath in [generated_dir] + list(generated_dir.rglob("*/")): 

120 if dirpath.is_dir(): 

121 init_file = dirpath / "__init__.py" 

122 if not init_file.exists(): 

123 init_file.touch() 

124 created.append(init_file) 

125 log( 

126 f"생성: {colorize(str(init_file.relative_to(generated_dir.parent)), Color.CYAN)}", 

127 LogLevel.DEBUG, 

128 ) 

129 

130 if created: 

131 log( 

132 f"총 {colorize(str(len(created)), Color.GREEN, bold=True)}개 __init__.py 파일 생성 완료", 

133 LogLevel.SUCCESS, 

134 ) 

135 else: 

136 log("__init__.py 파일 생성 불필요", LogLevel.INFO) 

137 

138 return created 

139 

140 

141def execute(args: argparse.Namespace, config: ProtoConfig) -> int: 

142 """Generate 명령 실행""" 

143 log_header("Proto 코드 생성") 

144 

145 # 1. Buf 코드 생성 

146 buf_generate(config) 

147 

148 # 2. Import 경로 수정 

149 if not args.skip_rewrite: 

150 # generated_root는 이미 src/mysingle/protos를 가리킴 

151 rewrite_generated_imports(config.generated_root, "mysingle") 

152 

153 # 3. __init__.py 파일 생성 

154 if not args.skip_init: 

155 ensure_init_files(config.generated_root) 

156 

157 log("\n✅ 모든 작업 완료!", LogLevel.SUCCESS) 

158 

159 return 0 

160 

161 

162def execute_interactive(config: ProtoConfig) -> int: 

163 """대화형 모드로 generate 명령 실행""" 

164 log_header("Python gRPC 스텁 생성") 

165 

166 print_success("Proto 파일로부터 Python 코드를 생성합니다.") 

167 

168 if not ask_confirm("계속하시겠습니까?", default=True): 

169 log("취소되었습니다.", LogLevel.INFO) 

170 return 0 

171 

172 # 기본값으로 실행 

173 args = argparse.Namespace(skip_rewrite=False, skip_init=False) 

174 return execute(args, config) 

175 

176 

177def setup_parser(parser: argparse.ArgumentParser) -> None: 

178 """Generate 명령 파서 설정""" 

179 parser.add_argument( 

180 "--skip-rewrite", 

181 action="store_true", 

182 help="import 경로 수정 단계 건너뛰기", 

183 ) 

184 parser.add_argument( 

185 "--skip-init", 

186 action="store_true", 

187 help="__init__.py 파일 생성 단계 건너뛰기", 

188 )