import sqlite3 from typing import IO import click from flask import Flask, current_app, g def get_database(): if "db" not in g: g.db = sqlite3.connect( current_app.config["DATABASE"], detect_types=sqlite3.PARSE_DECLTYPES ) g.db.row_factory = sqlite3.Row return g.db def initialize_database(): db = get_database() with current_app.open_resource("schema.sql") as f: f: IO[bytes] db.executescript(f.read().decode()) def close_database(e=None): db = g.pop("db", None) if db is not None: db.close() @click.command("init-db") def init_db_command(): """Wipe the existing database and create new tables.""" if input("Are you sure you wish to overwrite any existing database? (y/n) ") == "y": initialize_database() click.echo("Initialized the database.") else: click.echo("Aborted.") def initialize_app(app: Flask): app.teardown_appcontext(close_database) app.cli.add_command(init_db_command)