1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
|
import sqlite3
import click
from flask import current_app, g
from flask.cli import with_appcontext
def get_db():
if 'db' not in g:
g.db = sqlite3.connect(
current_app.config['DATABASE'],
detect_types=sqlite3.PARSE_DECLTYPES
)
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA foreighn_keys = ON;")
return g.db
def close_db(e=None):
db = g.pop('db', None)
if db is not None:
db.close()
migrations = [
"""
CREATE TABLE users (
username TEXT PRIMARY KEY,
password TEXT NOT NULL
);""",
"""
CREATE TABLE threads (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
creator TEXT NOT NULL REFERENCES users(username),
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TABLE posts (
id INTEGER PRIMARY KEY,
content TEXT,
thread INTEGER NOT NULL REFERENCES threads(id),
author TEXT NOT NULL REFERENCES users(username),
created TIMESTAMP NOT NULL
);
CREATE INDEX posts_thread_idx ON posts (thread);
""",
]
def init_db():
db = get_db()
version = db.execute("PRAGMA user_version;").fetchone()[0]
for i in range(version, len(migrations)):
db.executescript(migrations[i])
db.execute(f"PRAGMA user_version = {i+1}")
db.commit()
click.echo(f"migration {i}")
@click.command("migrate")
@with_appcontext
def migrate_command():
"""update database scheme etc"""
init_db()
click.echo("ok")
def init_app(app):
app.teardown_appcontext(close_db)
app.cli.add_command(migrate_command)
|
