(line)
| 18 | |
| 19 | |
| 20 | def pgcli_line_magic(line): |
| 21 | _logger.debug("pgcli magic called: %r", line) |
| 22 | parsed = sql.parse.parse(line, {}) |
| 23 | # "get" was renamed to "set" in ipython-sql: |
| 24 | # https://github.com/catherinedevlin/ipython-sql/commit/f4283c65aaf68f961e84019e8b939e4a3c501d43 |
| 25 | if hasattr(sql.connection.Connection, "get"): |
| 26 | conn = sql.connection.Connection.get(parsed["connection"]) |
| 27 | else: |
| 28 | try: |
| 29 | conn = sql.connection.Connection.set(parsed["connection"]) |
| 30 | # a new positional argument was added to Connection.set in version 0.4.0 of ipython-sql |
| 31 | except TypeError: |
| 32 | conn = sql.connection.Connection.set(parsed["connection"], False) |
| 33 | |
| 34 | try: |
| 35 | # A corresponding pgcli object already exists |
| 36 | pgcli = conn._pgcli |
| 37 | _logger.debug("Reusing existing pgcli") |
| 38 | except AttributeError: |
| 39 | # I can't figure out how to get the underylying psycopg2 connection |
| 40 | # from the sqlalchemy connection, so just grab the url and make a |
| 41 | # new connection |
| 42 | pgcli = PGCli() |
| 43 | u = conn.url |
| 44 | _logger.debug("New pgcli: %r", str(u)) |
| 45 | |
| 46 | pgcli.connect_uri(str(u._replace(drivername="postgres"))) |
| 47 | conn._pgcli = pgcli |
| 48 | |
| 49 | # For convenience, print the connection alias |
| 50 | print(f"Connected: {conn.name}") |
| 51 | |
| 52 | try: |
| 53 | pgcli.run_cli() |
| 54 | except SystemExit: |
| 55 | pass |
| 56 | |
| 57 | if not pgcli.query_history: |
| 58 | return |
| 59 | |
| 60 | q = pgcli.query_history[-1] |
| 61 | |
| 62 | if not q.successful: |
| 63 | _logger.debug("Unsuccessful query - ignoring") |
| 64 | return |
| 65 | |
| 66 | if q.meta_changed or q.db_changed or q.path_changed: |
| 67 | _logger.debug("Dangerous query detected -- ignoring") |
| 68 | return |
| 69 | |
| 70 | ipython = get_ipython() |
| 71 | return ipython.run_cell_magic("sql", line, q.query) |
nothing calls this directly
no test coverage detected