feat: make all db migrations SQL-only

src/langbot/pkg/persistence/migrations/dbm002_combine_quote_msg_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(2)
@@ -11,30 +10,45 @@ class DBMigrateCombineQuoteMsgConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure 'trigger' exists
+            if 'trigger' not in config:
+                config['trigger'] = {}
+
+            # Ensure 'misc' exists in 'trigger'
             if 'misc' not in config['trigger']:
                 config['trigger']['misc'] = {}
 
+            # Add 'combine-quote-message' if not exists
             if 'combine-quote-message' not in config['trigger']['misc']:
                 config['trigger']['misc']['combine-quote-message'] = False
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    {
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                    ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm003_n8n_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(3)
@@ -11,14 +10,23 @@ class DBMigrateN8nConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure 'ai' exists
+            if 'ai' not in config:
+                config['ai'] = {}
+
+            # Add 'n8n-service-api' if not exists
             if 'n8n-service-api' not in config['ai']:
                 config['ai']['n8n-service-api'] = {
                     'webhook-url': 'http://your-n8n-webhook-url',
@@ -33,16 +41,21 @@ class DBMigrateN8nConfig(migration.DBMigration):
                     'output-key': 'response',
                 }
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    {
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                    ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm004_rag_kb_uuid.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(4)
@@ -11,27 +10,43 @@ class DBMigrateRAGKBUUID(migration.DBMigration):
 
     async def upgrade(self):
         """升级"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure nested structure exists
+            if 'ai' not in config:
+                config['ai'] = {}
+            if 'local-agent' not in config['ai']:
+                config['ai']['local-agent'] = {}
+
+            # Add 'knowledge-base' if not exists
             if 'knowledge-base' not in config['ai']['local-agent']:
                 config['ai']['local-agent']['knowledge-base'] = ''
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    {
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                    ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """降级"""

src/langbot/pkg/persistence/migrations/dbm005_pipeline_remove_cot_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(5)
@@ -11,27 +10,43 @@ class DBMigratePipelineRemoveCotConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure nested structure exists
+            if 'output' not in config:
+                config['output'] = {}
+            if 'misc' not in config['output']:
+                config['output']['misc'] = {}
+
+            # Add 'remove-think' if not exists
             if 'remove-think' not in config['output']['misc']:
                 config['output']['misc']['remove-think'] = False
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    {
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                    ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm006_langflow_api_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(6)
@@ -11,14 +10,23 @@ class DBMigrateLangflowApiConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure 'ai' exists
+            if 'ai' not in config:
+                config['ai'] = {}
+
+            # Add 'langflow-api' if not exists
             if 'langflow-api' not in config['ai']:
                 config['ai']['langflow-api'] = {
                     'base-url': 'http://localhost:7860',
@@ -29,16 +37,21 @@ class DBMigrateLangflowApiConfig(migration.DBMigration):
                     'tweaks': '{}',
                 }
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    {
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                    ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm010_pipeline_multi_knowledge_base.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(10)
@@ -11,16 +10,20 @@ class DBMigratePipelineMultiKnowledgeBase(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
             # Convert knowledge-base from string to array
-            if 'local-agent' in config['ai']:
+            if 'ai' in config and 'local-agent' in config['ai']:
                 current_kb = config['ai']['local-agent'].get('knowledge-base', '')
 
                 # If it's already a list, skip
@@ -37,29 +40,38 @@ class DBMigratePipelineMultiKnowledgeBase(migration.DBMigration):
                 if 'knowledge-base' in config['ai']['local-agent']:
                     del config['ai']['local-agent']['knowledge-base']
 
-            await self.ap.persistence_mgr.execute_async(
+                # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+                if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                    await self.ap.persistence_mgr.execute_async(
-                .values(
+                        sqlalchemy.text(
-                    {
+                            'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                        ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                    )
-                )
+                else:
-            )
+                    await self.ap.persistence_mgr.execute_async(
+                        sqlalchemy.text(
+                            'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                        ),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                    )
 
     async def downgrade(self):
         """Downgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
             # Convert knowledge-bases from array back to string
-            if 'local-agent' in config['ai']:
+            if 'ai' in config and 'local-agent' in config['ai']:
                 current_kbs = config['ai']['local-agent'].get('knowledge-bases', [])
 
                 # If it's already a string, skip
@@ -76,13 +88,18 @@ class DBMigratePipelineMultiKnowledgeBase(migration.DBMigration):
                 if 'knowledge-bases' in config['ai']['local-agent']:
                     del config['ai']['local-agent']['knowledge-bases']
 
-            await self.ap.persistence_mgr.execute_async(
+                # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+                if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                    await self.ap.persistence_mgr.execute_async(
-                .values(
+                        sqlalchemy.text(
-                    {
+                            'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                        ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                    )
-                )
+                else:
-            )
+                    await self.ap.persistence_mgr.execute_async(
+                        sqlalchemy.text(
+                            'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                        ),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                    )

src/langbot/pkg/persistence/migrations/dbm011_dify_base_prompt_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(11)
@@ -11,29 +10,45 @@ class DBMigrateDifyApiConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure nested structure exists
+            if 'ai' not in config:
+                config['ai'] = {}
+            if 'dify-service-api' not in config['ai']:
+                config['ai']['dify-service-api'] = {}
+
+            # Add 'base-prompt' if not exists
             if 'base-prompt' not in config['ai']['dify-service-api']:
                 config['ai']['dify-service-api']['base-prompt'] = (
                     'When the file content is readable, please read the content of this file. When the file is an image, describe the content of this image.',
                 )
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    {
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                        'config': config,
+                    ),
-                        'for_version': self.ap.ver_mgr.get_current_version(),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
-                    }
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm012_pipeline_extensions_enable_all.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
+import json
-from ...entity.persistence import pipeline as persistence_pipeline
 
 
 @migration.migration_class(12)
@@ -11,14 +10,25 @@ class DBMigratePipelineExtensionsEnableAll(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
+        # Read all pipelines using raw SQL
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, extensions_preferences FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
+        current_version = self.ap.ver_mgr.get_current_version()
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-            extensions_preferences = serialized_pipeline['extensions_preferences']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            extensions_preferences = (
+                json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
+            )
 
+            # Ensure extensions_preferences is a dict
+            if extensions_preferences is None:
+                extensions_preferences = {}
+
+            # Add 'enable_all_plugins' if not exists
             if 'enable_all_plugins' not in extensions_preferences:
                 if 'plugins' in extensions_preferences:
                     extensions_preferences['enable_all_plugins'] = False
@@ -26,6 +36,7 @@ class DBMigratePipelineExtensionsEnableAll(migration.DBMigration):
                     extensions_preferences['enable_all_plugins'] = True
                     extensions_preferences['plugins'] = []
 
+            # Add 'enable_all_mcp_servers' if not exists
             if 'enable_all_mcp_servers' not in extensions_preferences:
                 if 'mcp_servers' in extensions_preferences:
                     extensions_preferences['enable_all_mcp_servers'] = False
@@ -33,14 +44,29 @@ class DBMigratePipelineExtensionsEnableAll(migration.DBMigration):
                     extensions_preferences['enable_all_mcp_servers'] = True
                     extensions_preferences['mcp_servers'] = []
 
-            await self.ap.persistence_mgr.execute_async(
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
+            if self.ap.persistence_mgr.db.name == 'postgresql':
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
+                await self.ap.persistence_mgr.execute_async(
-                .values(
+                    sqlalchemy.text(
-                    extensions_preferences=extensions_preferences,
+                        'UPDATE legacy_pipelines SET extensions_preferences = :extensions_preferences::jsonb, for_version = :for_version WHERE uuid = :uuid'
-                    for_version=self.ap.ver_mgr.get_current_version(),
+                    ),
+                    {
+                        'extensions_preferences': json.dumps(extensions_preferences),
+                        'for_version': current_version,
+                        'uuid': uuid,
+                    },
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET extensions_preferences = :extensions_preferences, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {
+                        'extensions_preferences': json.dumps(extensions_preferences),
+                        'for_version': current_version,
+                        'uuid': uuid,
+                    },
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""