mirror of
https://github.com/johndoe6345789/metabuilder.git
synced 2026-04-25 14:25:02 +00:00
johndoe6345789
6e2f0c08c0
Created 11 packagerepo-specific workflow plugins: - auth_verify_jwt - JWT token verification - auth_check_scopes - Scope-based authorization - parse_path - URL path parameter extraction (Express-style) - normalize_entity - Field normalization (trim, lower, unique, sort) - validate_entity - JSON schema validation - kv_get/kv_put - RocksDB key-value operations - blob_put - Filesystem blob storage with SHA-256 hashing - index_upsert - Index entry management - respond_json/respond_error - Response formatting Created string.sha256 plugin: - Compute SHA256 hash of strings/bytes - Optional "sha256:" prefix - Used by packagerepo for content-addressed storage All plugins follow standard pattern: - Class extending NodeExecutor - Factory with create() function - package.json with metadata - Access external state via runtime parameter Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
"""Workflow plugin: parse URL path with Express-style parameters."""
|
|
|
|
import re
|
|
from typing import Dict, Any
|
|
|
|
from ...base import NodeExecutor
|
|
|
|
|
|
class ParsePath(NodeExecutor):
|
|
"""Parse URL path with Express-style :param patterns."""
|
|
|
|
node_type = "packagerepo.parse_path"
|
|
category = "packagerepo"
|
|
description = "Parse URL path with Express-style :param patterns"
|
|
|
|
def execute(self, inputs: Dict[str, Any], runtime: Any = None) -> Dict[str, Any]:
|
|
"""Parse URL path against pattern."""
|
|
path = inputs.get("path")
|
|
pattern = inputs.get("pattern")
|
|
|
|
if not path:
|
|
return {"error": "path is required"}
|
|
|
|
if not pattern:
|
|
return {"error": "pattern is required"}
|
|
|
|
# Convert Express-style pattern to regex
|
|
# Example: /packages/:owner/:name -> /packages/(?P<owner>[^/]+)/(?P<name>[^/]+)
|
|
regex_pattern = pattern
|
|
|
|
# Replace :param with named regex groups
|
|
regex_pattern = re.sub(r':([a-zA-Z_][a-zA-Z0-9_]*)', r'(?P<\1>[^/]+)', regex_pattern)
|
|
|
|
# Escape forward slashes and add anchors
|
|
regex_pattern = f'^{regex_pattern}$'
|
|
|
|
try:
|
|
match = re.match(regex_pattern, path)
|
|
if match:
|
|
params = match.groupdict()
|
|
return {"result": {"params": params, "matched": True}}
|
|
else:
|
|
return {"result": {"params": {}, "matched": False}}
|
|
except re.error as e:
|
|
return {"error": f"invalid pattern: {str(e)}", "error_code": "INVALID_PATTERN"}
|