refac

backend/open_webui/routers/pipelines.py

@@ -18,6 +18,7 @@ from fastapi import (
 from open_webui.config import CACHE_DIR
 from open_webui.constants import ERROR_MESSAGES
 from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL
+from open_webui.events import EVENTS, publish_event
 from open_webui.models.config import Config
 from open_webui.routers.openai import get_all_models_responses
 from open_webui.utils.auth import get_admin_user
@@ -262,6 +263,13 @@ async def upload_pipeline(
                     response.raise_for_status()
                     data = await response.json()
 
+        await publish_event(
+            request,
+            EVENTS.PIPELINE_UPLOADED,
+            actor=user,
+            subject_id=data.get('id') or filename,
+            data={'url_idx': urlIdx, 'filename': filename},
+        )
         return {**data}
     except Exception as e:
         # Handle connection error here
@@ -311,6 +319,13 @@ async def add_pipeline(request: Request, form_data: AddPipelineForm, user=Depend
                 response.raise_for_status()
                 data = await response.json()
 
+        await publish_event(
+            request,
+            EVENTS.PIPELINE_ADDED,
+            actor=user,
+            subject_id=data.get('id') or form_data.url,
+            data={'url_idx': urlIdx, 'url': form_data.url},
+        )
         return {**data}
     except Exception as e:
         # Handle connection error here
@@ -354,6 +369,13 @@ async def delete_pipeline(request: Request, form_data: DeletePipelineForm, user=
                 response.raise_for_status()
                 data = await response.json()
 
+        await publish_event(
+            request,
+            EVENTS.PIPELINE_DELETED,
+            actor=user,
+            subject_id=form_data.id,
+            data={'url_idx': urlIdx},
+        )
         return {**data}
     except Exception as e:
         # Handle connection error here
@@ -389,6 +411,13 @@ async def get_pipelines(request: Request, urlIdx: Optional[int] = None, user=Dep
                 response.raise_for_status()
                 data = await response.json()
 
+        await publish_event(
+            request,
+            EVENTS.PIPELINE_VALVES_UPDATED,
+            actor=user,
+            subject_id=pipeline_id,
+            data={'url_idx': urlIdx},
+        )
         return {**data}
     except Exception as e:
         # Handle connection error here

backend/open_webui/routers/skills.py

@@ -4,6 +4,7 @@ from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL
 from open_webui.constants import ERROR_MESSAGES
+from open_webui.events import EVENTS, publish_event
 from open_webui.internal.db import get_async_session
 from open_webui.models.access_grants import AccessGrants
 from open_webui.models.config import Config
@@ -190,6 +191,13 @@ async def create_new_skill(
     try:
         skill = await Skills.insert_new_skill(user.id, form_data, db=db)
         if skill:
+            await publish_event(
+                request,
+                EVENTS.SKILL_CREATED,
+                actor=user,
+                subject_id=skill.id,
+                data={'name': skill.name},
+            )
             return skill
         else:
             raise HTTPException(
@@ -308,6 +316,13 @@ async def update_skill_by_id(
         skill = await Skills.update_skill_by_id(id, updated, db=db)
 
         if skill:
+            await publish_event(
+                request,
+                EVENTS.SKILL_UPDATED,
+                actor=user,
+                subject_id=skill.id,
+                data={'name': skill.name},
+            )
             return skill
         else:
             raise HTTPException(
@@ -371,7 +386,15 @@ async def update_skill_access_by_id(
 
     await AccessGrants.set_access_grants('skill', id, form_data.access_grants, db=db)
 
-    return await Skills.get_skill_by_id(id, db=db)
+    skill = await Skills.get_skill_by_id(id, db=db)
+    await publish_event(
+        request,
+        EVENTS.SKILL_UPDATED,
+        actor=user,
+        subject_id=id,
+        data={'access_updated': True, 'name': skill.name if skill else None},
+    )
+    return skill
 
 
 ############################
@@ -380,7 +403,12 @@ async def update_skill_access_by_id(
 
 
 @router.post('/id/{id}/toggle', response_model=Optional[SkillModel])
-async def toggle_skill_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)):
+async def toggle_skill_by_id(
+    request: Request,
+    id: str,
+    user=Depends(get_verified_user),
+    db: AsyncSession = Depends(get_async_session),
+):
     skill = await Skills.get_skill_by_id(id, db=db)
     if skill:
         if (
@@ -397,6 +425,13 @@ async def toggle_skill_by_id(id: str, user=Depends(get_verified_user), db: Async
             skill = await Skills.toggle_skill_by_id(id, db=db)
 
             if skill:
+                await publish_event(
+                    request,
+                    EVENTS.SKILL_ENABLED if skill.is_active else EVENTS.SKILL_DISABLED,
+                    actor=user,
+                    subject_id=skill.id,
+                    data={'name': skill.name},
+                )
                 return skill
             else:
                 raise HTTPException(
@@ -451,4 +486,12 @@ async def delete_skill_by_id(
         )
 
     result = await Skills.delete_skill_by_id(id, db=db)
+    if result:
+        await publish_event(
+            request,
+            EVENTS.SKILL_DELETED,
+            actor=user,
+            subject_id=id,
+            data={'name': skill.name},
+        )
     return result