Approval Workflows

litestar-flags integrates with litestar-workflows to provide human-in-the-loop approval workflows for feature flag governance.

This is ideal for enterprise environments where feature flag changes require:

• Manager approval before going live

• QA validation in staging

• Audit trails for compliance

• Scheduled rollouts with wait periods

Installation

Install the workflows extra:

uv (Recommended)

uv add litestar-flags[workflows]

pip

pip install litestar-flags[workflows]

Key Concepts

Change Request

A FlagChangeRequest encapsulates all information needed to request a change:

from litestar_flags.contrib.workflows import FlagChangeRequest, ChangeType

request = FlagChangeRequest(
    flag_key="new_checkout_flow",
    change_type=ChangeType.CREATE,
    requested_by="developer@example.com",
    reason="Launch new checkout experience",
    flag_data={
        "name": "New Checkout Flow",
        "description": "Enables the redesigned checkout",
        "default_enabled": False,
    },
)

Change Types

The following change types are supported:

• ChangeType.CREATE - Create a new feature flag

• ChangeType.UPDATE - Modify an existing flag

• ChangeType.DELETE - Remove a flag

• ChangeType.TOGGLE - Toggle a flag’s enabled state

• ChangeType.ROLLOUT - Update rollout percentage

Workflow Steps

The integration provides several pre-built workflow steps:

Step	Type	Purpose
ValidateFlagChangeStep	Machine	Validates the change request
ManagerApprovalStep	Human	Manager reviews and approves
QAValidationStep	Human	QA validates in staging
ApplyFlagChangeStep	Machine	Executes the flag change
RolloutStep	Machine	Increases rollout percentage
NotifyStakeholdersStep	Machine	Sends notifications

Flag Approval Workflow

The FlagApprovalWorkflow implements a standard approval chain:

1. Validate - Check the request is valid

2. Manager Approval - Wait for manager sign-off

3. QA Validation - Wait for QA to verify in staging

4. Apply Change - Execute the flag modification

5. Notify - Inform stakeholders

Basic Usage

from litestar import Litestar
from litestar_flags import FeatureFlagsPlugin, FeatureFlagsConfig
from litestar_flags.contrib.workflows import FlagApprovalWorkflow
from litestar_workflows import WorkflowPlugin, WorkflowRegistry, LocalExecutionEngine

# Set up workflow registry
registry = WorkflowRegistry()
registry.register(FlagApprovalWorkflow)

# Configure plugins
flags_plugin = FeatureFlagsPlugin(config=FeatureFlagsConfig())
workflow_plugin = WorkflowPlugin(registry=registry)

app = Litestar(
    route_handlers=[...],
    plugins=[flags_plugin, workflow_plugin],
)

Starting an Approval Workflow

from litestar_flags.contrib.workflows import FlagChangeRequest, ChangeType

# Create a change request
request = FlagChangeRequest(
    flag_key="new_feature",
    change_type=ChangeType.CREATE,
    requested_by="developer@example.com",
    reason="New feature for Q1 launch",
    flag_data={
        "name": "New Feature",
        "description": "Enables the new feature",
        "default_enabled": False,
        "tags": ["q1", "frontend"],
    },
)

# Start the workflow
engine = LocalExecutionEngine(registry=registry)
instance = await engine.start_workflow(
    "flag_approval",
    initial_data={"request": request.to_dict()},
)

print(f"Workflow started: {instance.id}")

Customizing the Workflow

You can customize the workflow when getting the definition:

# Without QA step (manager approval only)
definition = FlagApprovalWorkflow.get_definition(
    storage=storage,
    require_qa=False,
)

# Without notification
definition = FlagApprovalWorkflow.get_definition(
    storage=storage,
    notify_on_complete=False,
)

Scheduled Rollout Workflow

The ScheduledRolloutWorkflow implements gradual feature rollouts with configurable wait periods between stages.

Default Stages

• Initial: 5% of users

• Early: 25% of users

• Half: 50% of users

• Majority: 75% of users

• Full: 100% of users

Between each stage, the workflow waits for a configurable period to allow monitoring for issues.

Usage

from litestar_flags.contrib.workflows import ScheduledRolloutWorkflow

# Register the workflow
registry.register(ScheduledRolloutWorkflow)

# Start a rollout
instance = await engine.start_workflow(
    "scheduled_rollout",
    initial_data={"flag_key": "new_feature"},
)

Custom Stages and Timing

from litestar_flags.contrib.workflows import RolloutStage

# Custom stages with 2-hour delay between each
definition = ScheduledRolloutWorkflow.get_definition(
    storage=storage,
    stage_delay_minutes=120,
    stages=[
        RolloutStage.INITIAL,  # 5%
        RolloutStage.HALF,     # 50%
        RolloutStage.FULL,     # 100%
    ],
)

