Coverage for src/pyadql/__main__.py: 84%

1# PyADQL - pyadql turns an ADQL query into an AST 

2# Copyright (C) 2026 - Centre National d'Etudes Spatiales (Jean-Christophe Malapert for PDSSP) 

3# This file is part of PyADQL <https://gitlab.cnes.fr/pdssp/common/pyadql> 

4# SPDX-License-Identifier: Apache-2.0 

5"""Main program.""" 

6 

7import argparse 

8import json 

9import signal 

10import sys 

11import time 

12from dataclasses import fields, is_dataclass 

13from pprint import pformat 

14from typing import Any 

15 

16from loguru import logger 

17 

18from pyadql import __author__, __copyright__, __description__, __version__ 

19 

20from .config import configure_logging 

21from .parser import parse_adql, parse_tree 

22 

23logger.remove() 

24 

25 

26class SmartFormatter(argparse.HelpFormatter): 

27 """Smart formatter for argparse - The lines are split for long text""" 

28 

29 def _split_lines(self, text, width): 

30 if text.startswith("R|"): 

31 return text[2:].splitlines() 

32 # this is the RawTextHelpFormatter._split_lines 

33 return argparse.HelpFormatter._split_lines( # pylint: disable=protected-access 

34 self, text, width 

35 ) 

36 

37 

38class SigintHandler: # pylint: disable=too-few-public-methods 

39 """Handles the signal""" 

40 

41 def __init__(self): 

42 self.SIGINT = False # pylint: disable=invalid-name 

43 

44 def signal_handler(self, sig: int, frame): 

45 """Trap the signal 

46 

47 Args: 

48 sig (int): the signal number 

49 frame: the current stack frame 

50 """ 

51 # pylint: disable=unused-argument 

52 logger.error("You pressed Ctrl+C") 

53 self.SIGINT = True 

54 sys.exit(2) 

55 

56 

57def str2bool(string_to_test: str) -> bool: 

58 """Checks if a given string is a boolean 

59 

60 Args: 

61 string_to_test (str): string to test 

62 

63 Returns: 

64 bool: True when the string is a boolean otherwise False 

65 """ 

66 return string_to_test.lower() in ("yes", "true", "True", "t", "1") 

67 

68 

69def parse_cli() -> argparse.ArgumentParser: 

70 """Parse command line inputs. 

71 

72 Returns 

73 ------- 

74 argparse.ArgumentParser 

75 Command line options 

76 """ 

77 parser = argparse.ArgumentParser( 

78 description=__description__, 

79 formatter_class=SmartFormatter, 

80 epilog=__author__ + " - " + __copyright__, 

81 ) 

82 parser.add_argument( 

83 "-v", "--version", action="version", version="%(prog)s " + __version__ 

84 ) 

85 

86 parser.add_argument( 

87 "--level", 

88 choices=[ 

89 "INFO", 

90 "DEBUG", 

91 "WARNING", 

92 "ERROR", 

93 "CRITICAL", 

94 "TRACE", 

95 ], 

96 default="ERROR", 

97 help="set Level log (default: %(default)s)", 

98 ) 

99 

100 parser.add_argument( 

101 "query", 

102 nargs="?", 

103 help="ADQL query to parse. If omitted, read from --file or standard input.", 

104 ) 

105 parser.add_argument( 

106 "-f", 

107 "--file", 

108 metavar="PATH", 

109 help="Read the ADQL query from a file instead of the argument or stdin.", 

110 ) 

111 parser.add_argument( 

112 "--tree", 

113 action="store_true", 

114 help="Print the raw Lark parse tree instead of the typed AST.", 

115 ) 

116 parser.add_argument( 

117 "--json", 

118 action="store_true", 

119 help="Print the AST as JSON instead of a Python repr.", 

120 ) 

121 

122 parser.add_argument( 

123 "-q", 

124 "--quiet", 

125 action="store_true", 

126 help="Only print errors (overrides -v).", 

127 ) 

128 

129 return parser 

130 

131 

132def _read_query(args: argparse.Namespace) -> str: 

133 if args.query: 

134 logger.debug("Query supplied as an argument ({} characters)", len(args.query)) 

135 return args.query 

136 if args.file: 

137 logger.info("Reading the query from file {}", args.file) 

138 with open(args.file, encoding="utf-8") as f: 

139 return f.read() 

140 if not sys.stdin.isatty(): 

141 logger.info("Reading the query from standard input") 

142 return sys.stdin.read() 

143 logger.error("No query supplied (neither an argument, --file, nor stdin)") 

144 raise SystemExit(2) 

145 

146 

147def node_to_dict(obj: Any) -> Any: 

148 """Recursively convert an AST node (dataclass) into a JSON-compatible 

149 dict, keeping the class name under the "_type" key to disambiguate 

150 between the various expression types.""" 

151 if is_dataclass(obj) and not isinstance(obj, type): 

152 result = {"_type": type(obj).__name__} 

153 for f in fields(obj): 

154 result[f.name] = node_to_dict(getattr(obj, f.name)) 

155 return result 

156 if isinstance(obj, (list, tuple)): 

157 return [node_to_dict(item) for item in obj] 

158 return obj 

159 

160 

161def run(argv: list[str] | None = None) -> int: 

162 """Main function that instanciates the library.""" 

163 handler = SigintHandler() 

164 signal.signal(signal.SIGINT, handler.signal_handler) 

165 args = parse_cli().parse_args(argv) 

166 configure_logging(args.level) 

167 

168 t_start = time.perf_counter() 

169 

170 try: 

171 query = _read_query(args) 

172 except SystemExit as e: 

173 return e.code or 2 

174 

175 query = query.strip() 

176 if not query: 

177 logger.error("The query is empty") 

178 return 2 

179 

180 logger.debug("Query to parse ({} characters):\n{}", len(query), query) 

181 

182 try: 

183 if args.tree: 

184 logger.info("Parsing in raw Lark tree mode") 

185 tree = parse_tree(query) 

186 print(tree.pretty()) 

187 else: 

188 logger.info("Parsing into a typed AST") 

189 ast = parse_adql(query) 

190 logger.success( 

191 "Query parsed successfully (root type: {})", type(ast).__name__ 

192 ) 

193 if args.json: 

194 print(json.dumps(node_to_dict(ast), indent=2, ensure_ascii=False)) 

195 else: 

196 print(pformat(ast, width=100, compact=False)) 

197 except Exception as exc: 

198 logger.opt(exception=exc).error("ADQL parsing failed: {}", exc) 

199 return 1 

200 finally: 

201 logger.debug( 

202 "Command finished in {:.1f} ms", (time.perf_counter() - t_start) * 1000 

203 ) 

204 

205 return 0 

206 

207 

208if __name__ == "__main__": 

209 # execute only if run as a script 

210 raise SystemExit(run)