mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-16 23:43:03 +00:00
fix: add idempotency to postgres migrations to prevent duplicate key/table errors
This commit is contained in:
Juan Equihua
parent
4e2240aada
commit
cf77c47a03
6 changed files with 211 additions and 117 deletions
|
|
@ -20,58 +20,90 @@ depends_on: Union[str, Sequence[str], None] = None
|
|||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
'calendar',
|
||||
sa.Column('id', sa.Text(), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('name', sa.Text(), nullable=False),
|
||||
sa.Column('color', sa.Text(), nullable=True),
|
||||
sa.Column('is_default', sa.Boolean(), nullable=False),
|
||||
sa.Column('data', sa.JSON(), nullable=True),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
op.create_index('ix_calendar_user', 'calendar', ['user_id'], unique=False)
|
||||
# Create index if it does not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename='calendar' AND indexname='ix_calendar_user'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.create_index('ix_calendar_user', 'calendar', ['user_id'], unique=False)
|
||||
|
||||
op.create_table(
|
||||
'calendar_event',
|
||||
sa.Column('id', sa.Text(), nullable=False),
|
||||
sa.Column('calendar_id', sa.Text(), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('title', sa.Text(), nullable=False),
|
||||
sa.Column('description', sa.Text(), nullable=True),
|
||||
sa.Column('start_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('end_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('all_day', sa.Boolean(), nullable=False),
|
||||
sa.Column('rrule', sa.Text(), nullable=True),
|
||||
sa.Column('color', sa.Text(), nullable=True),
|
||||
sa.Column('location', sa.Text(), nullable=True),
|
||||
sa.Column('data', sa.JSON(), nullable=True),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('is_cancelled', sa.Boolean(), nullable=False),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
op.create_index('ix_calendar_event_calendar', 'calendar_event', ['calendar_id', 'start_at'], unique=False)
|
||||
op.create_index('ix_calendar_event_user_date', 'calendar_event', ['user_id', 'start_at'], unique=False)
|
||||
# Create calendar_event table if it does not exist
|
||||
if not conn.dialect.has_table(conn, 'calendar_event'):
|
||||
op.create_table(
|
||||
'calendar_event',
|
||||
sa.Column('id', sa.Text(), nullable=False),
|
||||
sa.Column('calendar_id', sa.Text(), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('title', sa.Text(), nullable=False),
|
||||
sa.Column('description', sa.Text(), nullable=True),
|
||||
sa.Column('start_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('end_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('all_day', sa.Boolean(), nullable=False),
|
||||
sa.Column('rrule', sa.Text(), nullable=True),
|
||||
sa.Column('color', sa.Text(), nullable=True),
|
||||
sa.Column('location', sa.Text(), nullable=True),
|
||||
sa.Column('data', sa.JSON(), nullable=True),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('is_cancelled', sa.Boolean(), nullable=False),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
|
||||
op.create_table(
|
||||
'calendar_event_attendee',
|
||||
sa.Column('id', sa.Text(), nullable=False),
|
||||
sa.Column('event_id', sa.Text(), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('status', sa.Text(), nullable=False),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
sa.UniqueConstraint('event_id', 'user_id', name='uq_event_attendee'),
|
||||
)
|
||||
op.create_index('ix_calendar_event_attendee_user', 'calendar_event_attendee', ['user_id', 'status'], unique=False)
|
||||
# Create indexes if they do not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename='calendar_event' AND indexname='ix_calendar_event_calendar'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.create_index(
|
||||
'ix_calendar_event_calendar',
|
||||
'calendar_event',
|
||||
['calendar_id', 'start_at'],
|
||||
unique=False,
|
||||
)
|
||||
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename='calendar_event' AND indexname='ix_calendar_event_user_date'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.create_index(
|
||||
'ix_calendar_event_user_date',
|
||||
'calendar_event',
|
||||
['user_id', 'start_at'],
|
||||
unique=False,
|
||||
)
|
||||
|
||||
# Create calendar_event_attendee table if it does not exist
|
||||
if not conn.dialect.has_table(conn, 'calendar_event_attendee'):
|
||||
op.create_table(
|
||||
'calendar_event_attendee',
|
||||
sa.Column('id', sa.Text(), nullable=False),
|
||||
sa.Column('event_id', sa.Text(), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('status', sa.Text(), nullable=False),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
sa.UniqueConstraint('event_id', 'user_id', name='uq_event_attendee'),
|
||||
)
|
||||
|
||||
# Create index if it does not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename='calendar_event_attendee' "
|
||||
"AND indexname='ix_calendar_event_attendee_user'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.create_index(
|
||||
'ix_calendar_event_attendee_user',
|
||||
'calendar_event_attendee',
|
||||
['user_id', 'status'],
|
||||
unique=False,
|
||||
)
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index('ix_calendar_event_attendee_user', table_name='calendar_event_attendee')
|
||||
|
|
|
|||
|
|
@ -19,9 +19,22 @@ depends_on: Union[str, Sequence[str], None] = None
|
|||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column('chat', sa.Column('tasks', sa.JSON(), nullable=True))
|
||||
op.add_column('chat', sa.Column('summary', sa.Text(), nullable=True))
|
||||
conn = op.get_bind()
|
||||
# Add tasks column if it does not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name='chat' AND column_name='tasks'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.add_column('chat', sa.Column('tasks', sa.JSON(), nullable=True))
|
||||
|
||||
# Add summary column if it does not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name='chat' AND column_name='summary'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.add_column('chat', sa.Column('summary', sa.Text(), nullable=True))
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_column('chat', 'summary')
|
||||
|
|
|
|||
|
|
@ -18,9 +18,14 @@ depends_on = None
|
|||
|
||||
|
||||
def upgrade():
|
||||
op.add_column('chat', sa.Column('last_read_at', sa.BigInteger(), nullable=True))
|
||||
# Set existing chats to be marked as read
|
||||
op.execute('UPDATE chat SET last_read_at = updated_at')
|
||||
conn = op.get_bind()
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name='chat' AND column_name='last_read_at'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.add_column('chat', sa.Column('last_read_at', sa.BigInteger(), nullable=True))
|
||||
op.execute('UPDATE chat SET last_read_at = updated_at')
|
||||
|
||||
|
||||
def downgrade():
|
||||
|
|
|
|||
|
|
@ -63,16 +63,17 @@ def upgrade():
|
|||
conn = op.get_bind()
|
||||
|
||||
# 1. Create shared_chat table
|
||||
op.create_table(
|
||||
'shared_chat',
|
||||
sa.Column('id', sa.Text(), primary_key=True),
|
||||
sa.Column('chat_id', sa.Text(), sa.ForeignKey('chat.id', ondelete='CASCADE'), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('title', sa.Text(), nullable=True),
|
||||
sa.Column('chat', sa.JSON(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=True),
|
||||
)
|
||||
if not conn.dialect.has_table(conn, 'shared_chat'):
|
||||
op.create_table(
|
||||
'shared_chat',
|
||||
sa.Column('id', sa.Text(), primary_key=True),
|
||||
sa.Column('chat_id', sa.Text(), sa.ForeignKey('chat.id', ondelete='CASCADE'), nullable=False),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('title', sa.Text(), nullable=True),
|
||||
sa.Column('chat', sa.JSON(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=True),
|
||||
)
|
||||
|
||||
# 2. Migrate existing shared-* rows
|
||||
shared_rows = conn.execute(
|
||||
|
|
@ -96,31 +97,49 @@ def upgrade():
|
|||
if not original:
|
||||
continue
|
||||
|
||||
# Insert snapshot into shared_chat
|
||||
conn.execute(
|
||||
shared_chat_t.insert().values(
|
||||
id=share_token,
|
||||
chat_id=original_chat_id,
|
||||
user_id=original.user_id,
|
||||
title=row.title,
|
||||
chat=row.chat,
|
||||
created_at=row.created_at,
|
||||
updated_at=row.updated_at,
|
||||
)
|
||||
)
|
||||
# Insert snapshot into shared_chat (skip if already exists)
|
||||
existing_shared = conn.execute(
|
||||
sa.select(shared_chat_t.c.id).where(shared_chat_t.c.id == share_token)
|
||||
).fetchone()
|
||||
|
||||
# Create user:*:read grant for backward compat
|
||||
conn.execute(
|
||||
access_grant_t.insert().values(
|
||||
id=str(uuid.uuid4()),
|
||||
resource_type='shared_chat',
|
||||
resource_id=original_chat_id,
|
||||
principal_type='user',
|
||||
principal_id='*',
|
||||
permission='read',
|
||||
created_at=row.created_at or int(time.time()),
|
||||
if not existing_shared:
|
||||
conn.execute(
|
||||
shared_chat_t.insert().values(
|
||||
id=share_token,
|
||||
chat_id=original_chat_id,
|
||||
user_id=original.user_id,
|
||||
title=row.title,
|
||||
chat=row.chat,
|
||||
created_at=row.created_at,
|
||||
updated_at=row.updated_at,
|
||||
)
|
||||
)
|
||||
|
||||
# Create user:*:read grant for backward compat (skip if already exists)
|
||||
existing_grant = conn.execute(
|
||||
sa.select(access_grant_t.c.id).where(
|
||||
sa.and_(
|
||||
access_grant_t.c.resource_type == 'shared_chat',
|
||||
access_grant_t.c.resource_id == original_chat_id,
|
||||
access_grant_t.c.principal_type == 'user',
|
||||
access_grant_t.c.principal_id == '*',
|
||||
access_grant_t.c.permission == 'read',
|
||||
)
|
||||
)
|
||||
).fetchone()
|
||||
|
||||
if not existing_grant:
|
||||
conn.execute(
|
||||
access_grant_t.insert().values(
|
||||
id=str(uuid.uuid4()),
|
||||
resource_type='shared_chat',
|
||||
resource_id=original_chat_id,
|
||||
principal_type='user',
|
||||
principal_id='*',
|
||||
permission='read',
|
||||
created_at=row.created_at or int(time.time()),
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
# 3. Clean up old phantom rows
|
||||
conn.execute(
|
||||
|
|
@ -130,7 +149,6 @@ def upgrade():
|
|||
)
|
||||
conn.execute(chat_t.delete().where(chat_t.c.user_id.like('shared-%')))
|
||||
|
||||
|
||||
def downgrade():
|
||||
conn = op.get_bind()
|
||||
|
||||
|
|
|
|||
|
|
@ -17,35 +17,55 @@ depends_on = None
|
|||
|
||||
|
||||
def upgrade():
|
||||
op.create_table(
|
||||
'automation',
|
||||
sa.Column('id', sa.Text(), primary_key=True),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('name', sa.Text(), nullable=False),
|
||||
sa.Column('data', sa.JSON(), nullable=False),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
|
||||
sa.Column('last_run_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('next_run_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
)
|
||||
op.create_index('ix_automation_next_run', 'automation', ['next_run_at'])
|
||||
conn = op.get_bind()
|
||||
|
||||
op.create_table(
|
||||
'automation_run',
|
||||
sa.Column('id', sa.Text(), primary_key=True),
|
||||
sa.Column('automation_id', sa.Text(), nullable=False),
|
||||
sa.Column('chat_id', sa.Text(), nullable=True),
|
||||
sa.Column('status', sa.Text(), nullable=False),
|
||||
sa.Column('error', sa.Text(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
)
|
||||
op.create_index(
|
||||
'ix_automation_run_automation_id',
|
||||
'automation_run',
|
||||
['automation_id'],
|
||||
)
|
||||
# Create automation table if it does not exist
|
||||
if not conn.dialect.has_table(conn, 'automation'):
|
||||
op.create_table(
|
||||
'automation',
|
||||
sa.Column('id', sa.Text(), primary_key=True),
|
||||
sa.Column('user_id', sa.Text(), nullable=False),
|
||||
sa.Column('name', sa.Text(), nullable=False),
|
||||
sa.Column('data', sa.JSON(), nullable=False),
|
||||
sa.Column('meta', sa.JSON(), nullable=True),
|
||||
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
|
||||
sa.Column('last_run_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('next_run_at', sa.BigInteger(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
sa.Column('updated_at', sa.BigInteger(), nullable=False),
|
||||
)
|
||||
|
||||
# Create index if it does not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename='automation' AND indexname='ix_automation_next_run'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.create_index('ix_automation_next_run', 'automation', ['next_run_at'])
|
||||
|
||||
# Create automation_run table if it does not exist
|
||||
if not conn.dialect.has_table(conn, 'automation_run'):
|
||||
op.create_table(
|
||||
'automation_run',
|
||||
sa.Column('id', sa.Text(), primary_key=True),
|
||||
sa.Column('automation_id', sa.Text(), nullable=False),
|
||||
sa.Column('chat_id', sa.Text(), nullable=True),
|
||||
sa.Column('status', sa.Text(), nullable=False),
|
||||
sa.Column('error', sa.Text(), nullable=True),
|
||||
sa.Column('created_at', sa.BigInteger(), nullable=False),
|
||||
)
|
||||
|
||||
# Create index if it does not exist
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT indexname FROM pg_indexes "
|
||||
"WHERE tablename='automation_run' AND indexname='ix_automation_run_automation_id'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.create_index(
|
||||
'ix_automation_run_automation_id',
|
||||
'automation_run',
|
||||
['automation_id'],
|
||||
)
|
||||
|
||||
|
||||
def downgrade():
|
||||
|
|
|
|||
|
|
@ -16,7 +16,13 @@ depends_on = None
|
|||
|
||||
|
||||
def upgrade():
|
||||
op.add_column('note', sa.Column('is_pinned', sa.Boolean(), nullable=True))
|
||||
conn = op.get_bind()
|
||||
result = conn.execute(sa.text(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_name='note' AND column_name='is_pinned'"
|
||||
))
|
||||
if result.fetchone() is None:
|
||||
op.add_column('note', sa.Column('is_pinned', sa.Boolean(), nullable=True))
|
||||
|
||||
|
||||
def downgrade():
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue