Private Networking Operations

class PrivateEndpointConnectionsOperations:
    """
    Operations for managing private endpoint connections to App Configuration stores.
    
    Private endpoint connections allow secure access to App Configuration stores from
    within Azure virtual networks, keeping traffic on the Microsoft backbone network.
    """

class PrivateLinkResourcesOperations:
    """
    Operations for managing private link resources associated with App Configuration stores.
    
    Private link resources define the target sub-resources that can be connected to
    via private endpoints, such as 'configurationStores' for App Configuration.
    """

def list_by_configuration_store(
    self, 
    resource_group_name: str, 
    config_store_name: str, 
    **kwargs: Any
) -> ItemPaged[PrivateEndpointConnection]:
    """
    Lists all private endpoint connections for an App Configuration store.
    
    Args:
        resource_group_name: The name of the resource group containing the store.
        config_store_name: The name of the configuration store.
        **kwargs: Additional keyword arguments for the request.
        
    Returns:
        ItemPaged[PrivateEndpointConnection]: A collection of private endpoint connections.
        
    Example:
        >>> connections = client.private_endpoint_connections.list_by_configuration_store(
        ...     "my-rg", "my-store"
        ... )
        >>> for conn in connections:
        ...     print(f"Connection: {conn.name}")
        ...     print(f"Status: {conn.private_link_service_connection_state.status}")
        ...     print(f"Private Endpoint: {conn.private_endpoint.id}")
    """

def get(
    self, 
    resource_group_name: str, 
    config_store_name: str, 
    private_endpoint_connection_name: str, 
    **kwargs: Any
) -> PrivateEndpointConnection:
    """
    Gets the details of a specific private endpoint connection.
    
    Args:
        resource_group_name: The name of the resource group containing the store.
        config_store_name: The name of the configuration store.
        private_endpoint_connection_name: The name of the private endpoint connection.
        **kwargs: Additional keyword arguments for the request.
        
    Returns:
        PrivateEndpointConnection: The private endpoint connection details.
        
    Example:
        >>> connection = client.private_endpoint_connections.get(
        ...     "my-rg", "my-store", "my-private-endpoint-connection"
        ... )
        >>> print(f"Connection state: {connection.private_link_service_connection_state.status}")
        >>> print(f"Description: {connection.private_link_service_connection_state.description}")
    """

def begin_create_or_update(
    self, 
    resource_group_name: str, 
    config_store_name: str, 
    private_endpoint_connection_name: str, 
    private_endpoint_connection: PrivateEndpointConnection, 
    **kwargs: Any
) -> LROPoller[PrivateEndpointConnection]:
    """
    Creates or updates a private endpoint connection for an App Configuration store.
    
    Args:
        resource_group_name: The name of the resource group containing the store.
        config_store_name: The name of the configuration store.
        private_endpoint_connection_name: The name for the private endpoint connection.
        private_endpoint_connection: The private endpoint connection parameters.
        **kwargs: Additional keyword arguments for the request.
        
    Returns:
        LROPoller[PrivateEndpointConnection]: A poller for the long-running operation.
        
    Example:
        >>> from azure.mgmt.appconfiguration.models import (
        ...     PrivateEndpointConnection, 
        ...     PrivateLinkServiceConnectionState,
        ...     ConnectionStatus
        ... )
        >>> connection_state = PrivateLinkServiceConnectionState(
        ...     status=ConnectionStatus.APPROVED,
        ...     description="Approved by administrator"
        ... )
        >>> connection = PrivateEndpointConnection(
        ...     private_link_service_connection_state=connection_state
        ... )
        >>> poller = client.private_endpoint_connections.begin_create_or_update(
        ...     "my-rg", "my-store", "my-connection", connection
        ... )
        >>> result = poller.result()
    """

