feat: add get audit events support

docs/cli.md

@@ -22,6 +22,7 @@ $ pyaxm-cli [OPTIONS] COMMAND [ARGS]...
 * `mdm-server-assigned`: Get the server assignment for a device.
 * `assign-device`: Assign one or more devices to an MDM server.
 * `unassign-device`: Unassign one or more devices from an MDM...
+* `audit-events`: Get a list of audit events.
 
 ## `pyaxm-cli devices`
 
@@ -160,3 +161,28 @@ $ pyaxm-cli unassign-device [OPTIONS] DEVICE_IDS... SERVER_ID
 **Options**:
 
 * `--help`: Show this message and exit.
+
+## `pyaxm-cli audit-events`
+
+Get a list of audit events.
+
+**Usage**:
+
+```console
+$ pyaxm-cli audit-events [OPTIONS] START_TIMESTAMP END_TIMESTAMP
+```
+
+**Arguments**:
+
+* `START_TIMESTAMP`: [required]
+* `END_TIMESTAMP`: [required]
+
+**Options**:
+
+* `-a, --actor-id TEXT`
+* `-s, --subject-id TEXT`
+* `-e, --event-type TEXT`
+* `-l, --limit INTEGER`
+* `-f, --fields TEXT`
+* `-c, --cursor TEXT`
+* `--help`: Show this message and exit.

pyaxm/abm_requests.py

@@ -10,6 +10,7 @@
     OrgDeviceActivityCreateRequest,
     OrgDeviceActivityResponse,
     AppleCareCoverageResponse,
+    AuditEventsResponse,
 )
 import time
 from functools import wraps
@@ -110,6 +111,68 @@ def get_device(self, device_id, access_token) -> OrgDeviceResponse:
         else:
             response.raise_for_status()
 
