Coverage for cli / commands / files_cmd.py: 98%

1"""File system commands.""" 

2 

3import sys 

4from typing import Any 

5 

6import click 

7 

8from ..config import GCOConfig 

9from ..files import get_file_system_client 

10from ..output import format_file_system_table, get_output_formatter 

11 

12pass_config = click.make_pass_decorator(GCOConfig, ensure=True) 

13 

14 

15@click.group() 

16@pass_config 

17def files(config: Any) -> None: 

18 """Manage file systems (EFS/FSx).""" 

19 pass 

20 

21 

22@files.command("list") 

23@click.option("--region", "-r", help="Filter by region") 

24@pass_config 

25def list_file_systems(config: Any, region: Any) -> None: 

26 """List file systems across GCO stacks.""" 

27 formatter = get_output_formatter(config) 

28 fs_client = get_file_system_client(config) 

29 

30 try: 

31 file_systems = fs_client.get_file_systems(region) 

32 

33 if not file_systems: 

34 formatter.print_warning("No file systems found") 

35 return 

36 

37 if config.output_format == "table": 37 ↛ 40line 37 didn't jump to line 40 because the condition on line 37 was always true

38 print(format_file_system_table(file_systems)) 

39 else: 

40 formatter.print(file_systems) 

41 

42 except Exception as e: 

43 formatter.print_error(f"Failed to list file systems: {e}") 

44 sys.exit(1) 

45 

46 

47@files.command("get") 

48@click.argument("region") 

49@click.option( 

50 "--type", 

51 "-t", 

52 "fs_type", 

53 type=click.Choice(["efs", "fsx"]), 

54 default="efs", 

55 help="File system type", 

56) 

57@pass_config 

58def get_file_system(config: Any, region: Any, fs_type: Any) -> None: 

59 """Get file system details for a region.""" 

60 formatter = get_output_formatter(config) 

61 fs_client = get_file_system_client(config) 

62 

63 try: 

64 fs = fs_client.get_file_system_by_region(region, fs_type) 

65 if fs: 

66 formatter.print(fs) 

67 else: 

68 formatter.print_error(f"No {fs_type.upper()} file system found in {region}") 

69 sys.exit(1) 

70 except Exception as e: 

71 formatter.print_error(f"Failed to get file system: {e}") 

72 sys.exit(1) 

73 

74 

75@files.command("access-points") 

76@click.argument("file_system_id") 

77@click.option("--region", "-r", required=True, help="AWS region") 

78@pass_config 

79def list_access_points(config: Any, file_system_id: Any, region: Any) -> None: 

80 """List EFS access points for a file system.""" 

81 formatter = get_output_formatter(config) 

82 fs_client = get_file_system_client(config) 

83 

84 try: 

85 access_points = fs_client.get_access_point_info(file_system_id, region) 

86 

87 if not access_points: 

88 formatter.print_warning("No access points found") 

89 return 

90 

91 formatter.print(access_points) 

92 

93 except Exception as e: 

94 formatter.print_error(f"Failed to list access points: {e}") 

95 sys.exit(1) 

96 

97 

98@files.command("ls") 

99@click.argument("remote_path", default="/") 

100@click.option("--region", "-r", required=True, help="AWS region") 

101@click.option("--namespace", "-n", default="gco-jobs", help="Kubernetes namespace") 

102@click.option( 

103 "--storage-type", 

104 "-t", 

105 type=click.Choice(["efs", "fsx"]), 

106 default="efs", 

107 help="Storage type (default: efs)", 

108) 

109@click.option("--pvc", help="PVC name (default: gco-shared-storage for EFS)") 

110@pass_config 

111def list_storage_contents( 

112 config: Any, remote_path: Any, region: Any, namespace: Any, storage_type: Any, pvc: Any 

113) -> None: 