def begin_delete(
    self, 
    resource_group_name: str, 
    config_store_name: str, 
    private_endpoint_connection_name: str, 
    **kwargs: Any
) -> LROPoller[None]:
    """
    Deletes a private endpoint connection from an App Configuration store.
    
    Args:
        resource_group_name: The name of the resource group containing the store.
        config_store_name: The name of the configuration store.
        private_endpoint_connection_name: The name of the private endpoint connection to delete.
        **kwargs: Additional keyword arguments for the request.
        
    Returns:
        LROPoller[None]: A poller for the long-running delete operation.
        
    Example:
        >>> delete_poller = client.private_endpoint_connections.begin_delete(
        ...     "my-rg", "my-store", "my-connection"
        ... )
        >>> delete_poller.wait()  # Wait for deletion to complete
        >>> print("Private endpoint connection deleted")
    """

def list_by_configuration_store(
    self, 
    resource_group_name: str, 
    config_store_name: str, 
    **kwargs: Any
) -> ItemPaged[PrivateLinkResource]:
    """
    Lists the private link resources supported by an App Configuration store.
    
    Args:
        resource_group_name: The name of the resource group containing the store.
        config_store_name: The name of the configuration store.
        **kwargs: Additional keyword arguments for the request.
        
    Returns:
        ItemPaged[PrivateLinkResource]: A collection of private link resources.
        
    Example:
        >>> resources = client.private_link_resources.list_by_configuration_store(
        ...     "my-rg", "my-store"
        ... )
        >>> for resource in resources:
        ...     print(f"Resource: {resource.group_id}")
        ...     print(f"Required members: {resource.required_members}")
        ...     print(f"Required zone names: {resource.required_zone_names}")
    """

def get(
    self, 
    resource_group_name: str, 
    config_store_name: str, 
    group_name: str, 
    **kwargs: Any
) -> PrivateLinkResource:
    """
    Gets the details of a private link resource for an App Configuration store.
    
    Args:
        resource_group_name: The name of the resource group containing the store.
        config_store_name: The name of the configuration store.
        group_name: The name of the private link resource group.
        **kwargs: Additional keyword arguments for the request.
        
    Returns:
        PrivateLinkResource: The private link resource details.
        
    Example:
        >>> resource = client.private_link_resources.get(
        ...     "my-rg", "my-store", "configurationStores"
        ... )
        >>> print(f"Group ID: {resource.group_id}")
        >>> print(f"Required members: {resource.required_members}")
    """

class PrivateEndpointConnection:
    """
    Represents a private endpoint connection to an App Configuration store.
    
    Attributes:
        id (str): The resource ID of the connection (read-only).
        name (str): The name of the connection (read-only).
        type (str): The type of the connection (read-only).
        private_endpoint (PrivateEndpoint): The private endpoint details.
        private_link_service_connection_state (PrivateLinkServiceConnectionState): 
            The connection state information.
        provisioning_state (str): The provisioning state (read-only).
    """

class PrivateEndpoint:
    """
    Represents a private endpoint resource.
    
    Attributes:
        id (str): The resource ID of the private endpoint (read-only).
    """

class PrivateLinkServiceConnectionState:
    """
    Represents the connection state of a private link service connection.
    
    Attributes:
        status (ConnectionStatus): The connection status.
        description (str): Description of the connection state.
        actions_required (ActionsRequired): Actions required for the connection.
    """
    
    def __init__(
        self,
        *,
        status: Optional[ConnectionStatus] = None,
        description: Optional[str] = None,
        actions_required: Optional[ActionsRequired] = None,
        **kwargs: Any
    ) -> None:
        """
        Initialize connection state.
        
        Args:
            status: The connection status (Pending, Approved, Rejected, Disconnected).
            description: Human-readable description of the state.
            actions_required: Actions required beyond basic workflow.
        """

class PrivateLinkResource:
    """
    Represents a private link resource that can be connected to via private endpoint.
    
    Attributes:
        id (str): The resource ID (read-only).
        name (str): The name of the resource (read-only). 
        type (str): The type of the resource (read-only).
        group_id (str): The private link resource group ID.
        required_members (List[str]): Required member names for the resource.
        required_zone_names (List[str]): Required DNS zone names for the resource.
    """

class ConnectionStatus:
    """
    Enumeration of private endpoint connection statuses.
    
    Values:
        PENDING: Connection is pending approval.
        APPROVED: Connection has been approved.
        REJECTED: Connection has been rejected.
        DISCONNECTED: Connection has been disconnected.
    """
    PENDING = "Pending"
    APPROVED = "Approved" 
    REJECTED = "Rejected"
    DISCONNECTED = "Disconnected"