Custom Steps

You can create custom steps by extending the base classes:

Custom Machine Step

from litestar_workflows import BaseMachineStep, WorkflowContext
from litestar_flags.protocols import StorageBackend

class CheckMetricsStep(BaseMachineStep):
    """Check metrics before proceeding with rollout."""

    def __init__(self, metrics_client, storage: StorageBackend | None = None):
        super().__init__(
            name="check_metrics",
            description="Verify metrics are healthy before proceeding",
        )
        self.metrics_client = metrics_client
        self._storage = storage

    async def execute(self, context: WorkflowContext) -> dict:
        flag_key = context.get("flag_key")

        # Check error rates, latency, etc.
        metrics = await self.metrics_client.get_metrics(flag_key)

        if metrics.error_rate > 0.01:  # 1% error threshold
            return {
                "healthy": False,
                "reason": f"Error rate too high: {metrics.error_rate:.2%}",
            }

        context.set("metrics_healthy", True)
        return {"healthy": True, "metrics": metrics.to_dict()}

Custom Human Step

from litestar_workflows import BaseHumanStep, WorkflowContext

class SecurityReviewStep(BaseHumanStep):
    """Security team review for sensitive flags."""

    def __init__(self):
        form_schema = {
            "type": "object",
            "required": ["approved", "security_notes"],
            "properties": {
                "approved": {
                    "type": "boolean",
                    "title": "Security Approved",
                },
                "security_notes": {
                    "type": "string",
                    "title": "Security Notes",
                },
                "risk_level": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "title": "Risk Level",
                },
            },
        }
        super().__init__(
            name="security_review",
            title="Security Review Required",
            description="Security team reviews the flag change",
            form_schema=form_schema,
        )

    async def execute(self, context: WorkflowContext) -> dict:
        form_data = context.get("form_data", {})
        approved = form_data.get("approved", False)

        context.set("security_approved", approved)
        context.set("security_notes", form_data.get("security_notes", ""))

        return {
            "approved": approved,
            "risk_level": form_data.get("risk_level", "medium"),
        }

Custom Notifications

Override the NotifyStakeholdersStep to integrate with your notification system:

from litestar_flags.contrib.workflows import NotifyStakeholdersStep

class SlackNotifyStep(NotifyStakeholdersStep):
    """Send notifications via Slack."""

    def __init__(self, slack_client):
        super().__init__(notification_channels=["slack"])
        self.slack_client = slack_client

    async def send_notification(self, data: dict, channels: list[str]) -> None:
        message = (
            f":flag-checkered: Flag Change Applied\n"
            f"*Flag:* `{data['flag_key']}`\n"
            f"*Type:* {data['change_type']}\n"
            f"*Requested by:* {data['requested_by']}\n"
            f"*Approved by:* {data.get('manager_approver', 'N/A')}"
        )

        await self.slack_client.post_message(
            channel="#feature-flags",
            text=message,
        )

Building Custom Workflows

You can compose your own workflows using the provided steps:

from litestar_workflows import WorkflowDefinition, Edge
from litestar_flags.contrib.workflows import (
    ValidateFlagChangeStep,
    ManagerApprovalStep,
    ApplyFlagChangeStep,
)

class CustomFlagWorkflow:
    __workflow_name__ = "custom_flag_approval"
    __workflow_version__ = "1.0.0"

    @classmethod
    def get_definition(cls, storage):
        return WorkflowDefinition(
            name=cls.__workflow_name__,
            version=cls.__workflow_version__,
            description="Custom approval with security review",
            steps={
                "validate": ValidateFlagChangeStep(storage=storage),
                "security": SecurityReviewStep(),
                "manager": ManagerApprovalStep(),
                "apply": ApplyFlagChangeStep(storage=storage),
                "notify": SlackNotifyStep(slack_client),
            },
            edges=[
                Edge("validate", "security",
                     condition=lambda ctx: ctx.get("valid")),
                Edge("security", "manager",
                     condition=lambda ctx: ctx.get("security_approved")),
                Edge("manager", "apply",
                     condition=lambda ctx: ctx.get("manager_approved")),
                Edge("apply", "notify"),
            ],
            initial_step="validate",
            terminal_steps={"notify"},
        )

API Reference

Types

class litestar_flags.contrib.workflows.ChangeType

Bases: str, Enum

Type of flag change being requested.

CREATE = 'create'

UPDATE = 'update'

DELETE = 'delete'

TOGGLE = 'toggle'

ROLLOUT = 'rollout'

__new__(value)

class litestar_flags.contrib.workflows.RolloutStage

Bases: str, Enum

Stages for gradual rollout workflows.

INITIAL = 'initial'

EARLY = 'early'

HALF = 'half'

MAJORITY = 'majority'

FULL = 'full'

property percentage: int

