engineering/remarkbox
1
0
Fork 0
Code Issues 3 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Make Alembic migrations idempotent for SQLite

Browse source 
Check if columns/tables exist before adding them so migrations
don't fail on re-run when schema was applied out of band.
This commit is contained in:
russell@unturf.com 2026-02-02 08:31:36 -05:00
parent b6739fff70
commit 8d836dae34
2 changed files with 51 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

22
remarkbox/scripts/alembic/versions/896568b0752e_add_customizable_button_text_comment_.py
View file

@ -16,14 +16,26 @@ from alembic import op
import sqlalchemy as sa
def _column_exists(table, column):
"""Check if a column exists in a SQLite table."""
conn = op.get_bind()
result = conn.execute(sa.text("PRAGMA table_info('{}')".format(table)))
return any(row[1] == column for row in result)
def upgrade():
# T4: customizable button text and comment labels
op.add_column('rb_namespace', sa.Column('submit_button_text', sa.Unicode(length=256), nullable=True))
op.add_column('rb_namespace', sa.Column('comment_label_singular', sa.Unicode(length=256), nullable=True))
op.add_column('rb_namespace', sa.Column('comment_label_plural', sa.Unicode(length=256), nullable=True))
if not _column_exists('rb_namespace', 'submit_button_text'):
op.add_column('rb_namespace', sa.Column('submit_button_text', sa.Unicode(length=256), nullable=True))
if not _column_exists('rb_namespace', 'comment_label_singular'):
op.add_column('rb_namespace', sa.Column('comment_label_singular', sa.Unicode(length=256), nullable=True))
if not _column_exists('rb_namespace', 'comment_label_plural'):
op.add_column('rb_namespace', sa.Column('comment_label_plural', sa.Unicode(length=256), nullable=True))
# T8: nesting depth settings
op.add_column('rb_namespace', sa.Column('max_nesting_depth', sa.Integer(), nullable=True))
op.add_column('rb_namespace', sa.Column('collapse_depth', sa.Integer(), nullable=True))
if not _column_exists('rb_namespace', 'max_nesting_depth'):
op.add_column('rb_namespace', sa.Column('max_nesting_depth', sa.Integer(), nullable=True))
if not _column_exists('rb_namespace', 'collapse_depth'):
op.add_column('rb_namespace', sa.Column('collapse_depth', sa.Integer(), nullable=True))
def downgrade():

49
remarkbox/scripts/alembic/versions/b7f3a2d1e8c9_add_push_notifications_and_webmentions.py
View file

@ -18,25 +18,44 @@ import sqlalchemy as sa
from remarkbox.models.meta import UUIDType
def _column_exists(table, column):
"""Check if a column exists in a SQLite table."""
conn = op.get_bind()
result = conn.execute(sa.text("PRAGMA table_info('{}')".format(table)))
return any(row[1] == column for row in result)
def _table_exists(table):
"""Check if a table exists in SQLite."""
conn = op.get_bind()
result = conn.execute(sa.text(
"SELECT name FROM sqlite_master WHERE type='table' AND name='{}'".format(table)
))
return result.fetchone() is not None
def upgrade():
# T10: push notification preferences on user
op.add_column('rb_user', sa.Column('notification_preference', sa.Unicode(length=5), server_default='email', nullable=False))
op.add_column('rb_user', sa.Column('push_subscriptions', sa.UnicodeText(), nullable=True))
if not _column_exists('rb_user', 'notification_preference'):
op.add_column('rb_user', sa.Column('notification_preference', sa.Unicode(length=5), server_default='email', nullable=False))
if not _column_exists('rb_user', 'push_subscriptions'):
op.add_column('rb_user', sa.Column('push_subscriptions', sa.UnicodeText(), nullable=True))
# T7: webmention table
op.create_table(
'rb_webmention',
sa.Column('id', UUIDType, primary_key=True, index=True),
sa.Column('source', sa.Unicode(2048), nullable=False),
sa.Column('target', sa.Unicode(2048), nullable=False),
sa.Column('node_id', UUIDType, sa.ForeignKey('rb_node.id'), index=True, nullable=True),
sa.Column('verified', sa.Boolean(), default=False, nullable=False),
sa.Column('author_name', sa.Unicode(256), nullable=True),
sa.Column('author_url', sa.Unicode(2048), nullable=True),
sa.Column('content', sa.UnicodeText(), nullable=True),
sa.Column('created_timestamp', sa.BigInteger(), nullable=False),
sa.Column('updated_timestamp', sa.BigInteger(), nullable=False),
)
if not _table_exists('rb_webmention'):
op.create_table(
'rb_webmention',
sa.Column('id', UUIDType, primary_key=True, index=True),
sa.Column('source', sa.Unicode(2048), nullable=False),
sa.Column('target', sa.Unicode(2048), nullable=False),
sa.Column('node_id', UUIDType, sa.ForeignKey('rb_node.id'), index=True, nullable=True),
sa.Column('verified', sa.Boolean(), default=False, nullable=False),
sa.Column('author_name', sa.Unicode(256), nullable=True),
sa.Column('author_url', sa.Unicode(2048), nullable=True),
sa.Column('content', sa.UnicodeText(), nullable=True),
sa.Column('created_timestamp', sa.BigInteger(), nullable=False),
sa.Column('updated_timestamp', sa.BigInteger(), nullable=False),
)
def downgrade():