from azure.mgmt.appconfiguration import AppConfigurationManagementClient
from azure.mgmt.appconfiguration.models import (
    PrivateEndpointConnection,
    PrivateLinkServiceConnectionState,
    ConnectionStatus,
    ActionsRequired
)
from azure.identity import DefaultAzureCredential

# Initialize client
credential = DefaultAzureCredential()
client = AppConfigurationManagementClient(credential, "subscription-id")

resource_group = "my-resource-group"
store_name = "my-config-store"

# Step 1: Check available private link resources
print("Checking available private link resources...")
link_resources = client.private_link_resources.list_by_configuration_store(
    resource_group, store_name
)

for resource in link_resources:
    print(f"Private Link Resource: {resource.group_id}")
    print(f"  Required Members: {resource.required_members}")
    print(f"  Required Zone Names: {resource.required_zone_names}")
    print("---")

# Step 2: Get specific resource details
config_resource = client.private_link_resources.get(
    resource_group, store_name, "configurationStores"
)
print(f"Configuration Stores resource members: {config_resource.required_members}")

def manage_private_endpoint_connection_lifecycle():
    """Complete lifecycle management of private endpoint connections."""
    
    connection_name = "my-app-private-endpoint"
    
    # Step 1: Create/Update private endpoint connection
    print("Creating private endpoint connection...")
    
    # Define connection state (initially pending)
    connection_state = PrivateLinkServiceConnectionState(
        status=ConnectionStatus.PENDING,
        description="Requesting private endpoint connection for secure access"
    )
    
    # Create connection object
    connection = PrivateEndpointConnection(
        private_link_service_connection_state=connection_state
    )
    
    # Start creation operation
    create_poller = client.private_endpoint_connections.begin_create_or_update(
        resource_group, 
        store_name, 
        connection_name,
        connection
    )
    
    # Wait for creation to complete
    created_connection = create_poller.result()
    print(f"Connection created: {created_connection.name}")
    print(f"Initial status: {created_connection.private_link_service_connection_state.status}")
    
    # Step 2: Approve the connection
    print("Approving private endpoint connection...")
    
    approved_state = PrivateLinkServiceConnectionState(
        status=ConnectionStatus.APPROVED,
        description="Connection approved by network administrator",
        actions_required=ActionsRequired.NONE
    )
    
    approved_connection = PrivateEndpointConnection(
        private_link_service_connection_state=approved_state
    )
    
    approve_poller = client.private_endpoint_connections.begin_create_or_update(
        resource_group,
        store_name,
        connection_name, 
        approved_connection
    )
    
    final_connection = approve_poller.result()
    print(f"Connection approved: {final_connection.private_link_service_connection_state.status}")
    
    # Step 3: Monitor connection status
    print("Monitoring connection status...")
    
    current_connection = client.private_endpoint_connections.get(
        resource_group, store_name, connection_name
    )
    
    print(f"Current Status: {current_connection.private_link_service_connection_state.status}")
    print(f"Description: {current_connection.private_link_service_connection_state.description}")
    print(f"Provisioning State: {current_connection.provisioning_state}")
    
    if current_connection.private_endpoint:
        print(f"Private Endpoint ID: {current_connection.private_endpoint.id}")
    
    return current_connection

# Execute the lifecycle management
connection = manage_private_endpoint_connection_lifecycle()

