feat: implement user management functionality

docs/cli.md

@@ -23,6 +23,8 @@ $ pyaxm-cli [OPTIONS] COMMAND [ARGS]...
 * `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.
+* `users`: List all users in the organization.
+* `user`: Get a user by ID.
 
 ## `pyaxm-cli devices`
 
@@ -186,3 +188,35 @@ $ pyaxm-cli audit-events [OPTIONS] START_TIMESTAMP END_TIMESTAMP
 * `-f, --fields TEXT`
 * `-c, --cursor TEXT`
 * `--help`: Show this message and exit.
+
+## `pyaxm-cli users`
+
+List all users in the organization.
+
+**Usage**:
+
+```console
+$ pyaxm-cli users [OPTIONS]
+```
+
+**Options**:
+
+* `--help`: Show this message and exit.
+
+## `pyaxm-cli user`
+
+Get a user by ID.
+
+**Usage**:
+
+```console
+$ pyaxm-cli user [OPTIONS] USER_ID
+```
+
+**Arguments**:
+
+* `USER_ID`: [required]
+
+**Options**:
+
+* `--help`: Show this message and exit.

pyaxm/abm_requests.py

@@ -1,6 +1,6 @@
 import requests
 from http import HTTPStatus
-from typing import List
+from typing import List, Optional
 from pyaxm.models import (
     OrgDeviceResponse,
     MdmServersResponse,
@@ -11,6 +11,8 @@
     OrgDeviceActivityResponse,
     AppleCareCoverageResponse,
     AuditEventsResponse,
+    UsersResponse,
+    UserResponse,
 )
 import time
 from functools import wraps
@@ -39,6 +41,9 @@ def wrapper(*args, **kwargs):
 class DeviceError(Exception):
     pass
 
+class UserError(Exception):
+    pass
+
 class ABMRequests:
     def __init__(self):
         self.session = requests.Session()
@@ -326,3 +331,76 @@ def get_apple_care_coverage(self, device_id: str, access_token: str, fields: lis
             raise DeviceError(response.json()['errors'][0]['title'])
         else:
             response.raise_for_status()
+
+    @exponential_backoff(retries=5, backoff_factor=2)
+    def list_users(
+        self,
+        access_token: str,
+        limit: Optional[int] = None,
+        fields: Optional[List[str]] = None,
+        next: Optional[str] = None,
+    ) -> UsersResponse:
+        """
+        Get a list of users in an organization.
+
+        :param access_token: The access token for authentication.
+        :param limit: The number of resources to return. Maximum: 1000.
+        :param fields: Specific fields to return, e.g., ['firstName', 'lastName'].
+        :param next: Optional; the URL for the next page of results.
+        :return: A UsersResponse object containing the list of users.
+        """
+        if next:
+            url = next
+        else:
+            url = 'https://api-business.apple.com/v1/users'
+
+        params = {}
+        if limit:
+            params['limit'] = limit
+        if fields:
+            params['fields[users]'] = ','.join(fields)
+
+        response = self.session.get(
+            url,
+            headers=self._auth_headers(access_token),
+            params=None if next else params
+        )
+
+        if response.status_code == HTTPStatus.OK:
+            return UsersResponse.model_validate(response.json())
+        else:
+            response.raise_for_status()
+
+    @exponential_backoff(retries=5, backoff_factor=2)
+    def get_user(
+        self,
+        user_id: str,
+        access_token: str,
+        fields: Optional[List[str]] = None,
+    ) -> UserResponse:
+        """
+        Get information about a specific user.
+
+        :param user_id: The unique identifier for the user.
+        :param access_token: The access token for authentication.
+        :param fields: Specific fields to return, e.g., ['firstName', 'lastName'].
+        :return: A UserResponse object containing the user information.
+        """
+        url = f'https://api-business.apple.com/v1/users/{user_id}'
+
+        params = {}
+        if fields:
+            params['fields[users]'] = ','.join(fields)
+
+        response = self.session.get(
+            url,
+            headers=self._auth_headers(access_token),
+            params=params
+        )
+
+        if response.status_code == HTTPStatus.OK:
+            return UserResponse.model_validate(response.json())
+        elif response.status_code == HTTPStatus.NOT_FOUND:
+            raise UserError(response.json()['errors'][0]['title'])
+        else:
+            response.raise_for_status()

pyaxm/cli.py

@@ -130,5 +130,28 @@ def audit_events(
     df = pd.DataFrame(events_data)
     df.to_csv(sys.stdout, index=False)
 
+@app.command()
+def users():
+    """List all users in the organization."""
+    client = Client()
+    users = client.list_users()
+    users_data = []
+    for user in users:
+        user_info = {'id': user.id}
+        user_info.update(user.attributes.model_dump())
+        users_data.append(user_info)
+    df = pd.DataFrame(users_data)
+    df.to_csv(sys.stdout, index=False)
+
+@app.command()
+def user(user_id: Annotated[str, typer.Argument()]):
+    """Get a user by ID."""
+    client = Client()
+    user = client.get_user(user_id)
+    user_info = {'id': user.id}
+    user_info.update(user.attributes.model_dump())
+    df = pd.DataFrame([user_info])
+    df.to_csv(sys.stdout, index=False)
+
 if __name__ == "__main__":
     app()

pyaxm/client.py

@@ -13,7 +13,8 @@
     MdmServerDevicesLinkagesResponse,
     OrgDeviceAssignedServerLinkageResponse,
     OrgDeviceActivity,
-    AuditEvent
+    AuditEvent,
+    User,
 )
 from typing import List, Optional
 from functools import wraps
@@ -197,3 +198,20 @@ def assign_unassign_device_to_mdm_server(
             device_ids, server_id, action, self.access_token.value
         )
         return self._wait_for_device_activity_completion(response.data.id)
+
+    @ensure_valid_token
+    def list_users(self) -> List[User]:
+        response = self.abm.list_users(self.access_token.value)
+        users = response.data
+
+        while response.links.next:
+            next_page = response.links.next
+            response = self.abm.list_users(self.access_token.value, next=next_page)
+            users.extend(response.data)
+
+        return users
+
+    @ensure_valid_token
+    def get_user(self, user_id: str) -> User:
+        response = self.abm.get_user(user_id, self.access_token.value)
+        return response.data

pyaxm/models.py

@@ -6,6 +6,8 @@
 AppleCareCoveragePaymentType: TypeAlias = str
 MdmServerStatus: TypeAlias = str
 MdmServerProductFamily: TypeAlias = str
+UserStatus: TypeAlias = str
+UserPhoneNumberType: TypeAlias = str
 
 class DocumentLinks(BaseModel):
     self: AnyHttpUrl
@@ -557,3 +559,46 @@ class AuditEventsResponse(BaseModel):
     data: List[AuditEvent]
     links: PagedDocumentLinks
     meta: Optional[PagingInformation] = None
+
+# User
+class UserPhoneNumber(BaseModel):
+    phoneNumber: Optional[str] = None
+    type: Optional[UserPhoneNumberType] = None
+
+class UserRoleOuMapping(BaseModel):
+    roleName: Optional[str] = None
+    ouId: Optional[str] = None
+
+class User(BaseModel):
+    class Attributes(BaseModel):
+        firstName: Optional[str] = None
+        middleName: Optional[str] = None
+        lastName: Optional[str] = None
+        status: Optional[UserStatus] = None
+        managedAppleAccount: Optional[str] = None
+        isExternalUser: Optional[bool] = None
+        roleOuList: Optional[List[UserRoleOuMapping]] = None
+        email: Optional[str] = None
+        employeeNumber: Optional[str] = None
+        costCenter: Optional[str] = None
+        division: Optional[str] = None
+        department: Optional[str] = None
+        jobTitle: Optional[str] = None
+        phoneNumbers: Optional[List[UserPhoneNumber]] = None
+        startDateTime: Optional[AwareDatetime] = None
+        createdDateTime: Optional[AwareDatetime] = None
+        updatedDateTime: Optional[AwareDatetime] = None
+
+    attributes: Optional[Attributes] = None
+    id: str
+    links: Optional[ResourceLinks] = None
+    type: Literal['users']
+
+class UserResponse(BaseModel):
+    data: User
+    links: DocumentLinks
+
+class UsersResponse(BaseModel):
+    data: List[User]
+    links: PagedDocumentLinks
+    meta: Optional[PagingInformation] = None