114 """List contents of EFS/FSx storage. 

115 

116 Creates a temporary helper pod to mount the storage and list contents, 

117 then cleans up automatically. Useful for discovering job output directories. 

118 

119 REMOTE_PATH is relative to the storage root (default: / for root listing) 

120 

121 REQUIREMENTS: 

122 - kubectl installed and in PATH 

123 - EKS access entry configured for your IAM principal 

124 - AWS credentials with eks:DescribeCluster permission 

125 

126 Examples: 

127 gco files ls -r us-east-1 # List root of EFS 

128 gco files ls efs-output-example -r us-east-1 # List job output directory 

129 gco files ls -r us-west-2 -t fsx # List FSx root 

130 """ 

131 formatter = get_output_formatter(config) 

132 fs_client = get_file_system_client(config) 

133 

134 try: 

135 formatter.print_info(f"Listing {remote_path} on {storage_type.upper()}...") 

136 

137 result = fs_client.list_storage_contents( 

138 region=region, 

139 remote_path=remote_path, 

140 storage_type=storage_type, 

141 namespace=namespace, 

142 pvc_name=pvc, 

143 ) 

144 

145 if result["status"] == "success": 

146 formatter.print_success(result["message"]) 

147 if result["contents"]: 

148 # Format as table 

149 print("\n TYPE SIZE NAME") 

150 print(" " + "-" * 40) 

151 for item in result["contents"]: 

152 item_type = "DIR " if item["is_directory"] else "FILE" 

153 size = f"{item['size_bytes']:>10}" if not item["is_directory"] else " -" 

154 print(f" {item_type} {size} {item['name']}") 

155 else: 

156 formatter.print_warning("Directory is empty") 

157 else: 

158 formatter.print_error(result["message"]) 

159 sys.exit(1) 

160 

161 except Exception as e: 

162 formatter.print_error(f"Failed to list storage contents: {e}") 

163 sys.exit(1) 

164 

165 

166@files.command("download") 

167@click.argument("remote_path") 

168@click.argument("local_path") 

169@click.option("--region", "-r", required=True, help="AWS region") 

170@click.option("--namespace", "-n", default="gco-jobs", help="Kubernetes namespace") 

171@click.option( 

172 "--storage-type", 

173 "-t", 

174 type=click.Choice(["efs", "fsx"]), 

175 default="efs", 

176 help="Storage type (default: efs)", 

177) 

178@click.option("--pvc", help="PVC name (default: gco-shared-storage for EFS)") 

179@pass_config 

180def download_files( 

181 config: Any, 

182 remote_path: Any, 

183 local_path: Any, 

184 region: Any, 

185 namespace: Any, 

186 storage_type: Any, 

187 pvc: Any, 

188) -> None: 

189 """Download files from EFS/FSx storage. 

190 

191 Creates a temporary helper pod to mount the storage and copy files, 

192 then cleans up automatically. Works even after job pods are deleted. 

193 

194 REMOTE_PATH is relative to the storage root (e.g., efs-output-example/results.json) 

195 

196 REQUIREMENTS: 

197 - kubectl installed and in PATH 

198 - EKS access entry configured for your IAM principal 

199 - AWS credentials with eks:DescribeCluster permission 

200 

201 Examples: 

202 gco files download efs-output-example/results.json ./results.json -r us-east-1 

203 gco files download my-job/outputs ./outputs -r us-east-1 

204 gco files download checkpoints ./checkpoints -r us-west-2 -t fsx 

205 """ 

206 formatter = get_output_formatter(config) 

207 fs_client = get_file_system_client(config) 

208 

209 try: 

210 formatter.print_info( 

211 f"Downloading {remote_path} from {storage_type.upper()} to {local_path}..." 

212 ) 

213 

214 result = fs_client.download_from_storage( 

215 region=region, 

216 remote_path=remote_path, 

217 local_path=local_path, 

218 storage_type=storage_type, 

219 namespace=namespace, 

220 pvc_name=pvc, 

221 ) 

222 

223 formatter.print_success(f"Downloaded {result['size_bytes']} bytes to {local_path}") 

224 formatter.print(result) 

225 

226 except Exception as e: 

227 formatter.print_error(f"Failed to download files: {e}") 

228 sys.exit(1)