Added filters in the webhooks

migrations/versions/40affbf3022b_add_filter_colume_in_webhooks.py

@@ -0,0 +1,84 @@
+"""add filter colume in webhooks
+
+Revision ID: 40affbf3022b
+Revises: 5d5f801f28e7
+Create Date: 2023-08-28 12:30:35.171176
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '40affbf3022b'
+down_revision = '5d5f801f28e7'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index('ix_agent_schedule_agent_id', table_name='agent_schedule')
+    op.drop_index('ix_agent_schedule_expiry_date', table_name='agent_schedule')
+    op.drop_index('ix_agent_schedule_status', table_name='agent_schedule')
+    op.alter_column('agent_workflow_steps', 'unique_id',
+               existing_type=sa.VARCHAR(),
+               nullable=True)
+    op.alter_column('agent_workflow_steps', 'step_type',
+               existing_type=sa.VARCHAR(),
+               nullable=True)
+    op.drop_column('agent_workflows', 'organisation_id')
+    op.drop_index('ix_events_agent_id', table_name='events')
+    op.drop_index('ix_events_event_property', table_name='events')
+    op.drop_index('ix_events_org_id', table_name='events')
+    op.alter_column('knowledge_configs', 'knowledge_id',
+               existing_type=sa.INTEGER(),
+               nullable=True)
+    op.alter_column('knowledges', 'name',
+               existing_type=sa.VARCHAR(),
+               nullable=True)
+    op.alter_column('vector_db_configs', 'vector_db_id',
+               existing_type=sa.INTEGER(),
+               nullable=True)
+    op.alter_column('vector_db_indices', 'name',
+               existing_type=sa.VARCHAR(),
+               nullable=True)
+    op.alter_column('vector_dbs', 'name',
+               existing_type=sa.VARCHAR(),
+               nullable=True)
+    op.add_column('webhooks', sa.Column('filters', sa.JSON(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('webhooks', 'filters')
+    op.alter_column('vector_dbs', 'name',
+               existing_type=sa.VARCHAR(),
+               nullable=False)
+    op.alter_column('vector_db_indices', 'name',
+               existing_type=sa.VARCHAR(),
+               nullable=False)
+    op.alter_column('vector_db_configs', 'vector_db_id',
+               existing_type=sa.INTEGER(),
+               nullable=False)
+    op.alter_column('knowledges', 'name',
+               existing_type=sa.VARCHAR(),
+               nullable=False)
+    op.alter_column('knowledge_configs', 'knowledge_id',
+               existing_type=sa.INTEGER(),
+               nullable=False)
+    op.create_index('ix_events_org_id', 'events', ['org_id'], unique=False)
+    op.create_index('ix_events_event_property', 'events', ['event_property'], unique=False)
+    op.create_index('ix_events_agent_id', 'events', ['agent_id'], unique=False)
+    op.add_column('agent_workflows', sa.Column('organisation_id', sa.INTEGER(), autoincrement=False, nullable=True))
+    op.alter_column('agent_workflow_steps', 'step_type',
+               existing_type=sa.VARCHAR(),
+               nullable=False)
+    op.alter_column('agent_workflow_steps', 'unique_id',
+               existing_type=sa.VARCHAR(),
+               nullable=False)
+    op.create_index('ix_agent_schedule_status', 'agent_schedule', ['status'], unique=False)
+    op.create_index('ix_agent_schedule_expiry_date', 'agent_schedule', ['expiry_date'], unique=False)
+    op.create_index('ix_agent_schedule_agent_id', 'agent_schedule', ['agent_id'], unique=False)
+    # ### end Alembic commands ###

superagi/controllers/webhook.py

@@ -17,6 +17,7 @@ class WebHookIn(BaseModel):
     name: str
     url: str
     headers: dict
+    filters: dict
 
     class Config:
         orm_mode = True
@@ -52,7 +53,7 @@ def create_webhook(webhook: WebHookIn, Authorize: AuthJWT = Depends(check_auth),
             HTTPException (Status Code=404): If the associated project is not found.
     """
     db_webhook = Webhooks(name=webhook.name, url=webhook.url, headers=webhook.headers, org_id=organisation.id,
-                          is_deleted=False)
+                          is_deleted=False, filters=webhook.filters)
     db.session.add(db_webhook)
     db.session.commit()
     db.session.flush()

superagi/helper/webhook_manager.py

@@ -5,6 +5,7 @@
 import requests
 import json
 from superagi.lib.logger import logger
+
 class WebHookManager:
     def __init__(self,session):
         self.session=session
@@ -18,20 +19,21 @@
         org_webhooks=self.session.query(Webhooks).filter(Webhooks.org_id == org.id).all()
 
         for webhook_obj in org_webhooks:
-            webhook_obj_body={"agent_id":agent_id,"org_id":org.id,"event":f"{old_status} to {curr_status}"}
-            error=None
-            request=None
-            status='sent'
-            try:
-                request = requests.post(webhook_obj.url.strip(), data=json.dumps(webhook_obj_body), headers=webhook_obj.headers)
-            except Exception as e:
-                logger.error(f"Exception occured in webhooks {e}")
-                error=str(e)
-            if request is not None and request.status_code not in [200,201] and error is None:
-                error=request.text
-            if error is not None:
-                status='Error'
-            webhook_event=WebhookEvents(agent_id=agent_id, run_id=agent_execution_id, event=f"{old_status} to {curr_status}", status=status, errors=error)
-            self.session.add(webhook_event)
-            self.session.commit()
+            if "status" in webhook_obj.filters and curr_status in webhook_obj.filters["status"]:
+                webhook_obj_body={"agent_id":agent_id,"org_id":org.id,"event":f"{old_status} to {curr_status}"}
+                error=None
+                request=None
+                status='sent'
+                try:
+                    request = requests.post(webhook_obj.url.strip(), data=json.dumps(webhook_obj_body), headers=webhook_obj.headers)
+                except Exception as e:
+                    logger.error(f"Exception occured in webhooks {e}")
+                    error=str(e)
+                if request is not None and request.status_code not in [200,201] and error is None:
+                    error=request.text
+                if error is not None:
+                    status='Error'
+                webhook_event=WebhookEvents(agent_id=agent_id, run_id=agent_execution_id, event=f"{old_status} to {curr_status}", status=status, errors=error)
+                self.session.add(webhook_event)
+                self.session.commit()
 

superagi/models/webhooks.py

@@ -20,3 +20,4 @@ class Webhooks(DBBaseModel):
     url = Column(String)
     headers=Column(JSON)
     is_deleted=Column(Boolean)
+    filters=Column(JSON)