+    @exponential_backoff(retries=5, backoff_factor=2)
+    def get_audit_events(
+        self,
+        access_token: str,
+        start_timestamp: str,
+        end_timestamp: str,
+        actor_id: str = None,
+        subject_id: str = None,
+        event_type: str = None,
+        limit: int = None,
+        fields: List[str] = None,
+        cursor: str = None,
+        next: str = None,
+    ) -> AuditEventsResponse:
+        """
+        Get a list of audit events in an organization that satisfies the query criteria.
+
+        :param access_token: The access token for authentication.
+        :param start_timestamp: ISO8601 formatted start timestamp of query time range.
+        :param end_timestamp: ISO8601 formatted end timestamp of query time range.
+        :param actor_id: Id of actor of event.
+        :param subject_id: Id of subject of event.
+        :param event_type: Type of event.
+        :param limit: The number of included related resources to return. Maximum: 1000.
+        :param fields: Specific fields to return, e.g., ['eventDateTime', 'type'].
+        :param cursor: Opaque cursor for pagination.
+        :return: An AuditEventsResponse object containing the list of audit events.
+        """
+
+        if next:
+            url = next
+        else:
+            url = 'https://api-business.apple.com/v1/auditEvents'
+
+        params = {
+            'filter[startTimestamp]': start_timestamp,
+            'filter[endTimestamp]': end_timestamp,
+        }
+
+        if actor_id:
+            params['filter[actorId]'] = actor_id
+        if subject_id:
+            params['filter[subjectId]'] = subject_id
+        if event_type:
+            params['filter[type]'] = event_type
+        if limit:
+            params['limit'] = limit
+        if fields:
+            params['fields[auditEvents]'] = ','.join(fields)
+        if cursor:
+            params['cursor'] = cursor
+
+        response = self.session.get(
+            url,
+            headers=self._auth_headers(access_token),
+            params=None if next else params
+        )
+        if response.status_code == HTTPStatus.OK:
+            return AuditEventsResponse.model_validate(response.json())
+        else:
+            response.raise_for_status()
+
     @exponential_backoff(retries=5, backoff_factor=2)
     def list_mdm_servers(self, access_token) -> MdmServersResponse:
         """

pyaxm/cli.py

@@ -2,7 +2,7 @@
 import pandas as pd
 import typer
 from typing_extensions import Annotated
-from typing import List
+from typing import List, Optional
 from pyaxm.client import Client
 from pyaxm.utils import download_activity_csv
 
@@ -99,5 +99,36 @@ def unassign_device(device_ids: Annotated[List[str], typer.Argument()], server_i
     if file_path:
         typer.echo(f"Report downloaded successfully to: {file_path}")
 
+@app.command()
+def audit_events(
+    start_timestamp: Annotated[str, typer.Argument()],
+    end_timestamp: Annotated[str, typer.Argument()],
+    actor_id: Annotated[Optional[str], typer.Option("--actor-id", "-a")] = None,
+    subject_id: Annotated[Optional[str], typer.Option("--subject-id", "-s")] = None,
+    event_type: Annotated[Optional[str], typer.Option("--event-type", "-e")] = None,
+    limit: Annotated[Optional[int], typer.Option("--limit", "-l")] = None,
+    fields: Annotated[Optional[List[str]], typer.Option("--fields", "-f")] = None,
+    cursor: Annotated[Optional[str], typer.Option("--cursor", "-c")] = None,
+):
+    """Get a list of audit events."""
+    client = Client()
+    events = client.get_audit_events(
+        start_timestamp=start_timestamp,
+        end_timestamp=end_timestamp,
+        actor_id=actor_id,
+        subject_id=subject_id,
+        event_type=event_type,
+        limit=limit,
+        fields=fields,
+        cursor=cursor,
+    )
+    events_data = []
+    for event in events:
+        event_info = {'id': event.id}
+        event_info.update(event.attributes.model_dump())
+        events_data.append(event_info)
+    df = pd.DataFrame(events_data)
+    df.to_csv(sys.stdout, index=False)
+
 if __name__ == "__main__":
     app()

pyaxm/client.py

@@ -12,9 +12,10 @@
     MdmServer,
     MdmServerDevicesLinkagesResponse,
     OrgDeviceAssignedServerLinkageResponse,
-    OrgDeviceActivity
+    OrgDeviceActivity,
+    AuditEvent
 )
-from typing import List
+from typing import List, Optional
 from functools import wraps
 
 class AccessToken:
@@ -125,6 +126,43 @@ def list_mdm_servers(self) -> list[MdmServer]:
         response = self.abm.list_mdm_servers(self.access_token.value)
         return response.data
 
+    @ensure_valid_token
+    def get_audit_events(
+        self,
+        start_timestamp: str,
+        end_timestamp: str,
+        actor_id: Optional[str] = None,
+        subject_id: Optional[str] = None,
+        event_type: Optional[str] = None,
+        limit: Optional[int] = None,
+        fields: Optional[List[str]] = None,
+        cursor: Optional[str] = None,
+    ) -> List[AuditEvent]:
+        response = self.abm.get_audit_events(
+            self.access_token.value,
+            start_timestamp,
+            end_timestamp,
+            actor_id,
+            subject_id,
+            event_type,
+            limit,
+            fields,
+            cursor
+        )
+        events = response.data
+
+        while response.links.next:
+            next_page = response.links.next
+            response = self.abm.get_audit_events(
+                self.access_token.value,
+                start_timestamp,
+                end_timestamp,
+                next=next_page
+            )
+            events.extend(response.data)
+
+        return events
+
     @ensure_valid_token
     def list_devices_in_mdm_server(self, server_id: str) -> list[MdmServerDevicesLinkagesResponse.Data]:
         response = self.abm.list_devices_in_mdm_server(server_id, self.access_token.value)

pyaxm/models.py

@@ -233,3 +233,27 @@ class AppleCareCoverageResponse(BaseModel):
     data: List[AppleCareCoverage]
     links: DocumentLinks
     meta: Optional[PagingInformation] = None
+
+class AuditEvent(BaseModel):
+    class Attributes(BaseModel):
+        model_config = ConfigDict(extra='allow') # Allow extra fields for polymorphic attributes
+        eventDateTime: Optional[AwareDatetime] = None
+        type: Optional[str] = None # This is the event type like DEVICE_ADDED_TO_ORG
+        category: Optional[str] = None
+        actorType: Optional[str] = None
+        actorId: Optional[str] = None
+        actorName: Optional[str] = None
+        subjectType: Optional[str] = None
+        subjectId: Optional[str] = None
+        subjectName: Optional[str] = None
+        outcome: Optional[str] = None
+        groupId: Optional[str] = None
+
+    attributes: Optional[Attributes] = None
+    id: str
+    type: Literal['auditEvents']
+
+class AuditEventsResponse(BaseModel):
+    data: List[AuditEvent]
+    links: PagedDocumentLinks
+    meta: Optional[PagingInformation] = None