tessl/pypi-azure-mgmt-web

Microsoft Azure Web Apps Management Client Library for Python

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Workflow Management

Name: tessl/pypi-azure-mgmt-web
Author: tessl

Logic Apps workflow management including workflow definitions, runs, triggers, actions, and integration with other Azure services.

Package Information

Package: azure-mgmt-web
Modules: Multiple workflow-related operations classes
Access: client.workflows, client.workflow_runs, client.workflow_triggers, etc.

Core Imports

from azure.mgmt.web import WebSiteManagementClient
from azure.mgmt.web.models import (
    Workflow, WorkflowRun, WorkflowTrigger, WorkflowTriggerHistory,
    WorkflowVersion, WorkflowRunAction
)
from azure.identity import DefaultAzureCredential

Basic Usage

from azure.mgmt.web import WebSiteManagementClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = WebSiteManagementClient(credential, subscription_id)

# List workflows in a resource group
workflows = client.workflows.list(
    resource_group_name="my-resource-group",
    name="my-logic-app"
)

for workflow in workflows:
    print(f"Workflow: {workflow.name}, State: {workflow.state}")

# Get workflow details
workflow_details = client.workflows.get(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow"
)

Workflow Operations

Create or Update Workflow

Create a new workflow or update an existing one.

def create_or_update(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    workflow: Workflow,
    **kwargs
) -> Workflow:
    """
    Create or update a workflow.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        workflow: Workflow definition
        
    Returns:
        Workflow object
    """

Usage Example:

from azure.mgmt.web.models import Workflow

# Define workflow
workflow_definition = {
    "$schema": "https://schema.management.azure.com/providers/Microsoft.Logic/schemas/2016-06-01/workflowdefinition.json#",
    "contentVersion": "1.0.0.0",
    "triggers": {
        "manual": {
            "type": "Request",
            "kind": "Http",
            "inputs": {
                "schema": {}
            }
        }
    },
    "actions": {
        "Response": {
            "type": "Response",
            "inputs": {
                "statusCode": 200,
                "body": "Hello World"
            }
        }
    }
}

workflow = Workflow(
    location="East US",
    definition=workflow_definition,
    state="Enabled"
)

created_workflow = client.workflows.create_or_update(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    workflow=workflow
)

Get Workflow

Retrieve details for a specific workflow.

def get(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    **kwargs
) -> Workflow:
    """
    Get workflow details.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        
    Returns:
        Workflow object
    """

List Workflows

Get all workflows in a logic app.

def list(
    self,
    resource_group_name: str,
    name: str,
    **kwargs
) -> List[Workflow]:
    """
    List workflows in the logic app.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        
    Returns:
        List of Workflow objects
    """

Delete Workflow

Remove a workflow from the logic app.

def delete(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    **kwargs
) -> None:
    """
    Delete a workflow.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
    """

Workflow Run Management

List Workflow Runs

Get all runs for a specific workflow.

def list(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    **kwargs
) -> List[WorkflowRun]:
    """
    List workflow runs.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        
    Returns:
        List of WorkflowRun objects
    """

Get Workflow Run

Retrieve details for a specific workflow run.

def get(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    run_name: str,
    **kwargs
) -> WorkflowRun:
    """
    Get workflow run details.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        run_name: Name of the run
        
    Returns:
        WorkflowRun object
    """

Cancel Workflow Run

Stop a running workflow execution.

def cancel(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    run_name: str,
    **kwargs
) -> None:
    """
    Cancel a workflow run.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        run_name: Name of the run
    """

Usage Example:

# List recent workflow runs
runs = client.workflow_runs.list(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow"
)

for run in runs:
    print(f"Run: {run.name}, Status: {run.status}, Start Time: {run.start_time}")

# Get specific run details
run_details = client.workflow_runs.get(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    run_name="08586776228332053649"
)

# Cancel a running workflow
if run_details.status == "Running":
    client.workflow_runs.cancel(
        resource_group_name="my-resource-group",
        name="my-logic-app",
        workflow_name="my-workflow",
        run_name=run_details.name
    )

Workflow Run Actions

List Run Actions

Get all actions executed in a workflow run.

def list(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    run_name: str,
    **kwargs
) -> List[WorkflowRunAction]:
    """
    List actions in a workflow run.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        run_name: Name of the run
        
    Returns:
        List of WorkflowRunAction objects
    """

Get Run Action

Retrieve details for a specific action in a run.

def get(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    run_name: str,
    action_name: str,
    **kwargs
) -> WorkflowRunAction:
    """
    Get workflow run action details.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        run_name: Name of the run
        action_name: Name of the action
        
    Returns:
        WorkflowRunAction object
    """

Usage Example:

# List actions in a workflow run
actions = client.workflow_run_actions.list(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    run_name="08586776228332053649"
)

for action in actions:
    print(f"Action: {action.name}, Status: {action.status}")

# Get detailed action information
action_details = client.workflow_run_actions.get(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    run_name="08586776228332053649",
    action_name="HTTP_Action"
)
print(f"Action inputs: {action_details.inputs}")
print(f"Action outputs: {action_details.outputs}")

Workflow Triggers

List Triggers

Get all triggers defined for a workflow.