def manage_multiple_private_endpoints():
    """Manage multiple private endpoint connections for different environments."""
    
    environments = {
        "development": "dev-app-private-endpoint",
        "staging": "staging-app-private-endpoint", 
        "production": "prod-app-private-endpoint"
    }
    
    connections = {}
    
    for env, connection_name in environments.items():
        print(f"Setting up {env} private endpoint...")
        
        # Create environment-specific connection
        connection_state = PrivateLinkServiceConnectionState(
            status=ConnectionStatus.APPROVED,  # Pre-approve for automation
            description=f"Private endpoint for {env} environment"
        )
        
        connection = PrivateEndpointConnection(
            private_link_service_connection_state=connection_state
        )
        
        try:
            create_poller = client.private_endpoint_connections.begin_create_or_update(
                resource_group,
                store_name,
                connection_name, 
                connection
            )
            
            result = create_poller.result()
            connections[env] = result
            print(f"✓ {env} connection created successfully")
            
        except Exception as e:
            print(f"✗ Failed to create {env} connection: {e}")
            connections[env] = None
    
    # List all connections for verification
    print("\nAll private endpoint connections:")
    all_connections = client.private_endpoint_connections.list_by_configuration_store(
        resource_group, store_name
    )
    
    for conn in all_connections:
        env = next((k for k, v in environments.items() if v in conn.name), "unknown")
        status = conn.private_link_service_connection_state.status
        print(f"  {env}: {conn.name} - {status}")
    
    return connections

# Set up multiple environments
env_connections = manage_multiple_private_endpoints()

def demonstrate_connection_state_transitions():
    """Show different connection state transitions."""
    
    connection_name = "state-demo-private-endpoint"
    
    # Helper function to update connection state
    def update_connection_state(status, description, actions_required=None):
        state = PrivateLinkServiceConnectionState(
            status=status,
            description=description,
            actions_required=actions_required or ActionsRequired.NONE
        )
        
        connection = PrivateEndpointConnection(
            private_link_service_connection_state=state
        )
        
        poller = client.private_endpoint_connections.begin_create_or_update(
            resource_group, store_name, connection_name, connection
        )
        return poller.result()
    
    # State 1: Pending (initial request)
    print("1. Setting connection to Pending...")
    pending_conn = update_connection_state(
        ConnectionStatus.PENDING,
        "Initial connection request submitted"
    )
    print(f"   Status: {pending_conn.private_link_service_connection_state.status}")
    
    # State 2: Approved (admin approval)
    print("2. Approving connection...")
    approved_conn = update_connection_state(
        ConnectionStatus.APPROVED,
        "Connection reviewed and approved by security team"
    )
    print(f"   Status: {approved_conn.private_link_service_connection_state.status}")
    
    # State 3: Rejected (if needed)
    print("3. Demonstrating rejection...")
    rejected_conn = update_connection_state(
        ConnectionStatus.REJECTED,
        "Connection rejected due to security policy violation",
        ActionsRequired.RECREATE
    )
    print(f"   Status: {rejected_conn.private_link_service_connection_state.status}")
    print(f"   Actions Required: {rejected_conn.private_link_service_connection_state.actions_required}")
    
    # State 4: Back to Approved (after policy update)
    print("4. Re-approving after policy update...")
    final_conn = update_connection_state(
        ConnectionStatus.APPROVED,
        "Connection re-approved after security policy update"
    )
    print(f"   Status: {final_conn.private_link_service_connection_state.status}")
    
    return final_conn

# Demonstrate state transitions
final_connection = demonstrate_connection_state_transitions()