Get the rollout percentage for this stage.

__new__(value)

class litestar_flags.contrib.workflows.FlagChangeRequest

Bases: object

Request for a feature flag change.

This dataclass encapsulates all information needed to request a change to a feature flag through an approval workflow.

flag_key

The unique key of the flag to change.

change_type

The type of change (create, update, delete, toggle, rollout).

requested_by

Email or ID of the person requesting the change.

flag_data

Data for the flag (for create/update operations).

reason

Business justification for the change.

environment

Target environment (e.g., “production”, “staging”).

rollout_percentage

Target percentage for rollout changes.

metadata

Additional metadata for the request.

Parameters:

• flag_key (str)

• change_type (ChangeType)

• requested_by (str)

• flag_data (dict[str, Any] | None)

• reason (str)

• environment (str)

• rollout_percentage (int | None)

• metadata (dict[str, Any])

flag_key: str

change_type: ChangeType

requested_by: str

flag_data: dict[str, Any] | None = None

reason: str = ''

environment: str = 'production'

rollout_percentage: int | None = None

metadata: dict[str, Any]

to_dict()

Convert to dictionary for workflow context storage.

Return type:

dict[str, Any]

classmethod from_dict(data)

Create from dictionary (from workflow context).

Parameters:

data (dict[str, Any])

Return type:

FlagChangeRequest

__init__(flag_key, change_type, requested_by, flag_data=None, reason='', environment='production', rollout_percentage=None, metadata=<factory>)

Parameters:

• flag_key (str)

• change_type (ChangeType)

• requested_by (str)

• flag_data (dict[str, Any] | None)

• reason (str)

• environment (str)

• rollout_percentage (int | None)

• metadata (dict[str, Any])

Return type:

None

Steps

class litestar_flags.contrib.workflows.ValidateFlagChangeStep

Bases: BaseMachineStep

Validate a flag change request.

This step performs validation on the incoming change request: - Verifies required fields are present - Checks flag key format - Validates rollout percentages - For updates/deletes, verifies the flag exists

Parameters:

• storage (StorageBackend | None)

• name (str)

• description (str)

__init__(storage=None, name='validate_flag_change', description='Validate the flag change request')

Initialize with optional storage backend.

Parameters:

• storage (StorageBackend | None) – Storage backend for flag lookups. If None, will be retrieved from workflow context metadata.

• name (str) – Step name.

• description (str) – Step description.

Return type:

None

async execute(context)

Execute validation.

Parameters:

context (WorkflowContext) – Workflow context containing the change request.

Return type:

dict[str, Any]

Returns:

Validation result with any error messages.

class litestar_flags.contrib.workflows.ManagerApprovalStep

Bases: BaseHumanStep

Human step requiring manager approval.

This step pauses the workflow until a manager reviews and approves or rejects the flag change request.

Parameters:

• approver_roles (list[str] | None)

• timeout_hours (int)

• name (str)

• title (str)

• description (str)

__init__(approver_roles=None, timeout_hours=72, name='manager_approval', title='Manager Approval Required', description='Manager reviews and approves the flag change')

Initialize manager approval step.

Parameters:

• approver_roles (list[str] | None) – List of roles that can approve (e.g., [“manager”, “lead”]).

• timeout_hours (int) – Hours before the approval request times out.

• name (str) – Step name.

• title (str) – Step title for display.

• description (str) – Step description.

Return type:

None

async execute(context)

Process the manager’s decision.

Parameters:

context (WorkflowContext) – Workflow context with form submission data.

Return type:

dict[str, Any]

Returns:

Approval result.

class litestar_flags.contrib.workflows.QAValidationStep

Bases: BaseHumanStep

Human step for QA validation.

This step allows QA team members to verify the flag change is ready for production.

Parameters:

• name (str)

• title (str)

• description (str)

__init__(name='qa_validation', title='QA Validation Required', description='QA validates the flag change in staging')

Initialize QA validation step.

Parameters:

• name (str) – Step name.

• title (str) – Step title for display.

• description (str) – Step description.

Return type:

None

async execute(context)

Process QA validation results.

Parameters:

context (WorkflowContext) – Workflow context with form submission data.

Return type:

dict[str, Any]

Returns:

Validation result.

class litestar_flags.contrib.workflows.ApplyFlagChangeStep

Bases: BaseMachineStep

Apply the approved flag change.

This step executes the actual flag modification once all approvals have been obtained.

Parameters:

• storage (StorageBackend | None)

• name (str)

• description (str)

__init__(storage=None, name='apply_flag_change', description='Apply the approved flag change to the storage backend')

Initialize with optional storage backend.

Parameters:

• storage (StorageBackend | None) – Storage backend for flag operations. If None, will be retrieved from workflow context metadata.

• name (str) – Step name.

• description (str) – Step description.

Return type:

