Dynamic API Implementation

1
# Workflows define their metadata directly in the class
2
@workflow.defn
3
class BackupWorkflow(WorkflowMetadataMixin, StageMixin, DeviceMixin, ArchiveMixin):
4
    """Network device configuration backup workflow."""
5
6
    # Workflow metadata
7
    workflow_description = "Backup network device configuration to Config Store"
8
    workflow_input_class = BackupInput
9
    workflow_api_endpoint = "/ngc/backup"
10
    workflow_namespace = "ngc"
11
12
    @workflow.run
13
    async def run(self, input_data: BackupInput) -> str:
14
        # Workflow implementation
15
        pass

1
# In dynamic_endpoints.py - Automatic discovery and registration
2
def register_dynamic_endpoints(router: APIRouter) -> None:
3
    """Register all workflow endpoints dynamically."""
4
    for workflow_class in get_registered_workflows():
5
        if hasattr(workflow_class, 'has_complete_metadata') and workflow_class.has_complete_metadata():
6
            endpoint_path = workflow_class.get_workflow_api_endpoint()
7
            input_class = workflow_class.get_workflow_input_class()
8
            endpoint_func = create_workflow_endpoint(workflow_class, input_class, endpoint_path)
9
            router.add_api_route(
10
                endpoint_path,
11
                endpoint_func,
12
                methods=["POST"],
13
                response_model=WorkflowResponse,
14
                name=f"{workflow_class.__name__.lower()}_endpoint"
15
            )

1
class WorkflowMetadataMixin:
2
    """Mixin for defining workflow metadata."""
3
4
    workflow_description: str = ""
5
    workflow_input_class: Type[BaseModel]
6
    workflow_api_endpoint: str = ""
7
    workflow_namespace: str = ""
8
    workflow_cli_name: Optional[str] = None
9
10
    @classmethod
11
    def get_workflow_description(cls) -> str:
12
        """Get the workflow description."""
13
        return cls.workflow_description or cls.__doc__.strip().split('\n')[0]
14
15
    @classmethod
16
    def has_complete_metadata(cls) -> bool:
17
        """Check if the workflow has complete metadata defined."""
18
        return bool(cls.workflow_description and cls.workflow_input_class and cls.workflow_api_endpoint)

1
def create_workflow_endpoint(workflow_class: Type, input_class: Type[BaseModel], endpoint_path: str):
2
    """Create a FastAPI endpoint function for a workflow."""
3
4
    async def workflow_endpoint(body: BaseModel, request: Request) -> WorkflowResponse:
5
        """Execute the workflow with provided parameters."""
6
        if start_workflow is None:
7
            raise RuntimeError("start_workflow function not set.")
8
9
        # Auto-populate user fields from request auth data
10
        user = getattr(request.state, 'user', 'unknown')
11
        if hasattr(body, 'user') and not body.user:
12
            body.user = user
13
        if hasattr(body, 'user_domain') and not body.user_domain:
14
            body.user_domain = user.split("@")[1] if "@" in user else "nvidia.com"
15
16
        workflow_id = await start_workflow(request, workflow_class, body)
17
        return WorkflowResponse(id=workflow_id)
18
19
    # Set function metadata for FastAPI
20
    workflow_endpoint.__name__ = f"{workflow_class.__name__.lower()}_endpoint"
21
    workflow_endpoint.__doc__ = f"Execute the {workflow_class.__name__} workflow."
22
    workflow_endpoint.__annotations__ = {
23
        'body': input_class,
24
        'request': Request,
25
        'return': WorkflowResponse
26
    }
27
28
    return workflow_endpoint

1
def register_dynamic_endpoints(router: APIRouter) -> None:
2
    """Register all workflow endpoints dynamically."""
3
    workflows = get_registered_workflows()
4
5
    for workflow_class in workflows:
6
        workflow_class = cast(Type[WorkflowMetadataMixin], workflow_class)
7
8
        if workflow_class.has_complete_metadata():
9
            endpoint_path = workflow_class.get_workflow_api_endpoint()
10
            input_class = workflow_class.get_workflow_input_class()
11
12
            endpoint_func = create_workflow_endpoint(workflow_class, input_class, endpoint_path)
13
14
            router.add_api_route(
15
                endpoint_path,
16
                endpoint_func,
17
                methods=["POST"],
18
                response_model=WorkflowResponse,
19
                name=f"{workflow_class.__name__.lower()}_endpoint",
20
                summary=workflow_class.get_workflow_description()
21
            )

1
from nv_config_manager.temporal.common.mixins.metadata import WorkflowMetadataMixin
2
3
@workflow.defn
4
class MyNewWorkflow(WorkflowMetadataMixin, StageMixin):
5
    """My new workflow for doing amazing things."""
6
7
    # Workflow metadata
8
    workflow_description = "Perform amazing network operations with validation"
9
    workflow_input_class = MyNewWorkflowInput
10
    workflow_api_endpoint = "/ngc/my_new_workflow"
11
    workflow_namespace = "ngc"
12
13
    @workflow.run
14
    async def run(self, input_data: MyNewWorkflowInput) -> str:
15
        # Workflow implementation
16
        pass

1
class MyNewWorkflowInput(BaseModel):
2
    """Input model for MyNewWorkflow."""
3
4
    device_id: str
5
    operation_type: str
6
    validate_before: bool = True

1
# In src/nv_config_manager/temporal/ngc/workflows/__init__.py
2
REGISTERED_WORKFLOWS = [
3
    # ... existing workflows
4
    MyNewWorkflow,
5
]

1
def test_workflow_metadata():
2
    """Test that workflow has complete metadata."""
3
    assert BackupWorkflow.has_complete_metadata()
4
    assert BackupWorkflow.get_workflow_description() == "Backup network device configuration to Config Store"
5
    assert BackupWorkflow.get_workflow_api_endpoint() == "/ngc/backup"
6
    assert BackupWorkflow.get_workflow_input_class() == BackupInput
7
8
def test_dynamic_endpoint_generation():
9
    """Test that endpoints are generated correctly."""
10
    router = APIRouter()
11
    register_dynamic_endpoints(router)
12
13
    # Check that backup endpoint was registered
14
    backup_route = next((route for route in router.routes if route.path == "/ngc/backup"), None)
15
    assert backup_route is not None
16
    assert "POST" in backup_route.methods

1
def test_backup_workflow_endpoint():
2
    """Test the dynamically generated backup endpoint."""
3
    response = client.post(
4
        "/v1/workflow/ngc/backup",
5
        json={"device_id": "test-device-id"},
6
        headers={"ssl-client-cert": test_cert}
7
    )
8
    assert response.status_code == 200
9
    assert "id" in response.json()
10
    assert "href" in response.json()