import os import re from datetime import datetime import click from flask.cli import with_appcontext from flask import Blueprint from psycopg2 import ProgrammingError from capsulflask.db import get_model, my_exec_info_message bp = Blueprint('cli', __name__) @bp.cli.command('sql') @click.option('-f', help='script filename') @click.option('-c', help='sql command') @with_appcontext def sql_script(f, c): """Run a sql script against the database. script is run 1 command at a time inside a single transaction.""" model = get_model() script = "" if f: filepath = os.path.join(os.getcwd(), f) if not os.path.isfile(filepath): raise f"{filepath} is not a file" with open(filepath, 'rb') as file: script = file.read().decode("utf8") elif c: script = c else: click.echo(f"you must provide sql to run either inline with the -c argument or in a file with the -f argument") return commands = re.split(";\\s+", script) for command in commands: if command.strip() != "": model.cursor.execute(command) if re.match("^\\s*select", command, re.IGNORECASE) is not None: for row in model.cursor.fetchall(): def format_value(x): if isinstance(x, bool): return "TRUE" if x else "FALSE" if not x : return "null" if isinstance(x, datetime): return x.isoformat() return f"{x}" click.echo(", ".join(list(map(format_value, row)))) else: click.echo(f"{model.cursor.rowcount} rows affected.") model.connection.commit() @bp.cli.command('cron-task') @with_appcontext def cron_task(): print('a')