Policy Events Operations

def list_query_results_for_management_group(
    management_group_name: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_subscription(
    subscription_id: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_resource_group(
    subscription_id: str,
    resource_group_name: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_resource(
    resource_id: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_policy_set_definition(
    subscription_id: str,
    policy_set_definition_name: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_policy_definition(
    subscription_id: str,
    policy_definition_name: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_subscription_level_policy_assignment(
    subscription_id: str,
    policy_assignment_name: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

def list_query_results_for_resource_group_level_policy_assignment(
    subscription_id: str,
    resource_group_name: str,
    policy_assignment_name: str,
    query_options: Optional[QueryOptions] = None,
    **kwargs
) -> ItemPaged[PolicyEvent]

class PolicyEvent:
    additional_properties: Optional[Dict[str, Any]]
    odata_id: Optional[str]
    odata_context: Optional[str]
    timestamp: Optional[datetime.datetime]
    resource_id: Optional[str]
    policy_assignment_id: Optional[str]
    policy_definition_id: Optional[str]
    effective_parameters: Optional[str]
    is_compliant: Optional[bool]
    subscription_id: Optional[str]
    resource_type: Optional[str]
    resource_location: Optional[str]
    resource_group: Optional[str]
    resource_tags: Optional[str]
    policy_assignment_name: Optional[str]
    policy_assignment_owner: Optional[str]
    policy_assignment_parameters: Optional[str]
    policy_assignment_scope: Optional[str]
    policy_definition_name: Optional[str]
    policy_definition_action: Optional[str]
    policy_definition_category: Optional[str]
    policy_set_definition_id: Optional[str]
    policy_set_definition_name: Optional[str]
    policy_set_definition_owner: Optional[str]
    policy_set_definition_category: Optional[str]
    policy_set_definition_parameters: Optional[str]
    management_group_ids: Optional[str]
    policy_definition_reference_id: Optional[str]
    tenant_id: Optional[str]
    principal_oid: Optional[str]
    components: Optional[List[ComponentEventDetails]]

class ComponentEventDetails:
    id: Optional[str]
    type: Optional[str]
    name: Optional[str]
    timestamp: Optional[datetime.datetime]
    tenant_id: Optional[str]
    principal_oid: Optional[str]
    policy_definition_action: Optional[str]
    additional_properties: Optional[Dict[str, Any]]

class PolicyEventsQueryResults:
    odata_context: Optional[str]
    odata_count: Optional[int]
    odata_next_link: Optional[str]
    value: Optional[List[PolicyEvent]]

from datetime import datetime, timedelta

# Query policy events from the last 24 hours
yesterday = datetime.utcnow() - timedelta(days=1)
query_options = QueryOptions(
    filter=f"timestamp ge {yesterday.isoformat()}Z",
    top=50,
    orderby="timestamp desc"
)

events = client.policy_events.list_query_results_for_subscription(
    subscription_id=subscription_id,
    query_options=query_options
)

for event in events:
    print(f"Event: {event.timestamp} - {event.resource_id} - Compliant: {event.is_compliant}")

# Query events for a specific policy definition
policy_events = client.policy_events.list_query_results_for_policy_definition(
    subscription_id=subscription_id,
    policy_definition_name="my-policy-definition",
    query_options=QueryOptions(top=100)
)

for event in policy_events:
    print(f"Resource: {event.resource_id}")
    print(f"Action: {event.policy_definition_action}")
    print(f"Compliant: {event.is_compliant}")
    print(f"Timestamp: {event.timestamp}")
    print("---")

# Query only non-compliant events
query_options = QueryOptions(
    filter="isCompliant eq false",
    orderby="timestamp desc",
    select="timestamp,resourceId,policyDefinitionName,policyDefinitionAction"
)

non_compliant_events = client.policy_events.list_query_results_for_resource_group(
    subscription_id=subscription_id,
    resource_group_name="production-rg",
    query_options=query_options
)

print("Non-compliant policy events:")
for event in non_compliant_events:
    print(f"- {event.timestamp}: {event.resource_id} failed {event.policy_definition_name}")

# Handle pagination manually
events_iterator = client.policy_events.list_query_results_for_subscription(
    subscription_id=subscription_id,
    query_options=QueryOptions(top=10)  # Small page size for demonstration
)

page_count = 0
event_count = 0

for event in events_iterator:
    event_count += 1
    if event_count % 10 == 1:  # New page
        page_count += 1
        print(f"Page {page_count}:")
    
    print(f"  Event {event_count}: {event.resource_id}")
    
    # Break after processing a reasonable number
    if event_count >= 50:
        break