import argparse
import logging
import logging.config
import warnings
from sqllineage import (
DEFAULT_DIALECT,
DEFAULT_HOST,
DEFAULT_LOGGING,
DEFAULT_PORT,
NAME as MAIN_NAME,
VERSION as MAIN_VERSION,
)
from sqllineage.core.metadata.dummy import DummyMetaDataProvider
from sqllineage.core.metadata.sqlalchemy import SQLAlchemyMetaDataProvider
from sqllineage.drawing import draw_lineage_graph
from sqllineage.runner import LineageRunner
from sqllineage.utils.constant import LineageLevel
from sqllineage.utils.helpers import extract_sql_from_args
logger = logging.getLogger(__name__)
[docs]
def main(args=None) -> None:
"""
The command line interface entry point.
:param args: the command line arguments for sqllineage command
"""
logging.config.dictConfig(DEFAULT_LOGGING)
parser = argparse.ArgumentParser(
prog="sqllineage", description="SQL Lineage Parser."
)
parser.add_argument(
"--version", action="version", version="%s %s" % (MAIN_NAME, MAIN_VERSION)
)
parser.add_argument(
"-e", metavar="<quoted-query-string>", help="SQL from command line"
)
parser.add_argument("-f", metavar="<filename>", help="SQL from files")
parser.add_argument(
"-v",
"--verbose",
help="increase output verbosity, show statement level lineage result",
action="store_true",
)
parser.add_argument(
"-l",
"--level",
help="lineage level, column or table, default at table level",
choices=[LineageLevel.TABLE, LineageLevel.COLUMN],
default=LineageLevel.TABLE,
)
parser.add_argument(
"-g",
"--graph-visualization",
help="show graph visualization of the lineage in a webserver",
action="store_true",
)
parser.add_argument(
"-H",
"--host",
help="the host visualization webserver will be bind to",
type=str,
default=DEFAULT_HOST,
metavar="<hostname>",
)
parser.add_argument(
"-p",
"--port",
help="the port visualization webserver will be listening on",
type=int,
default=DEFAULT_PORT,
metavar="<port_number>{0..65536}",
)
parser.add_argument(
"-d",
"--dialect",
help="the dialect used to analyze the lineage, use --dialects to show all available dialects",
type=str,
default=DEFAULT_DIALECT,
metavar="<dialect>",
)
parser.add_argument(
"-ds",
"--dialects",
help="list all the available dialects",
action="store_true",
)
parser.add_argument(
"--silent_mode",
help="skip unsupported statements",
action="store_true",
)
parser.add_argument(
"--sqlalchemy_url",
help="sqlalchemy url to provide metadata for lineage analysis",
type=str,
)
args = parser.parse_args(args)
metadata_provider = (
SQLAlchemyMetaDataProvider(args.sqlalchemy_url)
if args.sqlalchemy_url
else DummyMetaDataProvider()
)
if args.e and args.f:
warnings.warn("Both -e and -f options are specified. -e option will be ignored")
if args.f or args.e:
sql = extract_sql_from_args(args)
runner = LineageRunner(
sql,
dialect=args.dialect,
metadata_provider=metadata_provider,
verbose=args.verbose,
draw_options={
"host": args.host,
"port": args.port,
"f": args.f if args.f else None,
},
silent_mode=args.silent_mode,
)
if args.graph_visualization:
runner.draw()
elif args.level == LineageLevel.COLUMN:
runner.print_column_lineage()
else:
runner.print_table_lineage()
elif args.graph_visualization:
return draw_lineage_graph(
**{
"host": args.host,
"port": args.port,
"metadata_provider": metadata_provider,
}
)
elif args.dialects:
dialects = []
for _, supported_dialects in LineageRunner.supported_dialects().items():
dialects += supported_dialects
print("\n".join(dialects))
else:
parser.print_help()
if __name__ == "__main__":
main()