from logging import getLogger from sqlalchemy import Select, select from sqlalchemy.ext.asyncio import AsyncSession from src import models, schemas from src.config import settings from src.crud.workspace import get_workspace from src.exceptions import ResourceNotFoundException from src.utils.types import GetOrCreateResult logger = getLogger(__name__) async def get_or_create_webhook_endpoint( db: AsyncSession, workspace_name: str, webhook: schemas.WebhookEndpointCreate, ) -> GetOrCreateResult[schemas.WebhookEndpoint]: """ Get an existing webhook endpoint for a workspace or create it if missing. Args: db: Database session workspace_name: Name of the workspace webhook: Webhook endpoint creation schema Returns: GetOrCreateResult containing the webhook endpoint and whether it was created Raises: ResourceNotFoundException: If the workspace does not exist ValueError: If the workspace already has the maximum number of webhook endpoints """ # Verify workspace exists await get_workspace(db, workspace_name=workspace_name) stmt = select(models.WebhookEndpoint).where( models.WebhookEndpoint.workspace_name == workspace_name, ) result = await db.execute(stmt) endpoints = result.scalars().all() # Check if webhook already exists for this workspace for endpoint in endpoints: if endpoint.url == webhook.url: return GetOrCreateResult( schemas.WebhookEndpoint.model_validate(endpoint), created=False ) # No more than WORKSPACE_LIMIT webhooks per workspace if len(endpoints) >= settings.WEBHOOK.MAX_WORKSPACE_LIMIT: raise ValueError( f"Maximum number of webhook endpoints ({settings.WEBHOOK.MAX_WORKSPACE_LIMIT}) reached for this workspace." ) # Create new webhook endpoint webhook_endpoint = models.WebhookEndpoint( workspace_name=workspace_name, url=webhook.url, ) db.add(webhook_endpoint) await db.commit() logger.debug("Webhook endpoint created: %s", webhook.url) return GetOrCreateResult( schemas.WebhookEndpoint.model_validate(webhook_endpoint), created=True ) async def list_webhook_endpoints( workspace_name: str, ) -> Select[tuple[models.WebhookEndpoint]]: """ List all webhook endpoints, optionally filtered by workspace. Args: db: Database session workspace_name: Name of the workspace (optional) Returns: List of webhook endpoints """ return select(models.WebhookEndpoint).where( models.WebhookEndpoint.workspace_name == workspace_name ) async def delete_webhook_endpoint( db: AsyncSession, workspace_name: str, endpoint_id: str ) -> None: """ Delete a webhook endpoint. Args: db: Database session endpoint_id: ID of the webhook endpoint Raises: ResourceNotFoundException: If the webhook endpoint is not found """ # Verify webhook endpoint exists stmt = select(models.WebhookEndpoint).where( models.WebhookEndpoint.id == endpoint_id, models.WebhookEndpoint.workspace_name == workspace_name, ) result = await db.execute(stmt) endpoint = result.scalar_one_or_none() if not endpoint: raise ResourceNotFoundException( f"Webhook endpoint {endpoint_id} not found for workspace {workspace_name}" ) await db.delete(endpoint) await db.commit() logger.debug("Webhook endpoint %s deleted", endpoint_id)