None

async execute(context)

Apply the flag change.

Parameters:

context (WorkflowContext) – Workflow context with validated request.

Return type:

dict[str, Any]

Returns:

Result of the flag operation.

class litestar_flags.contrib.workflows.RolloutStep

Bases: BaseMachineStep

Execute a rollout stage increase.

This step is used in gradual rollout workflows to increase the rollout percentage in stages.

Parameters:

• target_stage (RolloutStage)

• storage (StorageBackend | None)

• name (str | None)

• description (str | None)

__init__(target_stage, storage=None, name=None, description=None)

Initialize rollout step.

Parameters:

• target_stage (RolloutStage) – The rollout stage to reach.

• storage (StorageBackend | None) – Storage backend for flag operations.

• name (str | None) – Step name (defaults to rollout_{stage}).

• description (str | None) – Step description.

Return type:

None

async execute(context)

Execute the rollout stage increase.

Parameters:

context (WorkflowContext) – Workflow context.

Return type:

dict[str, Any]

Returns:

Rollout result.

class litestar_flags.contrib.workflows.NotifyStakeholdersStep

Bases: BaseMachineStep

Send notifications to stakeholders.

This step notifies relevant parties about flag changes. Override the send_notification method to integrate with your notification system.

Parameters:

• notification_channels (list[str] | None)

• name (str)

• description (str)

__init__(notification_channels=None, name='notify_stakeholders', description='Notify stakeholders about the flag change')

Initialize notification step.

Parameters:

• notification_channels (list[str] | None) – Channels to notify (e.g., [“email”, “slack”]).

• name (str) – Step name.

• description (str) – Step description.

Return type:

None

async execute(context)

Send notifications.

Parameters:

context (WorkflowContext) – Workflow context.

Return type:

dict[str, Any]

Returns:

Notification result.

async send_notification(data, channels)

Send the notification.

Override this method to integrate with your notification system.

Parameters:

• data (dict[str, Any]) – Notification data.

• channels (list[str]) – Channels to send to.

Return type:

None

Workflows

class litestar_flags.contrib.workflows.FlagApprovalWorkflow

Bases: object

Workflow for approving feature flag changes.

This workflow implements a standard approval process: 1. Validate the change request 2. Manager approval (human step) 3. QA validation (human step) 4. Apply the change 5. Notify stakeholders

Example

Registering the workflow:

from litestar_workflows import WorkflowRegistry
from litestar_flags.contrib.workflows import FlagApprovalWorkflow

registry = WorkflowRegistry()
registry.register(FlagApprovalWorkflow)

Starting an approval workflow:

from litestar_flags.contrib.workflows import FlagChangeRequest, ChangeType

request = FlagChangeRequest(
    flag_key="new_checkout_flow",
    change_type=ChangeType.CREATE,
    requested_by="developer@example.com",
    reason="Launch new checkout experience",
    flag_data={
        "name": "New Checkout Flow",
        "description": "Enables the redesigned checkout",
        "default_enabled": False,
    },
)

instance = await engine.start_workflow(
    "flag_approval",
    initial_data={"request": request.to_dict()},
)

Parameters:

storage (StorageBackend | None)

__init__(storage=None)

Initialize the workflow.

Parameters:

storage (StorageBackend | None) – Storage backend for flag operations.

Return type:

None

classmethod get_definition(storage=None, require_qa=True, notify_on_complete=True)

Get the workflow definition.

Parameters:

• storage (StorageBackend | None) – Storage backend for flag operations.

• require_qa (bool) – Whether to require QA validation step.

• notify_on_complete (bool) – Whether to notify stakeholders on completion.

Return type:

WorkflowDefinition

Returns:

The workflow definition.

class litestar_flags.contrib.workflows.ScheduledRolloutWorkflow

Bases: object

Workflow for gradual feature flag rollouts.

This workflow implements a staged rollout process: 1. Start at 5% (INITIAL stage) 2. Wait, then increase to 25% (EARLY stage) 3. Wait, then increase to 50% (HALF stage) 4. Wait, then increase to 75% (MAJORITY stage) 5. Wait, then increase to 100% (FULL stage)

Between each stage, there’s a configurable wait period to monitor for issues before proceeding.

Example

Starting a rollout workflow:

instance = await engine.start_workflow(
    "scheduled_rollout",
    initial_data={
        "flag_key": "new_feature",
    },
)

classmethod get_definition(storage=None, stage_delay_minutes=60, stages=None)

Get the workflow definition.

Parameters:

• storage (StorageBackend | None) – Storage backend for flag operations.

• stage_delay_minutes (int) – Minutes to wait between rollout stages.

• stages (list[RolloutStage] | None) – List of rollout stages to execute. Defaults to all stages.

Return type:

WorkflowDefinition

Returns:

The workflow definition.