-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathcli.py
More file actions
186 lines (155 loc) · 4.99 KB
/
Copy pathcli.py
File metadata and controls
186 lines (155 loc) · 4.99 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
"""CLI interface for json2sql using Typer."""
import os
import sys
from pathlib import Path
import typer
# Lazy imports — converter/dialects pulled on command execution
# to reduce cold start from ~340ms to ~160ms.
try:
from revenueholdings_license import require_license
except ImportError:
import warnings
warnings.warn(
"revenueholdings-license not installed; license checks skipped", stacklevel=2
)
def require_license(product: str) -> None: # type: ignore[misc]
pass
_require_license_strict: bool = False
app = typer.Typer(
name="json2sql",
help="Convert JSON files/datasets to SQL INSERT statements.",
no_args_is_help=True,
)
@app.callback()
def _app_callback(
require_license_flag: bool = typer.Option(
False,
"--require-license",
help=(
"Exit with an error if revenueholdings-license is not installed "
"or if the license check fails. "
"Also enabled via REVENUEHOLDINGS_REQUIRE_LICENSE=1."
),
),
) -> None:
"""Convert JSON files/datasets to SQL INSERT statements."""
global _require_license_strict
_require_license_strict = require_license_flag or bool(
os.environ.get("REVENUEHOLDINGS_REQUIRE_LICENSE")
)
def _check_license(tool_name: str) -> None:
"""Check revenueholdings license; raise on failure if strict mode is enabled."""
if os.environ.get("REVENUEHOLDINGS_LICENSE_BYPASS"):
return
try:
from revenueholdings_license import require_license
require_license(tool_name)
except ImportError:
if _require_license_strict:
typer.echo(
"Error: revenueholdings-license is not installed. "
"Install it with: pip install revenueholdings-license",
err=True,
)
raise typer.Exit(code=1) from None
except Exception:
if _require_license_strict:
raise
@app.command()
def convert(
input_file: Path | None = typer.Argument( # noqa: B008
None,
help="Path to JSON file. Reads from stdin if not provided.",
exists=True,
),
dialect: str = typer.Option( # noqa: B008
"postgres",
"--dialect",
"-d",
help="SQL dialect: postgres, mysql, sqlite",
),
table: str = typer.Option(
"data",
"--table",
"-t",
help="Table name for INSERT statements.",
),
output: Path | None = typer.Option( # noqa: B008
None,
"--output",
"-o",
help="Output SQL file. Prints to stdout if not provided.",
),
flatten: bool = typer.Option(
False,
"--flatten",
"-f",
help="Flatten nested JSON into relational tables.",
),
schema_only: bool = typer.Option(
False,
"--schema-only",
help="Generate CREATE TABLE statements only (no INSERT).",
),
):
"""Convert a JSON file to SQL INSERT statements."""
_check_license("json2sql")
# Lazy imports — cold-start optimization (~180ms savings)
from .converter import JSONToSQLConverter
from .dialects import Dialect
# Validate dialect
try:
dialect_enum = Dialect(dialect)
except ValueError:
valid = ", ".join(d.value for d in Dialect)
typer.echo(
f"Error: Unknown dialect '{dialect}'. Choose from: {valid}", err=True
)
raise typer.Exit(code=1) from None
# Read input
if input_file:
json_text = input_file.read_text(encoding="utf-8")
elif not sys.stdin.isatty():
json_text = sys.stdin.read()
else:
typer.echo("Error: Provide a JSON file or pipe JSON to stdin.", err=True)
raise typer.Exit(code=1)
converter = JSONToSQLConverter(dialect=dialect_enum, flatten=flatten)
try:
if schema_only:
result = converter.generate_schema(json_text, table_name=table)
else:
result = converter.convert(json_text, table_name=table)
except Exception as e:
typer.echo(f"Error converting JSON: {e}", err=True)
raise typer.Exit(code=1) from e
# Write output
if output:
output.write_text(result, encoding="utf-8")
typer.echo(f"SQL written to {output}", err=True)
else:
typer.echo(result)
@app.command()
def mcp() -> None:
"""Run as an MCP (Model Context Protocol) server over stdio.
AI coding agents (Claude Code, Cursor, etc.) use this to interact
with json2sql tools directly.
"""
_check_license("json2sql")
try:
from click_to_mcp import run # type: ignore[import-untyped]
except ImportError:
typer.echo(
"Error: click_to_mcp is required for MCP mode. "
"Install it with: pip install click-to-mcp",
err=True,
)
raise typer.Exit(code=1) from None
run(app)
@app.command()
def version() -> None:
"""Show version."""
from . import __version__
typer.echo(f"json2sql {__version__}")
if __name__ == "__main__":
app()