def audit_private_networking():
    """Audit private networking configuration for compliance."""
    
    print("=== Private Networking Audit Report ===")
    
    # Check private link resources
    print("\n1. Private Link Resources:")
    link_resources = client.private_link_resources.list_by_configuration_store(
        resource_group, store_name
    )
    
    resource_count = 0
    for resource in link_resources:
        resource_count += 1
        print(f"   Resource: {resource.group_id}")
        print(f"   Members: {', '.join(resource.required_members)}")
        if resource.required_zone_names:
            print(f"   DNS Zones: {', '.join(resource.required_zone_names)}")
    
    print(f"   Total Resources: {resource_count}")
    
    # Check private endpoint connections
    print("\n2. Private Endpoint Connections:")
    connections = client.private_endpoint_connections.list_by_configuration_store(
        resource_group, store_name
    )
    
    connection_stats = {
        "total": 0,
        "approved": 0,
        "pending": 0,
        "rejected": 0,
        "disconnected": 0
    }
    
    for conn in connections:
        connection_stats["total"] += 1
        status = conn.private_link_service_connection_state.status.lower()
        
        if status in connection_stats:
            connection_stats[status] += 1
        
        print(f"   Connection: {conn.name}")
        print(f"   Status: {conn.private_link_service_connection_state.status}")
        print(f"   Description: {conn.private_link_service_connection_state.description}")
        
        if conn.private_endpoint:
            print(f"   Private Endpoint: {conn.private_endpoint.id}")
        
        print("   ---")
    
    # Summary
    print("\n3. Summary:")
    print(f"   Private Link Resources: {resource_count}")
    print(f"   Total Connections: {connection_stats['total']}")
    print(f"   Approved: {connection_stats['approved']}")
    print(f"   Pending: {connection_stats['pending']}")
    print(f"   Rejected: {connection_stats['rejected']}")
    print(f"   Disconnected: {connection_stats['disconnected']}")
    
    # Security recommendations
    print("\n4. Security Recommendations:")
    if connection_stats["pending"] > 0:
        print(f"   ⚠ Review {connection_stats['pending']} pending connections")
    
    if connection_stats["rejected"] > 0:
        print(f"   ⚠ Investigate {connection_stats['rejected']} rejected connections")
    
    if connection_stats["total"] == 0:
        print("   ⚠ No private endpoints configured - consider private access")
    
    approved_ratio = connection_stats["approved"] / max(connection_stats["total"], 1)
    if approved_ratio < 0.8:
        print("   ⚠ Low approval ratio - review connection policies")
    
    return connection_stats

# Run audit
audit_results = audit_private_networking()

def cleanup_private_endpoints():
    """Clean up unused or rejected private endpoint connections."""
    
    print("Cleaning up private endpoint connections...")
    
    connections = client.private_endpoint_connections.list_by_configuration_store(
        resource_group, store_name
    )
    
    cleanup_candidates = []
    
    for conn in connections:
        status = conn.private_link_service_connection_state.status
        
        # Mark rejected or disconnected connections for cleanup
        if status in [ConnectionStatus.REJECTED, ConnectionStatus.DISCONNECTED]:
            cleanup_candidates.append({
                "name": conn.name,
                "status": status,
                "description": conn.private_link_service_connection_state.description
            })
    
    if not cleanup_candidates:
        print("No connections require cleanup")
        return
    
    print(f"Found {len(cleanup_candidates)} connections for cleanup:")
    for candidate in cleanup_candidates:
        print(f"  - {candidate['name']} ({candidate['status']})")
    
    # Confirm cleanup
    if len(cleanup_candidates) > 0:
        response = input(f"Delete {len(cleanup_candidates)} connections? (y/N): ")
        
        if response.lower() == 'y':
            for candidate in cleanup_candidates:
                try:
                    print(f"Deleting {candidate['name']}...")
                    delete_poller = client.private_endpoint_connections.begin_delete(
                        resource_group, store_name, candidate['name']
                    )
                    delete_poller.wait()
                    print(f"✓ Deleted {candidate['name']}")
                    
                except Exception as e:
                    print(f"✗ Failed to delete {candidate['name']}: {e}")
        else:
            print("Cleanup cancelled")

# Run cleanup
cleanup_private_endpoints()

from azure.mgmt.appconfiguration.aio import AppConfigurationManagementClient
from azure.identity.aio import DefaultAzureCredential
import asyncio

async def async_private_networking_operations():
    """Demonstrate asynchronous private networking operations."""
    
    credential = DefaultAzureCredential()
    
    async with AppConfigurationManagementClient(credential, "subscription-id") as client:
        # Get private link resources asynchronously
        link_resources = client.private_link_resources.list_by_configuration_store(
            resource_group, store_name
        )
        
        print("Private Link Resources:")
        async for resource in link_resources:
            print(f"  {resource.group_id}: {resource.required_members}")
        
        # Get all private endpoint connections
        connections = client.private_endpoint_connections.list_by_configuration_store(
            resource_group, store_name  
        )
        
        print("\nPrivate Endpoint Connections:")
        async for conn in connections:
            status = conn.private_link_service_connection_state.status
            print(f"  {conn.name}: {status}")

# Run async operations
asyncio.run(async_private_networking_operations())