def list(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    **kwargs
) -> List[WorkflowTrigger]:
    """
    List workflow triggers.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        
    Returns:
        List of WorkflowTrigger objects
    """

Get Trigger

Retrieve details for a specific trigger.

def get(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    trigger_name: str,
    **kwargs
) -> WorkflowTrigger:
    """
    Get workflow trigger details.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        trigger_name: Name of the trigger
        
    Returns:
        WorkflowTrigger object
    """

Run Trigger

Manually execute a workflow trigger.

def run(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    trigger_name: str,
    **kwargs
) -> None:
    """
    Run a workflow trigger.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        trigger_name: Name of the trigger
    """

Get Trigger History

Retrieve execution history for a trigger.

def list(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    trigger_name: str,
    **kwargs
) -> List[WorkflowTriggerHistory]:
    """
    List workflow trigger history.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        trigger_name: Name of the trigger
        
    Returns:
        List of WorkflowTriggerHistory objects
    """

Usage Example:

# List triggers in a workflow
triggers = client.workflow_triggers.list(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow"
)

for trigger in triggers:
    print(f"Trigger: {trigger.name}, State: {trigger.state}")

# Run a trigger manually
client.workflow_triggers.run(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    trigger_name="manual"
)

# Get trigger execution history
trigger_history = client.workflow_trigger_histories.list(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    trigger_name="manual"
)

for history in trigger_history:
    print(f"Execution: {history.name}, Status: {history.status}, Start Time: {history.start_time}")

Workflow Versions

List Versions

Get all versions of a workflow.

def list(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    **kwargs
) -> List[WorkflowVersion]:
    """
    List workflow versions.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        
    Returns:
        List of WorkflowVersion objects
    """

Get Version

Retrieve details for a specific workflow version.

def get(
    self,
    resource_group_name: str,
    name: str,
    workflow_name: str,
    version_id: str,
    **kwargs
) -> WorkflowVersion:
    """
    Get workflow version details.
    
    Args:
        resource_group_name: Name of the resource group
        name: Name of the logic app
        workflow_name: Name of the workflow
        version_id: Version ID
        
    Returns:
        WorkflowVersion object
    """

Usage Example:

# List workflow versions
versions = client.workflow_versions.list(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow"
)

for version in versions:
    print(f"Version: {version.name}, Created: {version.created_time}")

# Get specific version details
version_details = client.workflow_versions.get(
    resource_group_name="my-resource-group",
    name="my-logic-app",
    workflow_name="my-workflow",
    version_id="08586776228332053649"
)

Types

Workflow

class Workflow:
    """Represents a Logic Apps workflow."""
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    location: Optional[str]
    tags: Optional[Dict[str, str]]
    state: Optional[str]  # Enabled, Disabled, Deleted, Suspended
    created_time: Optional[datetime]
    changed_time: Optional[datetime]
    version: Optional[str]
    access_endpoint: Optional[str]
    definition: Optional[Dict[str, Any]]
    parameters: Optional[Dict[str, WorkflowParameter]]

WorkflowRun

class WorkflowRun:
    """Represents a workflow execution run."""
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    start_time: Optional[datetime]
    end_time: Optional[datetime]
    status: Optional[str]  # NotSpecified, Paused, Running, Waiting, Succeeded, Skipped, Suspended, Cancelled, Failed, Faulted, TimedOut, Aborted, Ignored
    code: Optional[str]
    error: Optional[Dict[str, Any]]
    correlation_id: Optional[str]
    workflow: Optional[ResourceReference]
    trigger: Optional[WorkflowRunTrigger]
    outputs: Optional[Dict[str, WorkflowOutputParameter]]

WorkflowRunAction

class WorkflowRunAction:
    """Represents an action executed in a workflow run."""
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    start_time: Optional[datetime]
    end_time: Optional[datetime]
    status: Optional[str]
    code: Optional[str]
    error: Optional[Dict[str, Any]]
    tracking_id: Optional[str]
    correlation: Optional[RunActionCorrelation]
    inputs_link: Optional[ContentLink]
    outputs_link: Optional[ContentLink]
    tracked_properties: Optional[Dict[str, Any]]
    retry_history: Optional[List[RetryHistory]]

WorkflowTrigger

class WorkflowTrigger:
    """Represents a workflow trigger."""
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    provisioning_state: Optional[str]
    created_time: Optional[datetime]
    changed_time: Optional[datetime]
    state: Optional[str]  # NotSpecified, Completed, Enabled, Disabled, Deleted, Suspended
    status: Optional[str]
    last_execution_time: Optional[datetime]
    next_execution_time: Optional[datetime]
    recurrence: Optional[WorkflowTriggerRecurrence]
    workflow: Optional[ResourceReference]

WorkflowTriggerHistory

class WorkflowTriggerHistory:
    """Represents trigger execution history."""
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    start_time: Optional[datetime]
    end_time: Optional[datetime]
    scheduled_time: Optional[datetime]
    status: Optional[str]
    code: Optional[str]
    error: Optional[Dict[str, Any]]
    tracking_id: Optional[str]
    correlation: Optional[Correlation]
    inputs_link: Optional[ContentLink]
    outputs_link: Optional[ContentLink]
    fired: Optional[bool]
    run: Optional[ResourceReference]

Install with Tessl CLI