Add API endpoint to clear cached scripts for workflows (#4809)
Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Shuchang Zheng <wintonzheng0325@gmail.com> Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: Shuchang Zheng <wintonzheng@users.noreply.github.com>
This commit is contained in:
@@ -10,6 +10,7 @@ from skyvern.forge.sdk.routes.routers import base_router
|
||||
from skyvern.forge.sdk.schemas.organizations import Organization
|
||||
from skyvern.forge.sdk.services import org_auth_service
|
||||
from skyvern.schemas.scripts import (
|
||||
ClearCacheResponse,
|
||||
CreateScriptRequest,
|
||||
CreateScriptResponse,
|
||||
DeployScriptRequest,
|
||||
@@ -573,3 +574,68 @@ async def delete_workflow_cache_key_value(
|
||||
raise HTTPException(status_code=404, detail="Cache key value not found")
|
||||
|
||||
return {"message": "Cache key value deleted successfully"}
|
||||
|
||||
|
||||
@base_router.delete(
|
||||
"/scripts/{workflow_permanent_id}/cache",
|
||||
response_model=ClearCacheResponse,
|
||||
summary="Clear cached scripts for workflow",
|
||||
description="Clear all cached scripts for a specific workflow. This will trigger script regeneration on subsequent runs.",
|
||||
tags=["Scripts"],
|
||||
openapi_extra={
|
||||
"x-fern-sdk-method-name": "clear_workflow_cache",
|
||||
},
|
||||
)
|
||||
@base_router.delete(
|
||||
"/scripts/{workflow_permanent_id}/cache/",
|
||||
response_model=ClearCacheResponse,
|
||||
include_in_schema=False,
|
||||
)
|
||||
async def clear_workflow_cache(
|
||||
workflow_permanent_id: str = Path(
|
||||
...,
|
||||
description="The workflow permanent ID to clear cache for",
|
||||
examples=["wpid_abc123"],
|
||||
),
|
||||
current_org: Organization = Depends(org_auth_service.get_current_org),
|
||||
) -> ClearCacheResponse:
|
||||
"""Clear all cached scripts for a specific workflow."""
|
||||
LOG.info(
|
||||
"Clearing workflow cache",
|
||||
organization_id=current_org.organization_id,
|
||||
workflow_permanent_id=workflow_permanent_id,
|
||||
)
|
||||
|
||||
# Verify workflow exists
|
||||
workflow = await app.DATABASE.get_workflow_by_permanent_id(
|
||||
workflow_permanent_id=workflow_permanent_id,
|
||||
organization_id=current_org.organization_id,
|
||||
)
|
||||
|
||||
if not workflow:
|
||||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||||
|
||||
# Clear database cache (soft delete)
|
||||
deleted_count = await app.DATABASE.delete_workflow_scripts_by_permanent_id(
|
||||
organization_id=current_org.organization_id,
|
||||
workflow_permanent_id=workflow_permanent_id,
|
||||
)
|
||||
|
||||
# Clear in-memory cache
|
||||
cache_cleared_count = workflow_script_service.clear_workflow_script_cache(
|
||||
organization_id=current_org.organization_id,
|
||||
workflow_permanent_id=workflow_permanent_id,
|
||||
)
|
||||
|
||||
LOG.info(
|
||||
"Cleared workflow cache",
|
||||
organization_id=current_org.organization_id,
|
||||
workflow_permanent_id=workflow_permanent_id,
|
||||
deleted_count=deleted_count,
|
||||
cache_cleared_count=cache_cleared_count,
|
||||
)
|
||||
|
||||
return ClearCacheResponse(
|
||||
deleted_count=deleted_count,
|
||||
message=f"Successfully cleared {deleted_count} database record(s) and {cache_cleared_count} in-memory cache entry(s) for workflow {workflow_permanent_id}",
|
||||
)
|
||||
|
||||
@@ -182,3 +182,10 @@ class WorkflowScript(BaseModel):
|
||||
created_at: datetime
|
||||
modified_at: datetime
|
||||
deleted_at: datetime | None = None
|
||||
|
||||
|
||||
class ClearCacheResponse(BaseModel):
|
||||
"""Response model for cache clearing operations."""
|
||||
|
||||
deleted_count: int = Field(..., description="Number of cached entries deleted")
|
||||
message: str = Field(..., description="Status message")
|
||||
|
||||
@@ -57,6 +57,41 @@ def _make_workflow_script_cache_key(
|
||||
return (organization_id, workflow_permanent_id, cache_key_value, workflow_run_id, cache_key, statuses_key)
|
||||
|
||||
|
||||
def clear_workflow_script_cache(
|
||||
organization_id: str,
|
||||
workflow_permanent_id: str | None = None,
|
||||
) -> int:
|
||||
"""
|
||||
Clear in-memory cached scripts for a workflow or all workflows in an organization.
|
||||
|
||||
Args:
|
||||
organization_id: The organization ID to clear cache for.
|
||||
workflow_permanent_id: Optional workflow permanent ID. If None, clears all workflows.
|
||||
|
||||
Returns:
|
||||
The number of cache entries cleared.
|
||||
"""
|
||||
keys_to_delete = []
|
||||
|
||||
for key in list(_workflow_script_cache.keys()):
|
||||
# Key format: (org_id, workflow_permanent_id, cache_key_value, workflow_run_id, cache_key, statuses_key)
|
||||
if len(key) >= 2 and key[0] == organization_id:
|
||||
if workflow_permanent_id is None or key[1] == workflow_permanent_id:
|
||||
keys_to_delete.append(key)
|
||||
|
||||
for key in keys_to_delete:
|
||||
_workflow_script_cache.pop(key, None)
|
||||
|
||||
LOG.info(
|
||||
"Cleared workflow script in-memory cache",
|
||||
organization_id=organization_id,
|
||||
workflow_permanent_id=workflow_permanent_id,
|
||||
cleared_count=len(keys_to_delete),
|
||||
)
|
||||
|
||||
return len(keys_to_delete)
|
||||
|
||||
|
||||
async def generate_or_update_pending_workflow_script(
|
||||
workflow_run: WorkflowRun,
|
||||
workflow: Workflow,
|
||||
|
||||
Reference in New Issue
Block a user