88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
"""
|
|
Lightweight, idempotent schema top-ups.
|
|
|
|
`db.create_all()` creates missing tables but never alters existing ones, so a
|
|
new column on an existing table needs help. This adds only what's missing and
|
|
is safe to run on every boot, on both SQLite and Postgres.
|
|
|
|
It is not a replacement for real migrations — it handles additive column
|
|
changes, which is all LIM has needed so far.
|
|
|
|
Created by: Jess Rogerson (yelling commands at Claude.AI)
|
|
"""
|
|
|
|
from sqlalchemy import inspect, text
|
|
|
|
from app import db
|
|
|
|
# table -> column -> DDL fragment used in ALTER TABLE ... ADD COLUMN
|
|
ADDITIVE_COLUMNS = {
|
|
'sets': {
|
|
# Existing rows keep the behaviour they had before visibility existed:
|
|
# everyone could see every set, so they become 'public' rather than
|
|
# silently vanishing from other users' libraries. New sets use the
|
|
# site default instead.
|
|
'visibility': "VARCHAR(10) NOT NULL DEFAULT 'public'",
|
|
},
|
|
'instructions': {
|
|
# Written by the automatic fetcher; null on everything uploaded
|
|
# before it existed, which is fine - dedupe just can't help there.
|
|
'sha256': 'VARCHAR(64)',
|
|
'source': 'VARCHAR(20)',
|
|
},
|
|
}
|
|
|
|
# Tables added by later versions. create_all() is not called at boot, so new
|
|
# tables need an explicit (and idempotent) nudge.
|
|
NEW_TABLES = (
|
|
'app.models.instruction_fetch.InstructionFetchJob',
|
|
'app.models.api_call.BricksetApiCall',
|
|
'app.models.part_label.PartCategory',
|
|
'app.models.part_label.PartLabel',
|
|
)
|
|
|
|
|
|
def ensure_schema(app):
|
|
"""Add any missing columns. Returns a list of changes made."""
|
|
applied = []
|
|
with app.app_context():
|
|
try:
|
|
inspector = inspect(db.engine)
|
|
existing_tables = set(inspector.get_table_names())
|
|
except Exception as exc: # database not reachable yet
|
|
app.logger.warning('Schema check skipped: %s', exc)
|
|
return applied
|
|
|
|
for dotted in NEW_TABLES:
|
|
module_path, _, class_name = dotted.rpartition('.')
|
|
try:
|
|
module = __import__(module_path, fromlist=[class_name])
|
|
model = getattr(module, class_name)
|
|
if model.__tablename__ not in existing_tables:
|
|
model.__table__.create(db.engine, checkfirst=True)
|
|
applied.append(f'table {model.__tablename__}')
|
|
app.logger.info('Schema upgrade: created table %s',
|
|
model.__tablename__)
|
|
except Exception as exc:
|
|
app.logger.error('Schema upgrade failed for %s: %s',
|
|
dotted, exc)
|
|
|
|
for table, columns in ADDITIVE_COLUMNS.items():
|
|
if table not in existing_tables:
|
|
continue # create_all() will build it with everything present
|
|
present = {c['name'] for c in inspector.get_columns(table)}
|
|
for column, ddl in columns.items():
|
|
if column in present:
|
|
continue
|
|
statement = f'ALTER TABLE {table} ADD COLUMN {column} {ddl}'
|
|
try:
|
|
with db.engine.begin() as conn:
|
|
conn.execute(text(statement))
|
|
applied.append(f'{table}.{column}')
|
|
app.logger.info('Schema upgrade: added %s.%s', table, column)
|
|
except Exception as exc:
|
|
app.logger.error('Schema upgrade failed for %s.%s: %s',
|
|
table, column, exc)
|
|
|
|
return applied
|