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

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 

20 

21@files.command("list") 

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

23@pass_config 

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

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

26 formatter = get_output_formatter(config) 

27 fs_client = get_file_system_client(config) 

28 

29 try: 

30 file_systems = fs_client.get_file_systems(region) 

31 

32 if not file_systems: 

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

34 return 

35 

36 if config.output_format == "table": 

37 print(format_file_system_table(file_systems)) 

38 else: 

39 formatter.print(file_systems) 

40 

41 except Exception as e: 

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

43 sys.exit(1) 

44 

45 

46@files.command("get") 

47@click.argument("region") 

48@click.option( 

49 "--type", 

50 "-t", 

51 "fs_type", 

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

53 default="efs", 

54 help="File system type", 

55) 

56@pass_config 

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

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

59 formatter = get_output_formatter(config) 

60 fs_client = get_file_system_client(config) 

61 

62 try: 

63 fs = fs_client.get_file_system_by_region(region, fs_type) 

64 if fs: 

65 formatter.print(fs) 

66 else: 

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

68 sys.exit(1) 

69 except Exception as e: 

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

71 sys.exit(1) 

72 

73 

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

75@click.argument("file_system_id") 

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

77@pass_config 

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

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

80 formatter = get_output_formatter(config) 

81 fs_client = get_file_system_client(config) 

82 

83 try: 

84 access_points = fs_client.get_access_point_info(file_system_id, region) 

85 

86 if not access_points: 

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

88 return 

89 

90 formatter.print(access_points) 

91 

92 except Exception as e: 

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

94 sys.exit(1) 

95 

96 

97@files.command("ls") 

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

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

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

101@click.option( 

102 "--storage-type", 

103 "-t", 

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

105 default="efs", 

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

107) 

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

109@pass_config 

110def list_storage_contents( 

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

112) -> None: 

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

114 

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

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

117 

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

119 

120 REQUIREMENTS: 

121 - kubectl installed and in PATH 

122 - EKS access entry configured for your IAM principal 

123 - AWS credentials with eks:DescribeCluster permission 

124 

125 Examples: 

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

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

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

129 """ 

130 formatter = get_output_formatter(config) 

131 fs_client = get_file_system_client(config) 

132 

133 try: 

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

135 

136 result = fs_client.list_storage_contents( 

137 region=region, 

138 remote_path=remote_path, 

139 storage_type=storage_type, 

140 namespace=namespace, 

141 pvc_name=pvc, 

142 ) 

143 

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

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

146 if result["contents"]: 

147 # Format as table 

148 print("\n TYPE SIZE NAME") 

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

150 for item in result["contents"]: 

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

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

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

154 else: 

155 formatter.print_warning("Directory is empty") 

156 else: 

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

158 sys.exit(1) 

159 

160 except Exception as e: 

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

162 sys.exit(1) 

163 

164 

165@files.command("download") 

166@click.argument("remote_path") 

167@click.argument("local_path") 

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

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

170@click.option( 

171 "--storage-type", 

172 "-t", 

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

174 default="efs", 

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

176) 

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

178@pass_config 

179def download_files( 

180 config: Any, 

181 remote_path: Any, 

182 local_path: Any, 

183 region: Any, 

184 namespace: Any, 

185 storage_type: Any, 

186 pvc: Any, 

187) -> None: 

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

189 

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

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

192 

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

194 

195 REQUIREMENTS: 

196 - kubectl installed and in PATH 

197 - EKS access entry configured for your IAM principal 

198 - AWS credentials with eks:DescribeCluster permission 

199 

200 Examples: 

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

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

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

204 """ 

205 formatter = get_output_formatter(config) 

206 fs_client = get_file_system_client(config) 

207 

208 try: 

209 formatter.print_info( 

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

211 ) 

212 

213 result = fs_client.download_from_storage( 

214 region=region, 

215 remote_path=remote_path, 

216 local_path=local_path, 

217 storage_type=storage_type, 

218 namespace=namespace, 

219 pvc_name=pvc, 

220 ) 

221 

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

223 formatter.print(result) 

224 

225 except Exception as e: 

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

227 sys.exit(1)