Column Level Lineage with Python

[matsabella]

Hi! First of all, thanks for this really interesting and useful library.
I would like to extract in a python script the column level lineage from a SQL script and assign it directly to a variable, but in documentation it is shown only an example for extracting a table lineage.
How it is possible to extract the column level lineage in Python using the library?
Other question: I have seen that the library fails in evaluating sql statements that are particularly complex (with many CTE and subqueries), it can be expected an update soon for covering this kind of statements?
Thank you so much in advance

[reata]

For the column level lineage Python API, I didn't document it because it's not stable enough. But you're free to try and welcome to reporting bugs if you met any.

result = LineageRunner(sql)
column_lineage = result.get_column_lineage()
# this returns List[Tuple[Column, Column]]

CTE and subqueries are indeed among the biggest issues with column lineage. I hope I can cover most of the edge cases before long. The current develop progress is behind my schedule because of the covid-19 lockdown in Shanghai. I sure hope things get better soon and I can come back to this.

[matsabella]

Thank you so much for the clarification, wish you the best :)

[AnubhaJ27]

Hi. Thanks for introducing this library. It can help a great deal.
I am facing issue with two aspects - column level lineage for complex sql codes and visualization for sql codes.
I referred to the python code from GitHub itself which used arguments -e,-f,-l,-g,-H, etc.
Basically, nothing is working for me expect e and f.

There are no errors flashed in python but I simply don't get an output.
Am I missing anything from the code.
Kindly help with any related information/links.
Below is the script I used.

`
import argparse
import logging
import logging.config

from sqllineage import DEFAULT_HOST, DEFAULT_LOGGING, DEFAULT_PORT
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)

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(
    "-e", metavar="<quoted-query-string>", help="SQL from command line"  #<quoted-query-string>
)
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="localhost:8888",
)
parser.add_argument(
    "-p",
    "--port",
    help="the port visualization webserver will be listening on",
    type=int,
    default=DEFAULT_PORT,
    metavar="<port_number>{0..65536}",
)
args, unknown = parser.parse_known_args(args)
if args.e and args.f:
    logging.warning(
        "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,
        verbose=args.verbose,
        draw_options={
            "host": args.host,
            "port": args.port,
            "f": args.f if args.f else None,
        },
    )
    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})
else:
    parser.print_help()

if name == "main":
main()`

[reata]

If you want to call the functions in your Python script, you can refer to the document: https://sqllineage.readthedocs.io/en/latest/first_steps/beyond_command_line.html

I see your script is a copy of https://github.com/reata/sqllineage/blob/master/sqllineage/cli.py and this is just the command line interface. Calling it will give you back the command line help info.

$ python cli.py 
usage: sqllineage [-h] [-e <quoted-query-string>] [-f <filename>] [-v]
                  [-l {table,column}] [-g] [-H <hostname>]
                  [-p <port_number>{0..65536}]

SQL Lineage Parser.

optional arguments:
  -h, --help            show this help message and exit
  -e <quoted-query-string>
                        SQL from command line
  -f <filename>         SQL from files
  -v, --verbose         increase output verbosity, show statement level
                        lineage result
  -l {table,column}, --level {table,column}
                        lineage level, column or table, default at table level
  -g, --graph-visualization
                        show graph visualization of the lineage in a webserver
  -H <hostname>, --host <hostname>
                        the host visualization webserver will be bind to
  -p <port_number>{0..65536}, --port <port_number>{0..65536}
                        the port visualization webserver will be listening on