[refactor] Migrate v9 client layer and enforce proper sync/async model separation

- Add dedicated v9 sync and async client sub-clients (user, group, typedef,
  token, workflow, asset, admin, audit, credential, file, oauth_client,
  open_lineage, query, role, search_log, sso, task, contract, impersonate)
- Remove Pydantic remnants from v9 client layer: eliminate AtlanObject imports,
  exclude_unset parameter, and .json(by_alias=True) serialization
- Use creator/updater naming convention for v9 client CRUD methods;
  keep create_connection/create_sso_group compound names unchanged
- Remove unnecessary create=creator backward-compat aliases from v9 models
  (query, user, open_lineage event/job/input_dataset/output_dataset)
- Separate async models into pyatlan_v9/model/aio/ matching legacy structure:
  AsyncUserResponse, AsyncGroupResponse, AsyncOAuthClientListResponse,
  AsyncCustomMetadataDict/Proxy/Request
- Remove mixed async code (_get_next_page_async, __aiter__) from sync
  UserResponse and GroupResponse
- Fix entity.py to import AsyncCustomMetadataProxy from v9 instead of legacy
- Update all v9 unit tests for client naming, model changes, and async separation

Made-with: Cursor

Author: Aryamanz29

pyatlan_v9/client/admin.py

@@ -0,0 +1,89 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2025 Atlan Pte. Ltd.
+
+from __future__ import annotations
+
+import msgspec
+
+from pyatlan.client.common import AdminGetAdminEvents, AdminGetKeycloakEvents, ApiCaller
+from pyatlan.errors import ErrorCode
+from pyatlan.validate import validate_arguments
+from pyatlan_v9.model.keycloak_events import (
+    AdminEvent,
+    AdminEventRequest,
+    AdminEventResponse,
+    KeycloakEvent,
+    KeycloakEventRequest,
+    KeycloakEventResponse,
+)
+
+
+class V9AdminClient:
+    """
+    This class can be used to retrieve keycloak and admin events. This class does not need to be instantiated
+    directly but can be obtained through the admin property of AtlanClient.
+    """
+
+    def __init__(self, client: ApiCaller):
+        if not isinstance(client, ApiCaller):
+            raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters(
+                "client", "ApiCaller"
+            )
+        self._client = client
+
+    @validate_arguments
+    def get_keycloak_events(
+        self, keycloak_request: KeycloakEventRequest
+    ) -> KeycloakEventResponse:
+        """
+        Retrieve all events, based on the supplied filters.
+
+        :param keycloak_request: details of the filters to apply when retrieving events
+        :returns: the events that match the supplied filters
+        :raises AtlanError: on any API communication issue
+        """
+        endpoint, query_params = AdminGetKeycloakEvents.prepare_request(
+            keycloak_request
+        )
+        raw_json = self._client._call_api(
+            endpoint,
+            query_params=query_params,
+        )
+        if raw_json:
+            events = msgspec.convert(raw_json, list[KeycloakEvent], strict=False)
+        else:
+            events = []
+
+        return KeycloakEventResponse(
+            client=self._client,
+            criteria=keycloak_request,
+            start=keycloak_request.offset or 0,
+            size=keycloak_request.size or 100,
+            events=events,
+        )
+
+    @validate_arguments
+    def get_admin_events(self, admin_request: AdminEventRequest) -> AdminEventResponse:
+        """
+        Retrieve admin events based on the supplied filters.
+
+        :param admin_request: details of the filters to apply when retrieving admin events
+        :returns: the admin events that match the supplied filters
+        :raises AtlanError: on any API communication issue
+        """
+        endpoint, query_params = AdminGetAdminEvents.prepare_request(admin_request)
+        raw_json = self._client._call_api(
+            endpoint, query_params=query_params
+        )
+        if raw_json:
+            events = msgspec.convert(raw_json, list[AdminEvent], strict=False)
+        else:
+            events = []
+
+        return AdminEventResponse(
+            client=self._client,
+            criteria=admin_request,
+            start=admin_request.offset or 0,
+            size=admin_request.size or 100,
+            events=events,
+        )

pyatlan_v9/client/aio/__init__.py

@@ -0,0 +1,9 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2025 Atlan Pte. Ltd.
+"""v9 async client wrappers."""
+
+from pyatlan_v9.client.aio.asset import V9AsyncAssetClient
+from pyatlan_v9.client.aio.atlan import AsyncAtlanClient
+from pyatlan_v9.client.aio.group import V9AsyncGroupClient
+
+__all__ = ["AsyncAtlanClient", "V9AsyncAssetClient", "V9AsyncGroupClient"]

pyatlan_v9/client/aio/admin.py

@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2025 Atlan Pte. Ltd.
+
+from __future__ import annotations
+
+import msgspec
+
+from pyatlan.client.common import (
+    AdminGetAdminEvents,
+    AdminGetKeycloakEvents,
+    AsyncApiCaller,
+)
+from pyatlan.errors import ErrorCode
+from pyatlan.validate import validate_arguments
+from pyatlan_v9.model.aio.keycloak_events import (
+    AsyncAdminEventResponse,
+    AsyncKeycloakEventResponse,
+)
+from pyatlan_v9.model.keycloak_events import (
+    AdminEvent,
+    AdminEventRequest,
+    KeycloakEvent,
+    KeycloakEventRequest,
+)
+
+
+class V9AsyncAdminClient:
+    """
+    Async version of AdminClient for retrieving keycloak and admin events. This class does not need to be instantiated
+    directly but can be obtained through the admin property of the async Atlan client.
+    """
+
+    def __init__(self, client: AsyncApiCaller):
+        if not isinstance(client, AsyncApiCaller):
+            raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters(
+                "client", "AsyncApiCaller"
+            )
+        self._client = client
+
+    @validate_arguments
+    async def get_keycloak_events(
+        self, keycloak_request: KeycloakEventRequest
+    ) -> AsyncKeycloakEventResponse:
+        """
+        Retrieve all events, based on the supplied filters.
+
+        :param keycloak_request: details of the filters to apply when retrieving events
+        :returns: the events that match the supplied filters
+        :raises AtlanError: on any API communication issue
+        """
+        endpoint, query_params = AdminGetKeycloakEvents.prepare_request(
+            keycloak_request
+        )
+        raw_json = await self._client._call_api(
+            endpoint,
+            query_params=query_params,
+        )
+        if raw_json:
+            events = msgspec.convert(raw_json, list[KeycloakEvent], strict=False)
+        else:
+            events = []
+
+        return AsyncKeycloakEventResponse(
+            client=self._client,
+            criteria=keycloak_request,
+            start=keycloak_request.offset or 0,
+            size=keycloak_request.size or 100,
+            events=events,
+        )
+
+    @validate_arguments
+    async def get_admin_events(
+        self, admin_request: AdminEventRequest
+    ) -> AsyncAdminEventResponse:
+        """
+        Retrieve admin events based on the supplied filters.
+
+        :param admin_request: details of the filters to apply when retrieving admin events
+        :returns: the admin events that match the supplied filters
+        :raises AtlanError: on any API communication issue
+        """
+        endpoint, query_params = AdminGetAdminEvents.prepare_request(admin_request)
+        raw_json = await self._client._call_api(
+            endpoint, query_params=query_params
+        )
+        if raw_json:
+            events = msgspec.convert(raw_json, list[AdminEvent], strict=False)
+        else:
+            events = []
+
+        return AsyncAdminEventResponse(
+            client=self._client,
+            criteria=admin_request,
+            start=admin_request.offset or 0,
+            size=admin_request.size or 100,
+            events=events,
+        )