PixelClientAsync

class PixelClientAsync:
    """
    Asynchronous client for interacting with the Pixel API.

    This client provides methods for working with projects, data collections, images,
    rasters, and other resources in the Pixel system. It handles authentication,
    request management, and provides high-level operations for common tasks.
    """

    def __init__(
        self,
        url: str,
        keycloak_server_url: str | None = None,
        realm: str | None = None,
        client_id: str | None = None,
        username: str | None = None,
        password: str | None = None,
        api_key: str | None = None,
        timeout_pixel: httpx.Timeout = DEFAULT_PIXEL_TIMEOUT,
        timeout_upload: httpx.Timeout = DEFAULT_UPLOAD_TIMEOUT,
        upload_num_retries: int = DEFAULT_UPLOAD_RETRIES,
        upload_max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    ):
        """
        Initialize a new Pixel API client.

        The client authenticates either with Keycloak credentials
        (`keycloak_server_url`, `realm`, `client_id`, `username`, `password`) or
        with a Pixel api key (`api_key`). Provide one or the other, not both.

        Args:
            url: Base URL of the Pixel API.
            keycloak_server_url: URL of the Keycloak authentication server.
            realm: Keycloak realm name.
            client_id: Keycloak client ID.
            username: Username for authentication.
            password: Password for authentication.
            api_key: A Pixel api key, used instead of the Keycloak credentials.
            timeout_pixel: Optional timeout configuration for Pixel API requests.
            timeout_upload: Optional timeout configuration for file uploads. If not provided, defaults to the same as `timeout_pixel`.

        Raises:
            ValueError: If both an api key and Keycloak credentials are given, or if neither is.
        """
        self.url: str = url
        self._timeout_pixel = timeout_pixel
        self._timeout_upload = timeout_upload
        self.auth: httpx.Auth = self._build_auth(
            keycloak_server_url=keycloak_server_url,
            realm=realm,
            client_id=client_id,
            username=username,
            password=password,
            api_key=api_key,
        )
        self.pixel_client = httpx.AsyncClient(
            base_url=self.url, auth=self.auth, timeout=self._timeout_pixel
        )
        self._internal_client = httpx.AsyncClient(
            timeout=self._timeout_upload,  # Use a separate client for uploads
        )  # For s3 uploads
        self._semaphore: asyncio.Semaphore = asyncio.Semaphore(upload_max_concurrency)

        self._retry_policy = tenacity.AsyncRetrying(
            stop=tenacity.stop_after_attempt(upload_num_retries),
            wait=tenacity.wait_random_exponential(),
            before=tenacity.before_log(logger, logging.DEBUG),
            after=tenacity.after_log(logger, logging.DEBUG),
        )
        self._internal_client.request = self._retry_policy.wraps(
            self._internal_client.request
        )

    @staticmethod
    def _build_auth(
        keycloak_server_url: str | None,
        realm: str | None,
        client_id: str | None,
        username: str | None,
        password: str | None,
        api_key: str | None,
    ) -> httpx.Auth:
        """Pick the auth scheme from the credentials given.

        Returns:
            httpx.Auth: An ApiKeyAuth if an api key was given, otherwise a KeyCloakAuth.

        Raises:
            ValueError: If both an api key and Keycloak credentials are given, or if neither is.
        """
        keycloak_args = {
            "keycloak_server_url": keycloak_server_url,
            "realm": realm,
            "client_id": client_id,
            "username": username,
            "password": password,
        }
        given = [name for name, value in keycloak_args.items() if value is not None]
        if api_key is not None:
            if given:
                raise ValueError(
                    "Got both api_key and Keycloak credentials "
                    f"({', '.join(given)}). Provide one or the other."
                )
            return ApiKeyAuth(api_key)
        missing = [name for name, value in keycloak_args.items() if value is None]
        if missing:
            raise ValueError(
                "No api_key given, so Keycloak credentials are required. "
                f"Missing: {', '.join(missing)}."
            )
        return KeyCloakAuth(
            keycloak_server_url,  # type: ignore[arg-type]
            realm,  # type: ignore[arg-type]
            client_id,  # type: ignore[arg-type]
            username,  # type: ignore[arg-type]
            password,  # type: ignore[arg-type]
        )

    @classmethod
    def from_settings(
        cls, settings: PixelApiSettings, **kwargs: Unpack[PixelClientKwargs]
    ) -> Self:
        """
        Instantiate the client from settings.

        Args:
            settings: PixelApiSettings object containing API configuration.
            **kwargs: Additional keyword arguments to pass to the client constructor.

        Returns:
            PixelClientAsync: A new client instance configured with the provided settings.
        """
        if settings.PIXEL_API_KEY is not None:
            return cls(
                url=settings.PIXEL_API_URL,
                api_key=settings.PIXEL_API_KEY.get_secret_value(),
                **kwargs,
            )
        return cls(
            url=settings.PIXEL_API_URL,
            keycloak_server_url=settings.PIXEL_AUTH_SERVER_URL,
            client_id=settings.PIXEL_CLIENT_ID,
            username=settings.PIXEL_USERNAME,
            password=settings.PIXEL_PASSWORD.get_secret_value(),  # type: ignore[union-attr]
            realm=settings.PIXEL_REALM,
            **kwargs,
        )

    async def __aenter__(self):
        await self.pixel_client.__aenter__()
        await self._internal_client.__aenter__()
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        await self.pixel_client.__aexit__(exc_type, exc_value, traceback)
        await self._internal_client.__aexit__(exc_type, exc_value, traceback)

    async def _make_request(self, method: str, url: str, **kwargs) -> httpx.Response:
        """
        Make an HTTP request to the Pixel API with error handling.

        Args:
            method: The HTTP method to use (GET, POST, PUT, DELETE).
            url: The URL to request, relative to the base URL.
            **kwargs: Additional arguments to pass to the request method.

        Returns:
            httpx.Response: The HTTP response object.

        Raises:
            PixelBadRequestError: If the server returns a 4xx client error with JSON response.
            httpx.HTTPStatusError: For other HTTP errors.
        """
        sensitive_content = kwargs.pop("sensitive_content", False)
        headers = kwargs.pop("headers", {})
        if "X-Request-ID" not in headers:
            headers["X-Request-ID"] = str(uuid4())
        kwargs["headers"] = headers
        response = await self.pixel_client.request(method, url, **kwargs)
        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            logger.error(f"Error in request {method} {url}: {e}")
            if e.response.is_client_error:
                try:
                    response_json = e.response.json()
                except json.JSONDecodeError:
                    raise e  # If the response is not JSON, raise the original exception
                raise PixelBadRequestError(
                    response_json=response_json,
                    status_code=e.response.status_code,
                    method=method,
                    url=str(e.request.url),
                    request_id=e.response.headers.get("X-Request-ID"),
                    body=e.request.content.decode() if not sensitive_content else None,
                ) from e
            raise e
        return response

    async def _get(self, url: str, **kwargs) -> httpx.Response:
        """
        Make a GET request to the Pixel API.

        Args:
            url: The URL to request, relative to the base URL.
            **kwargs: Additional arguments to pass to the request method.

        Returns:
            httpx.Response: The HTTP response object.
        """
        return await self._make_request("GET", url, **kwargs)

    async def _post(self, url: str, **kwargs) -> httpx.Response:
        """
        Make a POST request to the Pixel API.

        Args:
            url: The URL to request, relative to the base URL.
            **kwargs: Additional arguments to pass to the request method.

        Returns:
            httpx.Response: The HTTP response object.
        """
        return await self._make_request("POST", url, **kwargs)

    async def _put(self, url: str, **kwargs) -> httpx.Response:
        """Make a PUT request to the Pixel API.

        Args:
            url: The URL to request, relative to the base URL.
            **kwargs: Additional arguments to pass to the request method.

        Returns:
            httpx.Response: The HTTP response object.
        """
        return await self._make_request("PUT", url, **kwargs)

    async def _patch(self, url: str, **kwargs) -> httpx.Response:
        """Make a PATCH request to the Pixel API.

        Args:
            url: The URL to request, relative to the base URL.
            **kwargs: Additional arguments to pass to the request method.

        Returns:
            httpx.Response: The HTTP response object.
        """
        return await self._make_request("PATCH", url, **kwargs)

    async def _delete(self, url: str, **kwargs) -> httpx.Response:
        """
        Make a DELETE request to the Pixel API.

        Args:
            url: The URL to request, relative to the base URL.
            **kwargs: Additional arguments to pass to the request method.

        Returns:
            httpx.Response: The HTTP response object.
        """
        return await self._make_request("DELETE", url, **kwargs)

    async def _paginate(
        self,
        url: str,
        page_size: int,
        method: str = "GET",
        use_body: bool = False,
        results_parser: Callable[[dict], list[dict]] | None = None,
        **kwargs,
    ) -> AsyncGenerator[dict, None]:
        """
        Internal helper function for paginating through API results.

        Args:
            url: The API endpoint URL to paginate through.
            page_size: Number of items to fetch per page.
            method: HTTP method to use for the request (default is "GET").
            use_body: Whether to use the request body for pagination parameters instead of query parameters.
            **kwargs: Additional parameters to pass to the _make_request method.
        Yields:
            dict: Individual items from the paginated results, one at a time.
        """
        params: dict = kwargs.pop("params", {})
        body: dict = kwargs.pop("json", {})
        offset: int = params.get("offset", 0) if not use_body else body.get("offset", 0)
        limit = page_size
        while True:
            new_body = {**body, "offset": offset, "limit": limit} if use_body else body
            new_params = (
                {**params, "offset": offset, "limit": limit} if not use_body else params
            )
            response = await self._make_request(
                method, url, params=new_params, json=new_body, **kwargs
            )
            data = response.json()
            if results_parser and isinstance(data, dict):
                data = results_parser(data)
            for item in data:
                yield item
            if (
                len(data) < page_size
            ):  # If the number of items is less than the page size, we are done
                break
            offset += limit

    async def me(self, extended: bool = False) -> dict:
        """
        Get information about the authenticated user.

        Args:
            extended: If True, also returns the projects you can access with your effective
                permissions on each, and your organisation-wide grants and permissions.

        Returns:
            dict: User information including username, email, and other profile details.

        Note:
            Cannot be called with an api key.
        """
        response = await self._get("/users/me", params={"extended": extended})
        return response.json()

    async def get_plugins(self) -> list[dict]:
        """
        Retrieve a list of available plugins for the pixel tenant.

        Possible plugins include:

        * `optimized_raster` - Optimize Raster capability
        * `image_service` - Image Service capability, dependent on the 'optimize_raster' plugin.

        Returns:
            list[dict]: A list of plugin objects with their details.
        """
        response = await self._get("/plugins/")
        return response.json()

    async def create_project(
        self,
        name: str,
        description: str,
        area_of_interest: Polygon,
        parent_project_id: int | None = None,
        tags: list[str] | None = None,
        groups: dict[int, RoleType] | None = None,
    ) -> dict:
        """
        Create a new project in the Pixel system.

        Args:
            name: The name of the project.
            description: A description of the project.
            area_of_interest: A GeoJSON polygon defining the project's geographic area of interest.
            parent_project_id: Optional ID of a parent project. If provided, this project will be created as a child of that project.
            tags: Optional list of tags to associate with the project.
            groups: Optional mapping of group ID to the role each group gets on the new project,
                e.g. `{3: "editor"}`. Written together with the project: either all succeed or
                nothing is created. You must be a group admin of every group listed.

        Returns:
            dict: The created project object.
        """
        body = {
            "name": name,
            "description": description,
            "area_of_interest": area_of_interest.__geo_interface__,
        }
        if parent_project_id:
            body["parent_project_id"] = parent_project_id
        if tags:
            body["tags"] = tags
        if groups:
            body["groups"] = [
                {"group_id": group_id, "role": role} for group_id, role in groups.items()
            ]
        response = await self._post("/projects/", json=body)
        return response.json()

    async def list_projects(
        self, params: ListParams | None = None, **kwargs
    ) -> list[dict]:
        """
        List projects with optional filtering.

        Args:
            params: Optional ListParams object containing filtering parameters such as offset, limit, search, etc.
            **kwargs: Alternative way to provide filtering parameters. Ignored if params is provided.

        Returns:
            list[dict]: A list of project objects matching the filter criteria.
        """
        if params and kwargs:
            logger.warning("Ignoring kwargs when params is provided")
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get("/projects/", params=params_dict)
        return response.json()

    async def list_deleted_projects(
        self, params: ListParams | None = None, **kwargs
    ) -> list[dict]:
        """
        List deleted projects in the Pixel system.

        Args:
            params: Optional ListParams object containing filtering parameters such as offset, limit, search, etc.
            **kwargs: Alternative way to provide filtering parameters. Ignored if params is provided.

        Returns:
            list[dict]: A list of deleted project objects.
        """
        if params and kwargs:
            logger.warning("Ignoring kwargs when params is provided")
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get("/projects/deleted", params=params_dict)
        return response.json()

    async def get_project(self, project_id: int, extended: bool = False) -> dict:
        """
        Retrieve a project by its ID.

        Args:
            project_id: The ID of the project to retrieve.
            extended: If True, returns extended project information including child projects and data collections.

        Returns:
            dict: The project object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}", params={"extended": extended}
        )
        return response.json()

    async def update_project(
        self,
        project_id: int,
        name: str | None = None,
        description: str | None = None,
        area_of_interest: Polygon | None = None,
        add_tags: list[str] | None = None,
        remove_tags: list[str] | None = None,
    ) -> dict:
        """
        Update an existing project with new values.

        Args:
            project_id: The ID of the project to update.
            name: Optional new name for the project.
            description: Optional new description for the project.
            area_of_interest: Optional new GeoJSON polygon defining the project's geographic area of interest.
            add_tags: Optional list of tags to add to the project.
            remove_tags: Optional list of tags to remove from the project.

        Returns:
            dict: The updated project object.

        Raises:
            ValueError: If no update parameters are provided.
        """
        body = {}
        for param_name, value in {
            "name": name,
            "description": description,
            "area_of_interest": area_of_interest.__geo_interface__
            if area_of_interest
            else None,
            "add_tags": add_tags,
            "remove_tags": remove_tags,
        }.items():
            if value is not None:
                body[param_name] = value
        if not body:
            raise ValueError("No update parameters provided")
        response = await self._put(f"/projects/{project_id}", json=body)
        return response.json()

    async def delete_project(self, project_id: int) -> dict:
        """
        Delete a project by its ID.

        Args:
            project_id: The ID of the project to delete.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(f"/projects/{project_id}")
        return response.json()

    async def restore_project(self, project_id: int) -> dict:
        """
        Restore a deleted project by its ID.

        Args:
            project_id: The ID of the project to restore.

        Returns:
            dict: The restored project object.
        """
        response = await self._post(f"/projects/{project_id}/restore")
        return response.json()

    async def move_project(
        self, project_id: int, new_parent_project_id: int | None = None
    ) -> dict:
        """
        Move a project to a new parent project or to the root level.

        Args:
            project_id: The ID of the project to move.
            new_parent_project_id: Optional ID of the new parent project. If None, the project will be moved to the root level.

        Returns:
            dict: The updated project object.
        """
        # If None is provided, set to -1 to move to root
        new_parent_project_id = new_parent_project_id or -1
        response = await self._put(
            f"/projects/{project_id}/move",
            params={"new_parent_id": new_parent_project_id},
        )
        return response.json()

    async def create_data_collection(
        self,
        project_id: int,
        name: str,
        description: str,
        data_collection_type: DataCollectionType,
        tags: list[str] | None = None,
        raster_info: RasterInfo | None = None,
    ) -> dict:
        """
        Create a new data collection within a project.

        Args:
            project_id: The ID of the project to create the data collection in.
            name: The name of the data collection.
            description: A description of the data collection.
            data_collection_type: The type of data collection (e.g., "image", "raster", "RGB", "DTM", "DSM").
            tags: Optional list of tags to associate with the data collection.
            raster_info: Optional RasterInfo object containing raster-specific configuration.

        Returns:
            dict: The created data collection object.
        """
        body: dict[str, Any] = {
            "name": name,
            "description": description,
            "data_collection_type": data_collection_type,
        }
        if tags:
            body["tags"] = tags
        if raster_info:
            body["raster_info"] = raster_info.model_dump(exclude_none=True, mode="json")
        response = await self._post(
            f"/projects/{project_id}/data_collections/", json=body
        )
        return response.json()

    async def list_data_collections(
        self, project_id: int, params: DataCollectionListParams | None = None, **kwargs
    ) -> list[dict]:
        """
        List data collections within a project with optional filtering.

        Args:
            project_id: The ID of the project to list data collections from.
            params: Optional DataCollectionListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters. Ignored if params is provided.

        Returns:
            list[dict]: A list of data collection objects matching the filter criteria.
        """
        if params and kwargs:
            logger.warning("Ignoring kwargs when params is provided")
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else DataCollectionListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/", params=params_dict
        )
        return response.json()

    async def list_deleted_data_collections(
        self, project_id: int, params: DataCollectionListParams | None = None, **kwargs
    ) -> list[dict]:
        """
        List deleted data collections within a project.

        Args:
            project_id: The ID of the project to list deleted data collections from.

        Returns:
            list[dict]: A list of deleted data collection objects.
        """
        if params and kwargs:
            logger.warning("Ignoring kwargs when params is provided")
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else DataCollectionListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/deleted", params=params_dict
        )
        return response.json()

    async def get_data_collection(
        self, project_id: int, data_collection_id: int
    ) -> dict:
        """
        Retrieve a data collection by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to retrieve.

        Returns:
            dict: The data collection object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}"
        )
        return response.json()

    async def update_data_collection(
        self,
        project_id: int,
        data_collection_id: int,
        name: str | None = None,
        description: str | None = None,
        add_tags: list[str] | None = None,
        remove_tags: list[str] | None = None,
    ) -> dict:
        """
        Update an existing data collection with new values.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to update.
            name: Optional new name for the data collection.
            description: Optional new description for the data collection.
            add_tags: Optional list of tags to add to the data collection.
            remove_tags: Optional list of tags to remove from the data collection.

        Returns:
            dict: The updated data collection object.

        Raises:
            ValueError: If no update parameters are provided.
        """
        body = {}
        for param_name, value in {
            "name": name,
            "description": description,
            "add_tags": add_tags,
            "remove_tags": remove_tags,
        }.items():
            if value is not None:
                body[param_name] = value
        if not body:
            raise ValueError("No update parameters provided")
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}", json=body
        )
        return response.json()

    async def move_data_collection(
        self, project_id: int, data_collection_id: int, new_project_id: int
    ) -> dict:
        """
        Move a data collection to a different project.

        Args:
            project_id: The ID of the project currently containing the data collection.
            data_collection_id: The ID of the data collection to move.
            new_project_id: The ID of the project to move the data collection to.

        Returns:
            dict: The updated data collection object.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/move",
            params={"new_project_id": new_project_id},
        )
        return response.json()

    async def delete_data_collection(
        self, project_id: int, data_collection_id: int
    ) -> dict:
        """
        Delete a data collection by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to delete.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}"
        )
        return response.json()

    async def restore_data_collection(
        self, project_id: int, data_collection_id: int
    ) -> dict:
        """
        Restore a deleted data collection by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to restore.

        Returns:
            dict: The restored data collection object.
        """
        response = await self._post(
            f"/projects/{project_id}/data_collections/{data_collection_id}/restore"
        )
        return response.json()

    async def get_images(
        self,
        project_id: int,
        data_collection_id: int,
        params: ListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """
        Retrieve images from a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the images.
            params: Optional ListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.

        Returns:
            list[dict]: A list of image objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json") if params else {}
        )
        params_dict.update(kwargs)
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/images/",
            params=params_dict,
        )
        return response.json()

    async def paginate_images(
        self,
        project_id: int,
        data_collection_id: int,
        page_size: int,
        params: ListParams | None = None,
        **kwargs,
    ) -> AsyncGenerator[dict, None]:
        """
        Paginate through images in a data collection with optional filtering.
        This method returns an async generator that yields images one at a time,
        automatically handling pagination in the background.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the images.
            page_size: The number of images to fetch per page.
            params: Optional ListParams object containing filtering parameters (excluding offset and limit).
            **kwargs: Alternative way to provide filtering parameters.

        Yields:
            dict: Image objects matching the filter criteria, one at a time.
        """
        params_dict = (
            params.model_dump(
                exclude_none=True, exclude={"offset", "limit"}, mode="json"
            )
            if params
            else {}
        )
        params_dict.update(kwargs)
        url = f"/projects/{project_id}/data_collections/{data_collection_id}/images/"
        async for item in self._paginate(url, page_size, params=params_dict):
            yield item

    async def get_image(
        self,
        project_id: int,
        data_collection_id: int,
        image_id: int,
    ) -> dict:
        """
        Retrieve a specific image by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the image.
            image_id: The ID of the image to retrieve.

        Returns:
            dict: The image object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/images/{image_id}"
        )
        return response.json()

    async def get_image_metadata(
        self, project_id: int, data_collection_id: int, image_id: int
    ) -> dict:
        """
        Retrieve metadata for a specific image by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the image.
            image_id: The ID of the image to retrieve metadata for.

        Returns:
            dict: The image metadata object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/images/{image_id}/metadata"
        )
        return response.json()

    async def update_image(
        self,
        project_id: int,
        data_collection_id: int,
        image_id: int,
        update_input: ImageUpdateInput,
    ) -> dict:
        """
        Update an existing image with new values.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the image.
            image_id: The ID of the image to update.
            update_input: ImageUpdateInput object containing the fields to update.

        Returns:
            dict: The updated image object.
        """

        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/images/{image_id}",
            json=update_input.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def move_image(
        self,
        project_id: int,
        data_collection_id: int,
        image_id: int,
        new_project_id: int,
        new_data_collection_id: int,
    ) -> dict:
        """
        Move an image to a different data collection within a project.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the image.
            image_id: The ID of the image to move.
            new_project_id: The ID of the project to move the image to.
            new_data_collection_id: The ID of the data collection to move the image to.
        Returns:
            dict: The updated image object.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/images/{image_id}/move",
            params={
                "new_project_id": new_project_id,
                "new_data_collection_id": new_data_collection_id,
            },
        )
        return response.json()

    async def delete_image(
        self,
        project_id: int,
        data_collection_id: int,
        image_id: int,
    ) -> dict:
        """
        Delete a specific image by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the image.
            image_id: The ID of the image to delete.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}/images/{image_id}"
        )
        return response.json()

    async def get_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        params: RasterListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """
        Retrieve rasters from a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the rasters.
            params: Optional ListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.

        Returns:
            dict: A list of raster objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json") if params else {}
        )
        params_dict.update(kwargs)
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/",
            params=params_dict,
        )
        return response.json()

    async def paginate_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        page_size: int,
        params: ListParams | None = None,
        **kwargs,
    ) -> AsyncGenerator[dict, None]:
        """
        Paginate through rasters in a data collection with optional filtering.
        This method returns an async generator that yields rasters one at a time,
        automatically handling pagination in the background.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the rasters.
            page_size: The number of rasters to fetch per page.
            params: Optional ListParams object containing filtering parameters (excluding offset and limit).
            **kwargs: Alternative way to provide filtering parameters.

        Yields:
            dict: Raster objects matching the filter criteria, one at a time.
        """
        params_dict = (
            params.model_dump(exclude_none=True, exclude={"offset", "limit"})
            if params
            else {}
        )
        params_dict.update(kwargs)
        url = f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/"
        async for item in self._paginate(url, page_size, params=params_dict):
            yield item

    async def get_raster(
        self,
        project_id: int,
        data_collection_id: int,
        raster_id: int,
    ) -> dict:
        """
        Retrieve a specific raster by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the raster.
            raster_id: The ID of the raster to retrieve.

        Returns:
            dict: The raster object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/{raster_id}"
        )
        return response.json()

    async def update_raster(
        self,
        project_id: int,
        data_collection_id: int,
        raster_id: int,
        update_input: RasterUpdateInput,
    ) -> dict:
        """
        Update an existing raster with new values.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the raster.
            raster_id: The ID of the raster to update.
            update_input: RasterUpdateInput object containing the fields to update.

        Returns:
            dict: The updated raster object.

        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/{raster_id}",
            json=update_input.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def move_raster(
        self,
        project_id: int,
        data_collection_id: int,
        raster_id: int,
        new_project_id: int,
        new_data_collection_id: int,
    ) -> dict:
        """
        Move a raster to a different data collection within a project.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the raster.
            raster_id: The ID of the raster to move.
            new_project_id: The ID of the project to move the raster to.
            new_data_collection_id: The ID of the data collection to move the raster to.
        Returns:
            dict: The updated raster object.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/{raster_id}/move",
            params={
                "new_project_id": new_project_id,
                "new_data_collection_id": new_data_collection_id,
            },
        )
        return response.json()

    async def delete_raster(
        self,
        project_id: int,
        data_collection_id: int,
        raster_id: int,
    ) -> dict:
        """
        Delete a specific raster by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the raster.
            raster_id: The ID of the raster to delete.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/{raster_id}"
        )
        return response.json()

    async def get_upload_jobs(
        self,
        project_id: int,
        data_collection_id: int,
        params: ListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """
        Retrieve upload jobs for a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to list upload jobs for.
            params: Optional ListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters. Ignored if params is provided.

        Returns:
            list[dict]: A list of upload job objects matching the filter criteria.
        """
        if params and kwargs:
            logger.warning("Ignoring kwargs when params is provided")
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/upload_jobs/",
            params=params_dict,
        )
        return response.json()

    async def _create_upload_jobs(
        self,
        project_id: int,
        data_collection_id: int,
        multipart: bool,
        files: list[dict],
    ) -> list[dict]:
        """
        Internal helper to create upload job objects in the API.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to create upload jobs for.
            multipart: If True, create multipart upload jobs for large files.
            files: List of file objects containing metadata for the upload jobs.

        Returns:
            list[dict]: The created upload job objects.
        """
        url = (
            f"/projects/{project_id}/data_collections/{data_collection_id}/upload_jobs/"
        )
        if multipart:
            url += "multipart"
        body = {"files": files}
        response = await self._post(url, json=body)
        return response.json()

    def _create_support_file_file(
        self,
        file_path: Path,
    ) -> dict:
        """
        Create a file object for a support file to be included in an upload job.

        Args:
            file_path: Path to the support file.

        Returns:
            dict: A file object containing the file name and MD5 hash.

        Raises:
            FileNotFoundError: If the file does not exist.
        """
        if not file_path.exists():
            raise FileNotFoundError(f"File {file_path} not found")
        return {
            "name": file_path.name,
            "md5": calculate_md5_base64_from_file(file_path),
        }

    def _create_single_part_upload_job_from_file(
        self,
        file_path: Path,
        support_files: list[Path] | None = None,
        metadata: dict | None = None,
    ) -> dict:
        """
        Create a file object for a single-part upload job.

        Args:
            file_path: Path to the file to upload.
            support_files: Optional list of support file paths to include with the upload.
            metadata: Optional metadata to associate with the file.

        Returns:
            dict: A file object containing the file name, MD5 hash, and optional metadata and support files.
        """
        body: dict[str, Any] = {
            "name": file_path.name,
            "md5": calculate_md5_base64_from_file(file_path),
        }
        if metadata:
            body["metadata"] = metadata
        if support_files:
            body["support_files"] = [
                self._create_support_file_file(f) for f in support_files
            ]
        return body

    def _create_multipart_upload_job_from_file(
        self,
        file_path: Path,
        part_size: int,
        support_files: list[Path] | None = None,
        metadata: dict | None = None,
    ) -> dict:
        """
        Create a file object for a multipart upload job.

        Args:
            file_path: Path to the file to upload.
            part_size: Size of each part in bytes.
            support_files: Optional list of support file paths to include with the upload.
            metadata: Optional metadata to associate with the file.

        Returns:
            dict: A file object containing the file name, parts information, and optional metadata and support files.
        """
        parts = []
        for part_number, (content_length, md5, _) in enumerate(
            iter_file_parts(file_path, part_size), start=1
        ):
            parts.append(
                {
                    "part_number": part_number,
                    "md5": md5,
                    "content_length": content_length,
                }
            )
        file_body = {
            "name": file_path.name,
            "parts": parts,
        }
        if metadata:
            file_body["metadata"] = metadata
        if support_files:
            file_body["support_files"] = [
                self._create_support_file_file(f) for f in support_files
            ]
        return file_body

    async def _upload_singlepart_to_s3(
        self,
        upload_job: dict,
        file_path: Path,
    ):
        """
        Upload a file to S3 using a single part upload.

        Args:
            upload_job: Upload job object containing the S3 URL and upload fields.
            file_path: Path to the file to upload.

        Returns:
            None: This method returns None on success.

        Raises:
            httpx.HTTPStatusError: If the upload fails.
        """
        async with self._semaphore:
            response = await self._internal_client.post(
                upload_job["url"],
                files={"file": file_path.read_bytes()},
                data=upload_job["upload_fields"],
            )
            response.raise_for_status()
        return None

    async def _upload_multipart_to_s3(
        self, upload_job: dict, file_path: Path, multipart_part_size: int
    ):
        parts = sorted(upload_job["upload_parts"], key=lambda p: p["part_number"])

        async def read_and_upload_single_part(part_index, part):
            async with self._semaphore:
                with open(file_path, "rb") as f:
                    f.seek(part_index * multipart_part_size)
                    part_data = f.read(multipart_part_size)

                response = await self._internal_client.put(
                    part["url"], content=part_data, headers=part["headers"]
                )
                response.raise_for_status()

        tasks = [read_and_upload_single_part(i, part) for i, part in enumerate(parts)]

        with silence_logger(logging.getLogger("httpx")):
            await tqdm_asyncio.gather(
                *tasks, total=len(tasks), desc=f"Uploading {file_path.name} to Pixel"
            )

        return None

    async def _upload_to_s3(
        self,
        file: Path,
        support_files: list[Path] | None,
        upload_job: dict,
        multipart: bool,
        multipart_part_size: int | None,
    ):
        """
        Upload a file and its support files to S3.

        Args:
            file: Path to the main file to upload.
            support_files: Optional list of support file paths to upload with the main file.
            upload_job: Upload job object containing the upload information.
            multipart: If True, use multipart upload for the main file.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            None: This method returns None on success.

        Raises:
            AssertionError: If multipart is True but multipart_part_size is not provided,
                           or if support_files are required but not provided.
        """
        logger.info(f"Uploading {file} to S3 with upload job {upload_job['id']}")
        if multipart:
            assert multipart_part_size, (
                "multipart_part_size must be provided for multipart upload"
            )
            await self._upload_multipart_to_s3(upload_job, file, multipart_part_size)
        else:
            await self._upload_singlepart_to_s3(upload_job, file)
        if upload_job.get("support_files"):
            assert support_files, "Support files must be provided for upload job"
            # Sort the support file by name
            support_files = sorted(support_files, key=lambda f: f.name)
            upload_job_support_files = sorted(
                upload_job["support_files"], key=lambda f: f["name"]
            )
            await asyncio.gather(
                *[
                    self._upload_singlepart_to_s3(uj_sp, sp_path)
                    for uj_sp, sp_path in zip(upload_job_support_files, support_files)
                ]
            )
        logger.info(f"Upload job {upload_job['id']} with {file} completed upload")
        return None

    async def _upload_file_finished(
        self,
        project_id: int,
        data_collection_id: int,
        upload_job_id: int,
    ) -> dict:
        """
        Signal to the API that a file upload has completed.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the upload job.
            upload_job_id: The ID of the upload job to mark as finished.

        Returns:
            dict: The response containing job information for the validation process.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/upload_jobs/{upload_job_id}/uploadFinished"
        )
        return response.json()

    async def _multiple_upload_file_finished(
        self,
        project_id: int,
        data_collection_id: int,
        upload_job_ids: list[int],
    ) -> dict:
        """
        Signal to the API that multiple file uploads have completed.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the upload jobs.
            upload_job_ids: List of upload job IDs to mark as finished.

        Returns:
            dict: The response containing group job information and updated upload jobs.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/upload_jobs/multiple_uploadFinished",
            json={"job_ids": upload_job_ids},
        )
        return response.json()

    async def _wait_for_multiple_upload_jobs(
        self,
        project_id: int,
        data_collection_id: int,
        group_job_id: int,
        upload_job_ids: list[int],
    ) -> tuple[list[dict], list[PixelUploadJobError]]:
        """
        Wait for multiple upload jobs to complete validation and processing.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the upload jobs.
            group_job_id: The ID of the group job managing the upload jobs.
            upload_job_ids: List of upload job IDs to wait for.

        Returns:
            tuple: A tuple containing:
                - list[dict]: List of successfully completed upload jobs.
                - list[PixelUploadJobError]: List of errors for failed upload jobs.
        """
        _ = await self.wait_for_group_job(group_job_id=group_job_id)
        upload_jobs = await self.get_upload_jobs(
            project_id,
            data_collection_id,
            ids=upload_job_ids,
            limit=len(upload_job_ids),
        )
        while any(
            uj["status"] in ("lobby", "validating", "submitted", "validated")
            for uj in upload_jobs
        ):
            upload_jobs = await self.get_upload_jobs(
                project_id,
                data_collection_id,
                ids=upload_job_ids,
                limit=len(upload_job_ids),
            )
            await asyncio.sleep(1)
        # Collect any errors
        errors = []
        finished_upload_jobs = []
        for uj in upload_jobs:
            try:
                if uj["status"] != "validated_and_moved":
                    raise PixelUploadJobError(
                        job_id=uj["id"], status=uj["status"], detail=uj["detail"]
                    )
            except PixelUploadJobError as e:
                errors.append(e)
            else:
                finished_upload_jobs.append(uj)
        return finished_upload_jobs, errors

    async def _wait_for_upload_job(
        self, project_id: int, data_collection_id: int, upload_job_id: int, job_id: int
    ) -> dict:
        """
        Wait for a single upload job to complete validation and processing.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the upload job.
            upload_job_id: The ID of the upload job to wait for.
            job_id: The ID of the job managing the upload job validation.

        Returns:
            dict: The completed upload job object.

        Raises:
            PixelUploadJobError: If the upload job fails validation or processing.
        """
        await self.wait_for_job(job_id)
        upload_jobs = await self.get_upload_jobs(
            project_id, data_collection_id, ids=[upload_job_id], limit=1
        )
        upload_job = upload_jobs[0]
        if upload_job["status"] == "validated_and_moved":
            return upload_job
        if upload_job["status"] != "validated_and_moved":
            raise PixelUploadJobError(
                job_id=upload_job_id,
                status=upload_job["status"],
                detail=upload_job["detail"],
            )
        return upload_job

    def _get_resource_func(
        self, resource_name: ResourceTypeName
    ) -> Callable[..., Awaitable[dict]]:
        """
        Helper method to get the appropriate resource retrieval function based on resource name.
        """
        return {
            "image": self.get_image,
            "raster": self.get_raster,
            "point_cloud": self.get_point_cloud,
        }[resource_name]

    def _list_resource_func(
        self, resource_name: ResourceTypeName
    ) -> Callable[..., Awaitable[list[dict]]]:
        """
        Helper method to get the appropriate resource listing function based on resource name.
        """
        return {
            "image": self.get_images,
            "raster": self.get_rasters,
            "point_cloud": self.get_point_clouds,
        }[resource_name]

    async def _resources_from_upload_jobs(
        self, project_id: int, data_collection_id: int, upload_jobs: list[dict]
    ) -> list[dict]:
        """
        Retrieve the created resources (images, rasters or point_clouds) from completed upload jobs.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the resources.
            upload_jobs: List of completed upload job objects.

        Returns:
            list[dict]: List of resource objects (images, rasters or point_clouds) created from the upload jobs.
        """
        resource_ids: defaultdict[ResourceTypeName, list[int]] = defaultdict(list)
        resources: defaultdict[ResourceTypeName, list[dict]] = defaultdict(list)
        # Chunk up the resources ids
        chunk_size = 50
        for upload_job in upload_jobs:
            resource_ids[upload_job["result"]["resource_name"]].append(
                upload_job["result"]["resource_id"]
            )
        for resource_name, ids in resource_ids.items():
            for chunk in chunks(ids, chunk_size):
                resources[resource_name].extend(
                    await self._list_resource_func(resource_name)(
                        project_id,
                        data_collection_id,
                        ids=",".join(map(str, chunk)),
                        limit=len(chunk),
                    )
                )
        all_resources = list(itertools.chain.from_iterable(resources.values()))
        return all_resources

    async def _resource_from_upload_job(
        self, project_id: int, data_collection_id: int, upload_job: dict
    ) -> dict:
        """
        Retrieve the created resource (image, raster or point_cloud) from a completed upload job.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the resource.
            upload_job: The completed upload job object.

        Returns:
            dict: The resource object (image, raster or point_cloud) created from the upload job.

        Raises:
            ValueError: If the resource type in the upload job result is unknown.
        """
        return await self._get_resource_func(upload_job["result"]["resource_name"])(
            project_id,
            data_collection_id,
            upload_job["result"]["resource_id"],
        )

    async def _upload_multiple_files(
        self,
        project_id: int,
        data_collection_id: int,
        files: list[PixelUploadFile],
        multipart: bool,
        multipart_part_size: int | None,
        raise_on_error: bool = False,
    ) -> tuple[list[dict], list[PixelUploadJobError]]:
        """
        Upload multiple files to a data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to upload to.
            files: List of PixelUploadFile objects containing the files to upload.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.
            raise_on_error: If True, raise a PixelMultipleUploadJobError if any uploads fail.

        Returns:
            tuple: A tuple containing:
                - list[dict]: List of created resource objects (image, raster or point_cloud).
                - list[PixelUploadJobError]: List of errors for failed uploads.

        Raises:
            ValueError: If any files have duplicate names.
            PixelMultipleUploadJobError: If raise_on_error is True and any uploads fail.
        """

        # Chunk the operation into 200 files at a time (maximum for API)
        all_resources: list[dict] = []
        all_errors: list[PixelUploadJobError] = []

        create_body_func = (
            partial(
                self._create_multipart_upload_job_from_file,
                part_size=multipart_part_size,  # type: ignore
            )
            if multipart
            else self._create_single_part_upload_job_from_file
        )
        # Assert unique file names
        file_names = [f.file.name for f in files]
        if len(file_names) != len(set(file_names)):
            raise ValueError("All files must have unique names")
        for files_chunk in chunks(files, 200):
            logger.info("Uploading chunk of %d files", len(files_chunk))
            file_bodies = [
                create_body_func(
                    file.file,
                    support_files=file.support_files,
                    metadata=file.metadata.model_dump(mode="json", exclude_none=True)
                    if file.metadata
                    else None,
                )
                for file in files_chunk
            ]
            upload_jobs = await self._create_upload_jobs(
                project_id, data_collection_id, multipart, file_bodies
            )
            # sort the upload jobs by the file name
            upload_jobs = sorted(upload_jobs, key=lambda uj: uj["name"])
            files = sorted(files, key=lambda f: f.file.name)
            await asyncio.gather(
                *[
                    self._upload_to_s3(
                        file.file,
                        file.support_files,
                        upload_job,
                        multipart,
                        multipart_part_size,
                    )
                    for upload_job, file in zip(upload_jobs, files)
                ]
            )
            upload_finished = await self._multiple_upload_file_finished(
                project_id, data_collection_id, [uj["id"] for uj in upload_jobs]
            )
            group_job = upload_finished["group_job"]
            upload_jobs = upload_finished["upload_jobs"]
            upload_jobs, errors = await self._wait_for_multiple_upload_jobs(
                project_id,
                data_collection_id,
                group_job["group_id"],
                [uj["id"] for uj in upload_jobs],
            )
            resources = await self._resources_from_upload_jobs(
                project_id, data_collection_id, upload_jobs
            )
            all_resources.extend(resources)
            all_errors.extend(errors)

        if raise_on_error and all_errors:
            raise PixelMultipleUploadJobError(errors)
        return all_resources, all_errors

    async def _upload_file(
        self,
        project_id: int,
        data_collection_id: int,
        file_path: Path,
        support_files: list[Path] | None = None,
        metadata: dict | None = None,
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> dict:
        """
        Upload a single file to a data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to upload to.
            file_path: Path to the file to upload.
            support_files: Optional list of support file paths to upload with the file.
            metadata: Optional dictionary of metadata to associate with the file.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            dict: The created resource object (image, raster or point_cloud).

        Raises:
            FileNotFoundError: If the file does not exist.
            ValueError: If multipart is True but multipart_part_size is not provided.
            PixelUploadJobError: If the upload job fails validation or processing.
        """
        if not file_path.exists():
            raise FileNotFoundError(f"File {file_path} not found")
        if multipart and not multipart_part_size:
            raise ValueError(
                "multipart_part_size must be provided for multipart upload"
            )
        if multipart:
            assert multipart_part_size
            file_body = self._create_multipart_upload_job_from_file(
                file_path,
                multipart_part_size,
                support_files=support_files,
                metadata=metadata,
            )
        else:
            file_body = self._create_single_part_upload_job_from_file(
                file_path,
                support_files=support_files,
                metadata=metadata,
            )
        upload_jobs = await self._create_upload_jobs(
            project_id, data_collection_id, multipart, [file_body]
        )
        upload_job = upload_jobs[0]
        await self._upload_to_s3(
            file_path,
            support_files,
            upload_job,
            multipart,
            multipart_part_size,
        )
        upload_job_finished = await self._upload_file_finished(
            project_id, data_collection_id, upload_job["id"]
        )
        job = upload_job_finished["job"]
        upload_job = await self._wait_for_upload_job(
            project_id, data_collection_id, upload_job["id"], job["job_id"]
        )
        return await self._resource_from_upload_job(
            project_id, data_collection_id, upload_job
        )

    async def upload_image(
        self,
        project_id: int,
        data_collection_id: int,
        file_path: Path,
        metadata: dict | None = None,
        support_files: list[Path] | None = None,
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> dict:
        """
        Upload an image file to an image data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the image data collection to upload to.
            file_path: Path to the image file to upload.
            metadata: Optional dictionary of metadata to associate with the image.
            support_files: Optional list of support file paths to upload with the image.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            dict: The created image object.

        Raises:
            ValueError: If the data collection is not an image data collection.
            FileNotFoundError: If the file does not exist.
            ValueError: If multipart is True but multipart_part_size is not provided.
        """
        dc = await self.get_data_collection(project_id, data_collection_id)
        if dc["data_collection_type"] != "image":
            raise ValueError("Data collection is not an image data collection")
        return await self._upload_file(
            project_id,
            data_collection_id,
            file_path,
            support_files=support_files,
            metadata=metadata,
            multipart=multipart,
            multipart_part_size=multipart_part_size,
        )

    async def upload_multiple_images(
        self,
        project_id: int,
        data_collection_id: int,
        files: list[PixelUploadFile],
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> tuple[list[dict], list[PixelUploadJobError]]:
        """
        Upload multiple image files to an image data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the image data collection to upload to.
            files: List of PixelUploadFile objects containing the files to upload.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            tuple: A tuple containing:
                - list[dict]: List of created image objects.
                - list[PixelUploadJobError]: List of errors that occurred during upload.

        Raises:
            ValueError: If the data collection is not an image data collection.
            ValueError: If any files have duplicate names.
            ValueError: If multipart is True but multipart_part_size is not provided.
        """
        dc = await self.get_data_collection(project_id, data_collection_id)
        if dc["data_collection_type"] != "image":
            raise ValueError("Data collection is not an image data collection")
        return await self._upload_multiple_files(
            project_id,
            data_collection_id,
            files,
            multipart,
            multipart_part_size,
        )

    async def upload_multiple_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        files: list[PixelUploadFile],
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> tuple[list[dict], list[PixelUploadJobError]]:
        """
        Upload multiple raster files to a raster data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the raster data collection to upload to.
            files: List of PixelUploadFile objects containing the files to upload.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            tuple: A tuple containing:
                - list[dict]: List of created raster objects.
                - list[PixelUploadJobError]: List of errors that occurred during upload.

        Raises:
            ValueError: If the data collection is not a raster data collection.
            ValueError: If any files have duplicate names.
            ValueError: If multipart is True but multipart_part_size is not provided.
        """
        dc = await self.get_data_collection(project_id, data_collection_id)
        if dc["data_collection_type"] not in ("raster", "RGB", "DTM", "DSM"):
            raise ValueError("Data collection is not a raster data collection")
        return await self._upload_multiple_files(
            project_id,
            data_collection_id,
            files,
            multipart,
            multipart_part_size,
        )

    async def upload_raster(
        self,
        project_id: int,
        data_collection_id: int,
        file_path: Path,
        support_files: list[Path] | None = None,
        metadata: dict | None = None,
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> dict:
        """
        Upload a raster file to a raster data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the raster data collection to upload to.
            file_path: Path to the raster file to upload.
            support_files: Optional list of support file paths to upload with the raster.
            metadata: Optional dictionary of metadata to associate with the raster.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            dict: The created raster object.

        Raises:
            ValueError: If the data collection is not a raster data collection.
            FileNotFoundError: If the file does not exist.
            ValueError: If multipart is True but multipart_part_size is not provided.
        """
        dc = await self.get_data_collection(project_id, data_collection_id)
        if dc["data_collection_type"] not in ("raster", "RGB", "DTM", "DSM"):
            raise ValueError("Data collection is not a raster data collection")
        return await self._upload_file(
            project_id,
            data_collection_id,
            file_path,
            support_files=support_files,
            metadata=metadata,
            multipart=multipart,
            multipart_part_size=multipart_part_size,
        )

    async def upload_multiple_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        files: list[PixelUploadFile],
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> tuple[list[dict], list[PixelUploadJobError]]:
        """
        Upload multiple point cloud files to a point cloud data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the point cloud data collection to upload to.
            files: List of PixelUploadFile objects containing the files to upload.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            tuple: A tuple containing:
                - list[dict]: List of created point cloud objects.
                - list[PixelUploadJobError]: List of errors that occurred during upload.

        Raises:
            ValueError: If the data collection is not a point cloud data collection.
            ValueError: If any files have duplicate names.
            ValueError: If multipart is True but multipart_part_size is not provided.
        """
        dc = await self.get_data_collection(project_id, data_collection_id)
        if dc["data_collection_type"] != "point_cloud":
            raise ValueError("Data collection is not a point cloud data collection")
        return await self._upload_multiple_files(
            project_id,
            data_collection_id,
            files,
            multipart,
            multipart_part_size,
        )

    async def upload_point_cloud(
        self,
        project_id: int,
        data_collection_id: int,
        file_path: Path,
        metadata: dict | None = None,
        multipart: bool = False,
        multipart_part_size: int | None = None,
    ) -> dict:
        """
        Upload a point cloud file to a point cloud data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the point cloud data collection to upload to.
            file_path: Path to the point cloud file to upload.
            metadata: Optional dictionary of metadata to associate with the point cloud.
            multipart: If True, use multipart upload for large files.
            multipart_part_size: Required size of each part in bytes when using multipart upload.

        Returns:
            dict: The created point cloud object.

        Raises:
            ValueError: If the data collection is not a point cloud data collection.
            FileNotFoundError: If the file does not exist.
            ValueError: If multipart is True but multipart_part_size is not provided.
        """
        dc = await self.get_data_collection(project_id, data_collection_id)
        if dc["data_collection_type"] != "point_cloud":
            raise ValueError("Data collection is not a point cloud data collection")
        return await self._upload_file(
            project_id,
            data_collection_id,
            file_path,
            metadata=metadata,
            multipart=multipart,
            multipart_part_size=multipart_part_size,
        )

    async def create_optimized_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        raster_ids: list[int] | None,
        profile: str | None = None,
        nearblack: NearblackOptions | None = None,
        overview_resampling: OverviewResampling = "average",
    ) -> list[dict]:
        """
        Create optimized raster objects in the database.
        This function creates optimized raster objects but does not run the actual optimization process.
        To run the optimization, use the run_optimize_rasters function with the returned optimized raster IDs.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the rasters.
            raster_ids: Optional list of raster IDs to optimize. If None, all rasters in the collection will be optimized.
            profile: Optional profile name to use for optimization.
            nearblack: Optional NearblackOptions object for configuring the nearblack process.
            overview_resampling: The resampling method to use for creating overviews. Default is "average".

        Returns:
            list[dict]: List of created optimized raster objects.
        """
        body: dict = {
            "creation_options": {
                "nearblack_options": nearblack.model_dump(
                    exclude_none=True, mode="json"
                )
                if nearblack
                else NearblackOptions(enabled=False).model_dump(
                    exclude_none=True, mode="json"
                ),
                "overview_resampling": overview_resampling,
            }
        }
        if raster_ids:
            body["raster_ids"] = raster_ids
        if profile:
            body["profile"] = profile
        response = await self._post(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/optimized/",
            json=body,
        )
        return response.json()

    async def run_optimize_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        optimize_raster_ids: list[int] | None,
        retry_failed: bool = False,
    ) -> dict:
        """
        Run the optimization process on optimized raster objects.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the optimized rasters.
            optimize_raster_ids: Optional list of optimized raster IDs to process. If None, all optimized rasters in the collection will be processed.
            retry_failed: If True, retry previously failed optimization jobs.

        Returns:
            dict: The group job object representing the optimization process.
        """
        body: dict = {"retry_failed": retry_failed}
        if optimize_raster_ids:
            body["optimized_raster_ids"] = optimize_raster_ids
        response = await self._post(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/optimized/run",
            json=body,
        )
        response_json = response.json()

        job = await self.wait_for_group_job(response_json["group_job"]["group_id"])
        return job

    async def create_optimized_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_ids: list[int] | None,
    ) -> list[dict]:
        """
        Create optimized point cloud objects in the database.

        This function creates optimized point clouds (copc format) but does not run the actual optimization process.
        Run the optimization using the `run_optimize_point_clouds` function with the returned same point cloud IDs.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point clouds.
            point_cloud_ids: Optional list of point cloud IDs to optimize. If None, all point clouds in the collection will be optimized.
        Returns:
            list[dict]: List of created optimized point cloud objects.
        """
        body: dict = {}
        if point_cloud_ids:
            body["point_cloud_ids"] = point_cloud_ids
        response = await self._post(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/optimized/",
            json=body,
        )
        return response.json()

    async def run_optimize_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_ids: list[int] | None,
        retry_failed: bool = False,
    ) -> dict:
        """
        Run the optimization process on optimized point cloud objects.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the optimized point clouds.
            point_cloud_ids: Optional list of optimized point cloud IDs to process. If None, all optimized point clouds in the collection will be processed.
        Returns:
            dict: The group job object representing the optimization process.
        """
        body: dict = {"retry_failed": retry_failed}
        if point_cloud_ids:
            body["point_cloud_ids"] = point_cloud_ids
        response = await self._post(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/optimized/run",
            json=body,
        )
        response_json = response.json()
        # Wait for the group job to complete
        job = await self.wait_for_group_job(response_json["group_job"]["group_id"])
        return job

    async def get_job(
        self,
        job_id: int,
    ) -> dict:
        """
        Retrieve information about a specific job.

        Args:
            job_id: The ID of the job to retrieve.

        Returns:
            dict: The job object with its details.
        """
        response = await self._get(f"/jobs/{job_id}")
        return response.json()

    async def get_job_group(
        self,
        job_id: int,
    ) -> dict:
        """
        Retrieve information about a specific job group.

        Args:
            job_id: The ID of the job group to retrieve.

        Returns:
            dict: The job group object with its details.
        """
        response = await self._get(f"/jobs/groups/{job_id}")
        return response.json()

    async def wait_for_job(
        self,
        job_id: int,
        timeout: int = 600,
    ) -> dict:
        """
        Wait for a job to complete, polling its status at regular intervals.

        Args:
            job_id: The ID of the job to wait for.
            timeout: Maximum time to wait in seconds before raising a TimeoutError. Default is 600 seconds (10 minutes).

        Returns:
            dict: The completed job object.

        Raises:
            TimeoutError: If the job does not complete within the specified timeout period.
        """
        job = await self.get_job(job_id)
        with silence_logger(logging.getLogger("httpx")):
            while not job["completed"]:
                job = await self.get_job(job_id)
                job_name = job["name"] or "unnamed"
                logger.info(f"Job {job_name} {job_id} status: {job['status']}")
                await asyncio.sleep(3)
                timeout -= 5
                if timeout <= 0:
                    raise TimeoutError("Timed out waiting for job")
        logger.info(f"Job {job_id} finished with status {job['status']}")
        return job

    async def wait_for_group_job(
        self,
        group_job_id: int,
        timeout: int = 1200,
    ) -> dict:
        """
        Wait for a group job to complete, polling its status at regular intervals.
        A group job consists of multiple individual jobs. This method displays a progress bar
        showing the completion status of all jobs in the group.

        Args:
            group_job_id: The ID of the group job to wait for.
            timeout: Maximum time to wait in seconds before raising a TimeoutError. Default is 1200 seconds (20 minutes).

        Returns:
            dict: The completed group job object.

        Raises:
            TimeoutError: If the group job does not complete within the specified timeout period.
        """
        job = await self.get_job_group(group_job_id)
        num_jobs = job["total_jobs"]
        with (
            silence_logger(logging.getLogger("httpx")),
            tqdm.tqdm(total=num_jobs, desc=f"Running group job {group_job_id}") as pbar,
        ):
            while not job["completed"]:
                job = await self.get_job_group(group_job_id)
                num_failed = job["failed_count"]
                num_success = job["success_count"]
                num_finished = num_failed + num_success
                pbar.n = num_finished
                pbar.refresh()
                await asyncio.sleep(3)
                timeout -= 5
                if timeout <= 0:
                    raise TimeoutError("Timed out waiting for optimize rasters job")
        logger.info(
            f"Group job {group_job_id} finished {job['total_jobs']} jobs, with {num_failed} failures and {num_success} successes"
        )
        return job

    async def get_optimized_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        params: ListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """
        Retrieve optimized rasters from a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the optimized rasters.
            params: Optional ListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.

        Returns:
            list[dict]: A list of optimized raster objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/optimized/",
            params=params_dict,
        )
        return response.json()

    async def delete_optimized_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        raster_id: int,
        profile: str | None = None,
    ) -> list[dict]:
        """
        Delete optimized rasters associated with a specific raster.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the raster.
            raster_id: The ID of the raster whose optimized versions should be deleted.
            profile: Optional profile name to filter which optimized rasters to delete.
                    If None, all optimized versions of the raster will be deleted.

        Returns:
            list[dict]: A list of the deleted optimized raster objects.
        """
        params = {}
        if profile:
            params["profile"] = profile
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}/rasters/{raster_id}/optimized",
            params=params,
        )
        return response.json()

    async def optimize_rasters(
        self,
        project_id: int,
        data_collection_id: int,
        raster_ids: list[int] | None = None,
        profile: str | None = None,
        nearblack: NearblackOptions | None = None,
        overview_resampling: OverviewResampling = "average",
    ) -> list[dict]:
        """
        Create and run optimization on rasters in a data collection.
        This is a convenience method that combines create_optimized_rasters and run_optimize_rasters
        into a single operation.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the rasters.
            raster_ids: Optional list of raster IDs to optimize. If None, all rasters in the collection will be optimized.
            profile: Optional profile name to use for optimization.
            nearblack: Optional NearblackOptions object for configuring the nearblack process.
            overview_resampling: The resampling method to use for creating overviews. Default is "average".

        Returns:
            list[dict]: List of optimized raster objects after the optimization process has completed.
        """
        optimized_rasters = await self.create_optimized_rasters(
            project_id,
            data_collection_id,
            raster_ids,
            profile,
            nearblack=nearblack,
            overview_resampling=overview_resampling,
        )
        await self.run_optimize_rasters(
            project_id, data_collection_id, [r["id"] for r in optimized_rasters]
        )
        optimized_rasters = await self.get_optimized_rasters(
            project_id,
            data_collection_id,
            params=ListParams(
                ids=[r["id"] for r in optimized_rasters], limit=len(optimized_rasters)
            ),
        )
        return optimized_rasters

    async def get_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        params: PointCloudListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """
        Retrieve point clouds from a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point clouds.
            params: Optional PointCloudListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.
        Returns:
            list[dict]: A list of point cloud objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json") if params else {}
        )
        params_dict.update(kwargs)
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/",
            params=params_dict,
        )
        return response.json()

    async def paginate_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        page_size: int,
        params: PointCloudListParams | None = None,
        **kwargs,
    ) -> AsyncGenerator[dict, None]:
        """
        Paginate through point clouds in a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point clouds.
            page_size: The number of point clouds to retrieve per page.
            params: Optional PointCloudListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.
        Yields:
            dict: Point cloud objects matching the filter criteria, one at a time.
        """
        params_dict = (
            params.model_dump(
                exclude_none=True, exclude={"offset", "limit"}, mode="json"
            )
            if params
            else {}
        )
        params_dict.update(kwargs)
        url = f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/"
        async for point_cloud in self._paginate(
            url=url,
            page_size=page_size,
            params=params_dict,
        ):
            yield point_cloud

    async def get_point_cloud(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_id: int,
    ) -> dict:
        """
        Retrieve a specific point cloud by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point cloud.
            point_cloud_id: The ID of the point cloud to retrieve.
        Returns:
            dict: The point cloud object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/{point_cloud_id}"
        )
        return response.json()

    async def update_point_cloud(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_id: int,
        update_input: PointCloudUpdateInput,
    ) -> dict:
        """
        Update an existing point cloud with new values.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point cloud.
            point_cloud_id: The ID of the point cloud to update.
            update_input: PointCloudUpdateInput object containing the fields to update.
        Returns:
            dict: The updated point cloud object.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/{point_cloud_id}",
            json=update_input.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def move_point_cloud(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_id: int,
        new_data_collection_id: int,
        new_project_id: int | None = None,
    ) -> dict:
        """
        Move a point cloud to a different data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the current data collection containing the point cloud.
            point_cloud_id: The ID of the point cloud to move.
            new_data_collection_id: The ID of the target data collection to move the point cloud to.
            new_project_id: The ID of the project containing the target data collection. Defaults to `project_id`.
        Returns:
            dict: The moved point cloud object.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/{point_cloud_id}/move",
            params={
                "new_project_id": new_project_id or project_id,
                "new_data_collection_id": new_data_collection_id,
            },
        )
        return response.json()

    async def delete_point_cloud(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_id: int,
    ) -> dict:
        """
        Delete a specific point cloud by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point cloud.
            point_cloud_id: The ID of the point cloud to delete.
        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/{point_cloud_id}"
        )
        return response.json()

    async def get_optimized_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        params: OptimizedPointCloudListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """
        Retrieve optimized point clouds from a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the optimized point clouds.
            params: Optional PointCloudListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.
        Returns:
            list[dict]: A list of optimized point cloud objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else OptimizedPointCloudListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/optimized/",
            params=params_dict,
        )
        return response.json()

    async def delete_optimized_point_cloud(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_id: int,
    ) -> dict:
        """
        Delete optimized point cloud associated with a specific point cloud.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the point cloud.
            point_cloud_id: The ID of the point cloud whose optimized versions should be deleted.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}/point_clouds/{point_cloud_id}/optimized"
        )
        return response.json()

    async def optimize_point_clouds(
        self,
        project_id: int,
        data_collection_id: int,
        point_cloud_ids: list[int] | None = None,
    ) -> list[dict]:
        """
        Create and run optimization on point clouds in a data collection.

        This is a convenience method that combines create_optimized_point_clouds and run_optimize_point_clouds
        into a single operation.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the rasters.
            point_cloud_ids: Optional list of raster IDs to optimize. If None, all rasters in the collection will be optimized.
        Returns:
            list[dict]: List of optimized point_cloud objects after the optimization process has completed.
        """
        optimized_point_clouds = await self.create_optimized_point_clouds(
            project_id,
            data_collection_id,
            point_cloud_ids,
        )
        await self.run_optimize_point_clouds(
            project_id,
            data_collection_id,
            [pc["id"] for pc in optimized_point_clouds],
        )
        optimized_point_clouds = await self.get_optimized_point_clouds(
            project_id,
            data_collection_id,
            params=OptimizedPointCloudListParams(
                ids=[pc["id"] for pc in optimized_point_clouds],
                limit=len(optimized_point_clouds),
            ),
        )
        return optimized_point_clouds

    async def list_gdo_users(self) -> list[str]:
        """Retrieve a list of GDO (GeoData Online) users.

        Returns:
            list[dict]: A list of GDO user names.
        """
        response = await self._get("/arcgis_services/gdo-users")
        return response.json()

    async def create_arcgis_service(
        self,
        service_type: Literal["Feature", "Image"],
        create_input: ArcgisServiceCreateInput,
    ) -> dict:
        """Create a new ArcGIS service.

        Args:
            service_type: The type of service to create, either "Feature" or "Image".
            create_input: ArcgisServiceCreateInput object containing the service configuration.

        Returns:
            dict: The created ArcGIS service object.

        Raises:
            AssertionError: If the create_input does not contain the appropriate service options for the specified service_type.
        """
        assert (
            getattr(
                create_input.options, f"{service_type.lower()}_service_options", None
            )
            is not None
        ), (
            f"Service options for {service_type} service must be provided in the create input"
        )
        response = await self._post(
            f"/arcgis_services/{service_type}/",
            json=create_input.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def list_arcgis_services(
        self,
        service_type: Literal["Feature", "Image"],
        params: ListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """List ArcGIS services of a specific type with optional filtering.

        Args:
            service_type: The type of services to list, either "Feature" or "Image".
            params: Optional ListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters.

        Returns:
            list[dict]: A list of ArcGIS service objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/arcgis_services/{service_type}/", params=params_dict
        )
        return response.json()

    async def get_arcgis_service(
        self, service_type: Literal["Feature", "Image"], service_id: int
    ) -> dict:
        """Retrieve a specific ArcGIS service by its ID.

        Args:
            service_type: The type of service, either "Feature" or "Image".
            service_id: The ID of the service to retrieve.

        Returns:
            dict: The ArcGIS service object with its details.
        """
        response = await self._get(f"/arcgis_services/{service_type}/{service_id}")
        return response.json()

    async def delete_arcgis_service(
        self, service_type: Literal["Feature", "Image"], service_id: int
    ) -> dict:
        """Delete a specific ArcGIS service by its ID.

        Args:
            service_type: The type of service, either "Feature" or "Image".
            service_id: The ID of the service to delete.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(f"/arcgis_services/{service_type}/{service_id}")
        return response.json()

    async def update_arcgis_service(
        self,
        service_type: Literal["Feature", "Image"],
        service_id: int,
        update_input: ArcgisServiceUpdateInput,
    ) -> dict:
        """Update an existing ArcGIS service with new values.

        Args:
            service_type: The type of service, either "Feature" or "Image".
            service_id: The ID of the service to update.
            update_input: ArcgisServiceUpdateInput object containing the fields to update.

        Returns:
            dict: The updated ArcGIS service object.
        """
        response = await self._put(
            f"/arcgis_services/{service_type}/{service_id}",
            json=update_input.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def start_arcgis_service(
        self,
        service_type: Literal["Feature", "Image"],
        service_id: int,
        wait: bool = True,
    ) -> dict:
        """Start a specific ArcGIS service.

        Args:
            service_type: The type of service, either "Feature" or "Image".
            service_id: The ID of the service to start.
            wait: If True, wait for the start operation to complete before returning.
                 If False, return immediately after initiating the start operation.

        Returns:
            dict: A response object containing job information and, if wait is True,
                 the updated service object after starting.
        """
        response = await self._put(
            f"/arcgis_services/{service_type}/{service_id}/start"
        )
        resp = response.json()
        job = resp["job"]
        if wait and job:
            job = await self.wait_for_job(job["job_id"], timeout=1000)
            resp["job"] = job
            resp["arcgis_service"] = await self.get_arcgis_service(
                service_type, service_id
            )
        return resp

    async def stop_arcgis_service(
        self,
        service_type: Literal["Feature", "Image"],
        service_id: int,
        wait: bool = True,
    ) -> dict:
        """Stop a specific ArcGIS service.

        Args:
            service_type: The type of service, either "Feature" or "Image".
            service_id: The ID of the service to stop.
            wait: If True, wait for the stop operation to complete before returning.
                 If False, return immediately after initiating the stop operation.

        Returns:
            dict: A response object containing job information and, if wait is True,
                 the updated service object after stopping.
        """
        response = await self._put(f"/arcgis_services/{service_type}/{service_id}/stop")
        resp = response.json()
        job = resp["job"]
        if wait and job:
            job = await self.wait_for_job(job["job_id"], timeout=1000)
            resp["job"] = job
            resp["arcgis_service"] = await self.get_arcgis_service(
                service_type, service_id
            )
        return resp

    async def refresh_arcgis_service(
        self,
        service_type: Literal["Feature", "Image"],
        service_id: int,
        refresh_data: bool = False,
        wait: bool = True,
    ) -> dict:
        """Refresh a specific ArcGIS service, optionally refreshing its data.

        Args:
            service_type: The type of service, either "Feature" or "Image".
            service_id: The ID of the service to refresh.
            refresh_data: If True, also refresh the data used by the service.
            wait: If True, wait for the refresh operation to complete before returning.
                 If False, return immediately after initiating the refresh operation.

        Returns:
            dict: A response object containing job information and, if wait is True,
                 the updated service object after refreshing.
        """
        response = await self._put(
            f"/arcgis_services/{service_type}/{service_id}/refresh",
            params={"refresh_data": refresh_data},
        )
        resp = response.json()
        job = resp["job"]
        if wait and job:
            job = await self.wait_for_job(job["job_id"], timeout=1000)
            resp["job"] = job
            resp["arcgis_service"] = await self.get_arcgis_service(
                service_type, service_id
            )
        return resp

    async def create_harvest_service(
        self,
        project_id: int,
        data_collection_id: int,
        create_input: HarvestServiceCreateInput,
    ) -> dict:
        """Create a new harvest service for a data collection.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to create the harvest service for.
            create_input: HarvestServiceCreateInput object containing the service configuration.

        Returns:
            dict: The created harvest service object.
        """
        response = await self._post(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/",
            json=create_input.model_dump(mode="json", exclude_none=True),
            sensitive_content=True,
        )
        return response.json()

    async def list_harvest_services(
        self,
        project_id: int,
        data_collection_id: int,
        params: ListParams | None = None,
        **kwargs,
    ) -> list[dict]:
        """List harvest services for a data collection with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection to list harvest services for.
            params: Optional ListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters. Ignored if params is provided.

        Returns:
            list[dict]: A list of harvest service objects matching the filter criteria.
        """
        if params and kwargs:
            logger.warning("Ignoring kwargs when params is provided")
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/",
            params=params_dict,
        )
        return response.json()

    async def get_harvest_service(
        self, project_id: int, data_collection_id: int, service_id: int
    ) -> dict:
        """Retrieve a specific harvest service by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the harvest service.
            service_id: The ID of the harvest service to retrieve.

        Returns:
            dict: The harvest service object with its details.
        """
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/{service_id}"
        )
        return response.json()

    async def update_harvest_service(
        self,
        project_id: int,
        data_collection_id: int,
        service_id: int,
        update_input: HarvestServiceUpdateInput,
    ) -> dict:
        """Update an existing harvest service with new values.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the harvest service.
            service_id: The ID of the harvest service to update.
            update_input: HarvestServiceUpdateInput object containing the fields to update.

        Returns:
            dict: The updated harvest service object.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/{service_id}",
            json=update_input.model_dump(mode="json", exclude_none=True),
            sensitive_content=True,
        )
        return response.json()

    async def delete_harvest_service(
        self, project_id: int, data_collection_id: int, service_id: int
    ) -> dict:
        """Delete a specific harvest service by its ID.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the harvest service.
            service_id: The ID of the harvest service to delete.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/{service_id}"
        )
        return response.json()

    async def get_harvest_service_tasks(
        self,
        project_id: int,
        data_collection_id: int,
        service_id: int,
        params: HarvestTaskListParams,
        **kwargs,
    ) -> list[dict]:
        """Retrieve tasks for a specific harvest service with optional filtering.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the harvest service.
            service_id: The ID of the harvest service to retrieve tasks for.
            params: HarvestTaskListParams object containing filtering parameters.
            **kwargs: Alternative way to provide filtering parameters. Ignored if params is provided.

        Returns:
            list[dict]: A list of harvest task objects matching the filter criteria.
        """
        params_dict = (
            params.model_dump(exclude_none=True, mode="json")
            if params
            else ListParams.model_validate(kwargs).model_dump(
                exclude_none=True, mode="json"
            )
        )
        response = await self._get(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/{service_id}/",
            params=params_dict,
        )
        return response.json()

    async def start_harvest_service(
        self, project_id: int, data_collection_id: int, service_id: int
    ) -> dict:
        """Start a specific harvest service.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the harvest service.
            service_id: The ID of the harvest service to start.

        Returns:
            dict: The response confirming the service has been started.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/{service_id}/start"
        )
        return response.json()

    async def stop_harvest_service(
        self, project_id: int, data_collection_id: int, service_id: int
    ) -> dict:
        """Stop a specific harvest service.

        Args:
            project_id: The ID of the project containing the data collection.
            data_collection_id: The ID of the data collection containing the harvest service.
            service_id: The ID of the harvest service to stop.

        Returns:
            dict: The response confirming the service has been stopped.
        """
        response = await self._put(
            f"/projects/{project_id}/data_collections/{data_collection_id}/harvest_services/{service_id}/stop"
        )
        return response.json()

    async def create_oidc_user(self, create_input: OIDCUserCreateInput) -> dict:
        """Create a new OIDC user in the system.

        Deprecated:
            Use [`create_user`][pixel_client.PixelClientAsync.create_user] instead.

        Args:
            create_input: OIDCUserCreateInput object containing the user information.

        Returns:
            dict: The created user object.
        """
        warnings.warn(
            "create_oidc_user is deprecated, use create_user instead",
            DeprecationWarning,
            stacklevel=2,
        )
        return await self._create_user(create_input)

    async def update_oidc_user(
        self, email: str, update_input: OIDCUserUpdateInput
    ) -> dict:
        """
        Update an existing OIDC user with new values.

        Deprecated:
            Use [`update_user`][pixel_client.PixelClientAsync.update_user] instead,
            which also keeps the user's active status unchanged.

        Args:
            email: The email address of the user to update.
            update_input: OIDCUserUpdateInput object containing the fields to update.

        Returns:
            dict: The updated user object.
        """
        warnings.warn(
            "update_oidc_user is deprecated, use update_user instead",
            DeprecationWarning,
            stacklevel=2,
        )
        return await self._update_user(email, update_input)

    async def _upload_attachment_to_s3(
        self,
        resource_type: AttachmentResourceType,
        resource_id: int,
        attachment: dict,
        file_path: Path,
    ) -> dict:
        """
        Upload an attachment file to S3 and mark it as completed.

        Args:
            resource_type: The type of resource the attachment belongs to.
            resource_id: The ID of the resource the attachment belongs to.
            attachment: The attachment object containing upload parameters.
            file_path: Path to the attachment file to upload.

        Returns:
            dict: The updated attachment object.

        Raises:
            AssertionError: If the attachment status is not "Pending".
            httpx.HTTPStatusError: If the upload fails.
        """
        assert attachment["status"] == "Pending", "Attachment status must be pending"
        response = await self._internal_client.post(
            attachment["upload_file_params"]["url"],
            files={"file": file_path.read_bytes()},
            data=attachment["upload_file_params"]["fields"],
        )
        response.raise_for_status()
        # Update the status to completed
        resp = await self._put(
            "/attachments/update",
            params={
                "resource_type": resource_type,
                "resource_id": resource_id,
                "attachment_id": attachment["id"],
                "status": "Completed",
            },
        )
        return resp.json()

    async def list_attachments(
        self,
        resource_type: AttachmentResourceType,
        resource_id: int,
        status: Literal["Pending", "Completed"] | None = None,
    ) -> list[dict]:
        """
        List attachments for a specific resource with optional status filtering.

        Args:
            resource_type: The type of resource the attachments belong to.
            resource_id: The ID of the resource to list attachments for.
            status: Optional filter for attachment status, either "Pending" or "Completed".

        Returns:
            list[dict]: A list of attachment objects matching the filter criteria.
        """
        params = {"resource_type": resource_type, "resource_id": resource_id}
        if status:
            params["status"] = status
        response = await self._get(
            "/attachments/",
            params=params,
        )
        return response.json()

    async def add_attachments(
        self,
        resource_type: AttachmentResourceType,
        resource_id: int,
        files: list[PixelAttachmentUpload | Path]
        | list[Path]
        | list[PixelAttachmentUpload],
    ) -> list[dict]:
        """
        Add one or more file attachments to a resource.

        Args:
            resource_type: The type of resource to attach files to.
            resource_id: The ID of the resource to attach files to.
            files: List of files to attach, which can be Path objects or PixelAttachmentUpload objects.

        Returns:
            list[dict]: A list of the created attachment objects.

        Raises:
            AssertionError: If any attachment names are not unique.
        """
        pixel_attachments = [
            PixelAttachmentUpload(
                file=f,
            )
            if isinstance(f, Path)
            else f
            for f in files
        ]
        assert len(set(pa.name for pa in pixel_attachments)) == len(
            pixel_attachments
        ), "Attachment names must be unique"
        response = await self._post(
            "/attachments/",
            params={"resource_type": resource_type, "resource_id": resource_id},
            json={
                "files": [
                    pa.model_dump(mode="json", exclude_none=True)
                    for pa in pixel_attachments
                ]
            },
        )
        attachments = response.json()
        sorted_attachments = sorted(attachments, key=lambda a: a["name"])
        sorted_files = sorted(pixel_attachments, key=lambda f: f.name)
        attachments = await asyncio.gather(
            *[
                self._upload_attachment_to_s3(resource_type, resource_id, a, pa.file)
                for a, pa in zip(sorted_attachments, sorted_files)
            ]
        )
        return attachments

    async def move_attachment(
        self,
        resource_type: AttachmentResourceType,
        resource_id: int,
        attachment_id: int,
        new_resource_type: AttachmentResourceType,
        new_resource_id: int,
    ) -> dict:
        """
        Move an attachment from one resource to another.

        Args:
            resource_type: The current resource type of the attachment.
            resource_id: The current resource ID the attachment belongs to.
            attachment_id: The ID of the attachment to move.
            new_resource_type: The target resource type to move the attachment to.
            new_resource_id: The target resource ID to move the attachment to.

        Returns:
            dict: The updated attachment object.
        """
        resp = await self._put(
            "/attachments/move",
            params={
                "resource_type": resource_type,
                "resource_id": resource_id,
                "attachment_id": attachment_id,
                "new_resource_type": new_resource_type,
                "new_resource_id": new_resource_id,
            },
        )
        return resp.json()

    async def delete_attachment(
        self,
        resource_type: AttachmentResourceType,
        resource_id: int,
        attachment_id: int,
    ) -> dict:
        """
        Delete a specific attachment from a resource.

        Args:
            resource_type: The resource type the attachment belongs to.
            resource_id: The resource ID the attachment belongs to.
            attachment_id: The ID of the attachment to delete.

        Returns:
            dict: The response confirming deletion.
        """
        resp = await self._delete(
            "/attachments/",
            params={
                "resource_type": resource_type,
                "resource_id": resource_id,
                "attachment_id": attachment_id,
            },
        )
        return resp.json()

    async def search_info(self, on: SearchOn):
        """
        Retrieve search metadata for a specific resource type.

        Args:
            on: The resource type to retrieve search metadata for.
        Returns:
            dict: A dictionary containing output fields, filterable fields and search capabilities.
        """
        response = await self._get(
            "/search/info",
            params={"on": on},
        )
        return response.json()

    async def search(
        self,
        search_query: dict | SearchQuery,
    ) -> SearchResults:
        """
        Perform a search across various resources.

        Args:
            search_query: SearchQuery object or dict containing the search parameters.
        Returns:
            SearchResults: The search results dictionary.
        """
        search_model = (
            SearchQuery.model_validate(search_query)
            if isinstance(search_query, dict)
            else search_query
        )
        response = await self._post(
            "/search/",
            json=search_model.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def paginate_search(
        self, search_query: dict | SearchQuery, page_size: int
    ) -> AsyncGenerator[dict, None]:
        """
        Perform a paginated search across various resources.

        Args:
            search_query: SearchQuery object or dict containing the search parameters.
            page_size: Number of results to retrieve per page.

        Yields:
            dict: Individual search result items.
        """
        search_model = (
            SearchQuery.model_validate(search_query)
            if isinstance(search_query, dict)
            else search_query
        )
        search_query_body = search_model.model_dump(
            mode="json", exclude_none=True, exclude={"limit", "offset"}
        )
        async for item in self._paginate(
            "/search/",
            method="POST",
            page_size=page_size,
            use_body=True,
            # The return of the search endpoint is in the "results" field
            results_parser=lambda resp: resp.get("results", []),
            json=search_query_body,
        ):
            yield item


    async def list_analysis_modules(self) -> list[dict]:
            """
            Retrieve a list of available analysis modules.

            Returns:
                list[dict]: A list of analysis module objects.
            """
            response = await self._get("/analysis/modules/")
            return response.json()

    async def list_analysis_services(self, params: AnalysisServiceListParams | None = None, **kwargs) -> list[dict]:
            """
            Retrieve a list of available analysis services.

            Returns:
                list[dict]: A list of analysis service objects.
            """
            if params and kwargs:
                logger.warning("Ignoring kwargs when params is provided")
            params_dict = (
                params.model_dump(exclude_none=True, mode="json")
                if params
                else AnalysisServiceListParams.model_validate(kwargs).model_dump(
                    exclude_none=True, mode="json"
                )
            )
            response = await self._get("/analysis/services/", params=params_dict)
            return response.json()

    async def create_analysis_service(self, create_input: AnalysisServiceCreateUpdateInput | None = None, **kwargs) -> dict:
            """
            Create a new analysis service.

            Returns:
                dict: The created analysis service object.
            """
            if create_input and kwargs:
                logger.warning("Ignoring kwargs when create_input is provided")
            input_dict = (
                create_input.model_dump(mode="json", exclude_none=True)
                if create_input
                else AnalysisServiceCreateUpdateInput.model_validate(kwargs).model_dump(
                    exclude_none=True, mode="json"
                )
            )
            response = await self._post(
                "/analysis/services/", json=input_dict
            )
            return response.json()

    async def patch_analysis_service(
        self,
        service_id: int,
        patch_input: AnalysisServicePatchInput | None = None,
        **kwargs,
    ) -> dict:
            """
            Partially update an analysis service.

            Only fields explicitly set on `patch_input` (or passed as kwargs)
            are sent. Backend deep-merges `config`; omitted fields are kept
            as-is. `module_name` cannot be changed.

            Returns:
                dict: The updated analysis service object.
            """
            if patch_input and kwargs:
                logger.warning("Ignoring kwargs when patch_input is provided")
            model = patch_input or AnalysisServicePatchInput.model_validate(kwargs)
            input_dict = model.model_dump(exclude_unset=True, mode="json")
            response = await self._patch(
                f"/analysis/services/{service_id}", json=input_dict
            )
            return response.json()

    async def delete_analysis_service(self, service_id: int) -> bool:
            """
            Delete an existing analysis service.

            Returns:
                dict: The response confirming deletion.
            """
            _ = await self._delete(f"/analysis/services/{service_id}")
            return True

    async def list_analysis_tasks(self, service_id: int, list_input: AnalysisTaskListParams | None = None, **kwargs) -> list[dict]:
            """
            Retrieve a list of analysis tasks.

            Returns:
                list[dict]: A list of analysis task objects.
            """
            if list_input and kwargs:
                logger.warning("Ignoring kwargs when list_input is provided")
            params_dict = (
                list_input.model_dump(exclude_none=True, mode="json")
                if list_input
                else AnalysisTaskListParams.model_validate(kwargs).model_dump(
                    exclude_none=True, mode="json"
                )
            )
            response = await self._get(f"/analysis/services/{service_id}/tasks/", params=params_dict)
            return response.json()

    async def create_analysis_task(self, service_id: int, params: AnalysisTaskCreateInput | None = None, **kwargs) -> dict:
            """
            Create a new analysis task.

            Returns:
                dict: The created analysis task object.
            """
            if params and kwargs:
                logger.warning("Ignoring kwargs when params is provided")
            input_dict = (
                params.model_dump(mode="json", exclude_none=True)
                if params
                else AnalysisTaskCreateInput.model_validate(kwargs).model_dump(
                    exclude_none=True, mode="json"
                )
            )


            response = await self._post(
                f"/analysis/services/{service_id}/tasks/", json=input_dict
            )
            return response.json()

    async def delete_analysis_task(self, service_id: int, task_id: int) -> bool:
            """
            Delete an existing analysis task.

            Returns:
                dict: The response confirming deletion.
            """
            _ = await self._delete(f"/analysis/services/{service_id}/tasks/{task_id}")
            return True

    async def run_analysis_task(self, service_id: int, task_id: int) -> dict:
            """
            Run an existing analysis task.

            Returns:
                dict: The response confirming the task has been started.
            """
            response = await self._post(f"/analysis/services/{service_id}/tasks/{task_id}/run")
            return response.json()

    async def revoke_analysis_task(self, service_id: int, task_id: int) -> dict:
            """
            Revoke an existing analysis task.

            Returns:
                dict: The response confirming the task has been revoked.
            """
            response = await self._post(f"/analysis/services/{service_id}/tasks/{task_id}/revoke")
            return response.json()

    async def pause_analysis_task(self, service_id: int, task_id: int) -> dict:
            """
            Pause a running analysis task. Only modules with `pausable` set support this.

            The task moves to `PAUSING`, then to `PAUSED` at the next step boundary.

            Returns:
                dict: The task ID and its new status.
            """
            response = await self._post(f"/analysis/services/{service_id}/tasks/{task_id}/pause")
            return response.json()

    async def resume_analysis_task(self, service_id: int, task_id: int) -> dict:
            """
            Resume a paused analysis task.

            Returns:
                dict: The task ID and its new status.
            """
            response = await self._post(f"/analysis/services/{service_id}/tasks/{task_id}/resume")
            return response.json()

    async def list_analysis_service_field_options(
        self, service_id: int, field: str
    ) -> list[dict]:
            """
            List the allowed values for a field of an analysis service's config or task parameters.

            Args:
                service_id: The ID of the analysis service.
                field: The name of the config or params field to list options for.

            Returns:
                list[dict]: Options with `value`, `label` and an optional `description`.
            """
            response = await self._get(f"/analysis/services/{service_id}/options/{field}")
            return response.json()

    async def list_roles(self) -> list[dict]:
        """List the roles that can be granted, with the permissions each one bundles.

        Returns:
            list[dict]: Role objects with `name`, `is_system` and `permissions`
                (each a `kind` and `verb`).
        """
        response = await self._get("/roles/")
        return response.json()

    async def get_role(self, name: str) -> dict:
        """Retrieve a single role and the permissions it bundles.

        Args:
            name: The role name, e.g. `"editor"`.

        Returns:
            dict: The role object.
        """
        response = await self._get(f"/roles/{name}")
        return response.json()

    async def list_permissions(self) -> list[dict]:
        """List every permission that exists, independent of which roles bundle it.

        Returns:
            list[dict]: Permission objects with a `kind` (e.g. `"project"`) and a `verb` (e.g. `"read"`).
        """
        response = await self._get("/permissions/")
        return response.json()

    async def paginate_grants(
        self,
        page_size: int,
        project_ids: list[int] | None = None,
        principal_kind: Literal["user", "group"] | None = None,
        principal_ids: list[int] | None = None,
        include_root: bool = False,
    ) -> AsyncGenerator[dict, None]:
        """Iterate over the user and group grants on the projects you administer, page by page.

        See [`list_grants`][pixel_client.PixelClientAsync.list_grants] for the filters.

        Args:
            page_size: Number of grants to fetch per request.

        Yields:
            dict: Individual grant objects.
        """
        params: dict[str, Any] = {"include_root": include_root}
        if project_ids:
            params["project_id"] = list(project_ids)
        if principal_kind:
            params["principal_kind"] = principal_kind
        if principal_ids:
            params["principal_id"] = list(principal_ids)
        offset = 0
        while True:
            response = await self._get(
                "/access/grants",
                params={**params, "offset": offset, "limit": page_size},
            )
            page = response.json()
            for item in page["items"]:
                yield item
            offset += len(page["items"])
            if not page["items"] or offset >= page["total"]:
                break

    async def list_grants(
        self,
        project_ids: list[int] | None = None,
        principal_kind: Literal["user", "group"] | None = None,
        principal_ids: list[int] | None = None,
        include_root: bool = False,
    ) -> list[dict]:
        """List the user and group grants on the projects you administer.

        Each grant says who holds which role, where it is held, and the project
        it is listed on. A grant held on a project above the ones you administer
        is listed on the highest of your projects it reaches, with `held_on`
        set to `"above"`. Group grants include the group's member ids in
        `member_ids`, so access people get through a group can be worked out.
        Fetches every page.

        Args:
            project_ids: Only grants listed on these projects.
            principal_kind: Only grants held by users, or only by groups.
            principal_ids: Only grants held by these users or groups.
            include_root: Also include organisation-wide grants.

        Returns:
            list[dict]: Grant objects.

        Note:
            Cannot be called with an api key.
        """
        return [
            grant
            async for grant in self.paginate_grants(
                GRANTS_PAGE_SIZE, project_ids, principal_kind, principal_ids, include_root
            )
        ]

    async def get_user_access(self, user: UserRef) -> dict:
        """What a user can do, and why.

        Returns the user's organisation-wide grants and every project they reach,
        each with their effective role and permissions and the grants behind
        them (direct, inherited from a parent project, or through a group).
        For another user you need to be able to list users, and you only see
        projects you administer.

        Args:
            user: The user's ID or email address.

        Returns:
            dict: The user's access, with `root` and `projects`.

        Note:
            Cannot be called with an api key.
        """
        user_id = await self._resolve_user_id(user)
        response = await self._get(f"/users/{user_id}/access")
        return response.json()

    async def get_group_access(self, group_id: int) -> dict:
        """What a group can do: its own grants, and every project it reaches with
        the effective permissions there. Only projects you can read are listed.

        Args:
            group_id: The ID of the group.

        Returns:
            dict: The group's access, with `root`, `projects` and `member_count`.

        Note:
            Cannot be called with an api key.
        """
        response = await self._get(f"/groups/{group_id}/access")
        return response.json()

    async def _resolve_user_id(self, user: UserRef) -> int:
        """Return the ID of a user given by ID or email address."""
        if isinstance(user, int):
            return user
        try:
            response = await self._get(f"/users/{user}/info")
        except PixelBadRequestError as e:
            if e.status_code == 404:
                raise ValueError(f"No user with email address {user!r}.") from e
            raise
        return response.json()["id"]

    async def _resolve_user_ids(self, users: list[UserRef] | None) -> list[int]:
        return [await self._resolve_user_id(user) for user in users or []]

    async def _resolve_user_email(self, user: UserRef) -> str:
        """Return the email address of a user given by ID or email address."""
        if isinstance(user, str):
            return user
        found = await self.list_users(ids=[user])
        if not found:
            raise ValueError(f"No user with ID {user}.")
        return found[0]["email"]

    async def _resolve_geo_service_user_ids(
        self, geo_service_users: list[GeoServiceUserRef] | None
    ) -> list[int]:
        """Return the IDs of geo service users given by ID or username."""
        refs = list(geo_service_users or [])
        if all(isinstance(ref, int) for ref in refs):
            return refs  # type: ignore[return-value]
        by_username = {u["username"]: u["id"] for u in await self.list_geo_service_users()}
        ids = []
        for ref in refs:
            if isinstance(ref, int):
                ids.append(ref)
            elif ref in by_username:
                ids.append(by_username[ref])
            else:
                raise ValueError(f"No geo service user with username {ref!r}.")
        return ids

    async def grant_access_on(
        self,
        resources: Resource | list[Resource],
        role: str | None = None,
        *,
        users: list[UserRef] | None = None,
        groups: list[int] | None = None,
        geo_service_users: list[GeoServiceUserRef] | None = None,
    ) -> list[dict]:
        """Give users, groups or geo service users a role on one or more resources.

        Everyone listed gets the same role; to give different roles, call again.
        A principal that already holds a role directly on the resource is moved
        to the requested role, also when that lowers it, so running the same
        call twice gives the same result. Access inherited from a parent project
        or through a group is not changed, and still applies.

        Which principals and roles a resource accepts depends on its type:

        - [`ProjectResource`][pixel_client.models.ProjectResource]: users and
          groups, with role `"viewer"`, `"editor"` or `"admin"`. You need the
          admin role on the project, can only give roles you hold there, and
          must be a member of each group.
        - [`ArcgisServiceResource`][pixel_client.models.ArcgisServiceResource]:
          geo service users only; the role can be omitted.

        Args:
            resources: The resource, or a list of resources, to give access on.
            role: The role to give.
            users: Users, by ID or email address.
            groups: Groups, by ID.
            geo_service_users: Geo service users, by ID or username.

        Returns:
            list[dict]: One entry per principal and resource, with `resource`,
                `principal_kind`, `principal_id`, `role` and `action`
                (`"granted"`, `"updated"` or `"unchanged"`).

        Raises:
            ValueError: If the resource does not accept these principals or this
                role, or a user or geo service user cannot be found. Raised before
                any change is made.
            PixelAccessUpdateError: If some changes failed. All other changes
                were still applied.
        """
        resources, principals = await self._prepare_access_change(
            resources, users, groups, geo_service_users, role=role, granting=True
        )
        outcome = Outcome()
        for resource in resources:
            handler = handler_for(resource)
            await handler.grant(
                self, resource, handler.check_role(resource, role), principals, outcome
            )
        if outcome.failed:
            raise PixelAccessUpdateError(outcome.succeeded, outcome.failed)
        return outcome.succeeded

    async def revoke_access_on(
        self,
        resources: Resource | list[Resource],
        *,
        users: list[UserRef] | None = None,
        groups: list[int] | None = None,
        geo_service_users: list[GeoServiceUserRef] | None = None,
    ) -> list[dict]:
        """Remove the access users, groups or geo service users hold directly on
        one or more resources.

        Principals without a direct grant on a resource are skipped. Access
        inherited from a parent project or through a group is not changed.

        Args:
            resources: The resource, or a list of resources, to remove access on.
            users: Users, by ID or email address.
            groups: Groups, by ID.
            geo_service_users: Geo service users, by ID or username.

        Returns:
            list[dict]: One entry per principal and resource, with `action`
                `"revoked"` or `"unchanged"`.

        Raises:
            ValueError: If the resource does not accept these principals, or a
                user or geo service user cannot be found.
            PixelAccessUpdateError: If some changes failed. All other changes
                were still applied.
        """
        resources, principals = await self._prepare_access_change(
            resources, users, groups, geo_service_users, role=None, granting=False
        )
        outcome = Outcome()
        for resource in resources:
            await handler_for(resource).revoke(self, resource, principals, outcome)
        if outcome.failed:
            raise PixelAccessUpdateError(outcome.succeeded, outcome.failed)
        return outcome.succeeded

    async def _prepare_access_change(
        self,
        resources: Resource | list[Resource],
        users: list[UserRef] | None,
        groups: list[int] | None,
        geo_service_users: list[GeoServiceUserRef] | None,
        role: str | None,
        granting: bool,
    ) -> tuple[list[Resource], Principals]:
        """Validate an access change and resolve its principals to IDs, before any write."""
        resources = [resources] if isinstance(resources, Resource) else list(resources)
        if not resources:
            raise ValueError("Give at least one resource to change access on.")
        given: dict[str, list] = {
            "user": list(users or []),
            "group": list(groups or []),
            "geo_service_user": list(geo_service_users or []),
        }
        if not any(given.values()):
            raise ValueError(
                "Give at least one principal: users, groups or geo_service_users."
            )
        for resource in resources:
            handler = handler_for(resource)
            handler.check(resource, given)  # type: ignore[arg-type]
            if granting:
                handler.check_role(resource, role)
        principals: Principals = {
            "user": await self._resolve_user_ids(given["user"]),
            "group": given["group"],
            "geo_service_user": await self._resolve_geo_service_user_ids(
                given["geo_service_user"]
            ),
        }
        return resources, principals

    async def grant_project_access(
        self,
        projects: int | list[int],
        role: RoleType,
        *,
        users: list[UserRef] | None = None,
        groups: list[int] | None = None,
    ) -> list[dict]:
        """Give users and groups a role on one or more projects (and their subprojects).

        Everyone listed gets the same role. Anyone who already has a role
        directly on a project is moved to `role`, also when that lowers it.
        Access inherited from a parent project or through a group still applies.

        You need the admin role on each project, can only give roles you hold
        there yourself, and must be a member of each group.

        Args:
            projects: The project ID, or a list of project IDs.
            role: `"viewer"`, `"editor"` or `"admin"`.
            users: Users, by ID or email address.
            groups: Groups, by ID.

        Returns:
            list[dict]: One entry per principal and project, with `action`
                `"granted"`, `"updated"` or `"unchanged"`.

        Raises:
            ValueError: If the role is not a project role or a user cannot be found.
            PixelAccessUpdateError: If some changes failed. All other changes
                were still applied.
        """
        return await self.grant_access_on(
            self._project_resources(projects), role, users=users, groups=groups
        )

    async def revoke_project_access(
        self,
        projects: int | list[int],
        *,
        users: list[UserRef] | None = None,
        groups: list[int] | None = None,
    ) -> list[dict]:
        """Remove the access users and groups hold directly on one or more projects.

        Access inherited from a parent project or through a group still applies.
        Admins of the project can remove anyone; group admins can also remove
        their group.

        Args:
            projects: The project ID, or a list of project IDs.
            users: Users, by ID or email address.
            groups: Groups, by ID.

        Returns:
            list[dict]: One entry per principal and project, with `action`
                `"revoked"` or `"unchanged"`.
        """
        return await self.revoke_access_on(
            self._project_resources(projects), users=users, groups=groups
        )

    @staticmethod
    def _project_resources(projects: int | list[int]) -> list[Resource]:
        project_ids = [projects] if isinstance(projects, int) else projects
        return [ProjectResource(project_id) for project_id in project_ids]

    async def list_project_users(self, project_id: int) -> list[dict]:
        """List the users with a role directly on a project.

        Users who reach the project through a parent project or a group are not
        included; see [`list_grants`][pixel_client.PixelClientAsync.list_grants].

        Args:
            project_id: The ID of the project.

        Returns:
            list[dict]: User objects, each with their `role` on the project.
        """
        response = await self._get(f"/projects/{project_id}/users")
        return response.json()

    async def list_project_groups(self, project_id: int) -> list[dict]:
        """List the groups with a role directly on a project.

        Args:
            project_id: The ID of the project.

        Returns:
            list[dict]: Group objects, each with their `role` on the project.
        """
        response = await self._get(f"/projects/{project_id}/groups")
        return response.json()

    async def _grant_projects_by_role(
        self,
        projects: dict[int, RoleType],
        result: dict,
        **principals: list,
    ) -> None:
        """Grant each project its role, then raise one error for anything that failed."""
        by_role: dict[RoleType, list[int]] = defaultdict(list)
        for project_id, role in projects.items():
            by_role[role].append(project_id)
        succeeded: list[dict] = []
        failed: list[tuple[dict, Exception]] = []
        for role, project_ids in by_role.items():
            try:
                succeeded += await self.grant_project_access(project_ids, role, **principals)
            except PixelAccessUpdateError as e:
                succeeded += e.succeeded
                failed += e.failed
        if failed:
            raise PixelAccessUpdateError(succeeded, failed, result=result)

    async def create_user(
        self,
        email: str,
        first_name: str,
        last_name: str,
        password: str | SecretStr,
        user_type: UserType = "viewer",
        *,
        groups: list[int] | None = None,
        projects: dict[int, RoleType] | None = None,
    ) -> dict:
        """Create a user, optionally adding them to groups and giving them project roles.

        A new user sees nothing until they are given access, directly or
        through a group. Superusers only.

        Args:
            email: The user's email address, used to sign in.
            first_name: The user's first name.
            last_name: The user's last name.
            password: A temporary password. The user chooses their own the first
                time they sign in, so share it securely.
            user_type: The account type. `"viewer"` sees only what they are given
                access to; `"creator"` can also create top-level projects, groups
                and api keys, and list users; `"superuser"` can see and change
                everything in the organisation.
            groups: Groups to add the user to, by ID. You must be a group admin
                of each.
            projects: Mapping of project ID to the role the user gets there,
                e.g. `{42: "editor"}`.

        Returns:
            dict: The created user.

        Raises:
            PixelAccessUpdateError: If the user was created but some project
                roles could not be given. The user is in `result`.

        Note:
            Cannot be called with an api key.
        """
        create_input = OIDCUserCreateInput(
            email=email,
            first_name=first_name,
            last_name=last_name,
            password=password,  # type: ignore[arg-type]
            user_type=user_type,
            groups=[GroupMembershipInput(group_id=group_id) for group_id in groups or []],
        )
        user = await self._create_user(create_input)
        if projects:
            await self._grant_projects_by_role(projects, result=user, users=[user["id"]])
        return user

    async def _create_user(self, create_input: OIDCUserCreateInput) -> dict:
        response = await self._post(
            "/users/",
            json=create_input.model_dump(exclude_none=True, mode="json"),
            sensitive_content=True,
        )
        return response.json()

    async def update_user(
        self,
        user: UserRef,
        *,
        first_name: str | None = None,
        last_name: str | None = None,
        password: str | SecretStr | None = None,
        user_type: UserType | None = None,
    ) -> dict:
        """Change a user's name, account type or password. Superusers only.

        Fields left out are not changed. The user's active status is kept.

        Args:
            user: The user's ID or email address.
            first_name: New first name.
            last_name: New last name.
            password: A new temporary password. The user chooses their own the
                next time they sign in.
            user_type: New account type: `"viewer"`, `"creator"` or `"superuser"`.

        Returns:
            dict: The updated user.

        Note:
            Cannot be called with an api key.
        """
        email = await self._resolve_user_email(user)
        current = await self.get_user(email)
        update_input = OIDCUserUpdateInput(
            first_name=first_name,
            last_name=last_name,
            password=password,  # type: ignore[arg-type]
            user_type=user_type,
            enabled=current["is_active"],
        )
        return await self._update_user(email, update_input)

    async def _update_user(self, email: str, update_input: OIDCUserUpdateInput) -> dict:
        response = await self._put(
            f"/users/{email}",
            json=update_input.model_dump(mode="json", exclude_none=True),
            sensitive_content=True,
        )
        return response.json()

    async def get_user(self, user: UserRef) -> dict:
        """Retrieve a user.

        Args:
            user: The user's ID or email address.

        Returns:
            dict: The user, including `id`, `email`, `is_active` and `user_type`.
        """
        email = await self._resolve_user_email(user)
        response = await self._get(f"/users/{email}/info")
        return response.json()

    async def list_users(
        self, params: UserListParams | None = None, **kwargs
    ) -> list[dict]:
        """List the users in the organisation, including deactivated ones.

        Needs the creator or superuser account type.

        Args:
            params: Optional UserListParams object with `name`, `ids`, `offset` and `limit`.
            **kwargs: Alternative way to provide the filtering parameters.

        Returns:
            list[dict]: User objects.
        """
        params = params or UserListParams.model_validate(kwargs)
        response = await self._get(
            "/users/", params=params.model_dump(exclude_none=True, mode="json")
        )
        return response.json()

    async def paginate_users(
        self, page_size: int, params: UserListParams | None = None, **kwargs
    ) -> AsyncGenerator[dict, None]:
        """Iterate over the users in the organisation, page by page.

        Args:
            page_size: Number of users to fetch per request.
            params: Optional UserListParams object with filtering parameters.
            **kwargs: Alternative way to provide the filtering parameters.

        Yields:
            dict: Individual user objects.
        """
        params = params or UserListParams.model_validate(kwargs)
        async for user in self._paginate(
            "/users/",
            page_size=page_size,
            params=params.model_dump(exclude_none=True, mode="json", exclude={"offset", "limit"}),
        ):
            yield user

    async def create_group(
        self,
        name: str,
        description: str = "",
        *,
        members: list[UserRef] | None = None,
        projects: dict[int, RoleType] | None = None,
    ) -> dict:
        """Create a group, optionally with members and project roles.

        You become the group's first member and admin. Members get the group's
        access on its projects.

        Args:
            name: The name of the group.
            description: A description of the group.
            members: Users to add, by ID or email address.
            projects: Mapping of project ID to the role the group gets there,
                e.g. `{42: "editor"}`. You need the admin role on each project.

        Returns:
            dict: The created group.

        Raises:
            ValueError: If a member cannot be found. Raised before the group is created.
            PixelAccessUpdateError: If the group was created but some members or
                project roles could not be added. The group is in `result`.
        """
        user_ids = await self._resolve_user_ids(members)
        response = await self._post(
            "/groups/",
            json=GroupCreateInput(name=name, description=description).model_dump(mode="json"),
        )
        group = response.json()
        succeeded: list[dict] = []
        failed: list[tuple[dict, Exception]] = []
        if user_ids:
            changes = [
                {
                    "resource": {"kind": "group", "id": group["id"]},
                    "principal_kind": "user",
                    "principal_id": user_id,
                    "role": None,
                    "action": "added",
                }
                for user_id in user_ids
            ]
            try:
                await self._add_group_members(group["id"], user_ids, admin=False)
            except (PixelBadRequestError, httpx.HTTPError) as e:
                failed += [(c, e) for c in changes]
            else:
                succeeded += changes
        if projects:
            try:
                await self._grant_projects_by_role(projects, result=group, groups=[group["id"]])
            except PixelAccessUpdateError as e:
                succeeded += e.succeeded
                failed += e.failed
        if failed:
            raise PixelAccessUpdateError(succeeded, failed, result=group)
        return group

    async def list_groups(
        self, params: GroupListParams | None = None, **kwargs
    ) -> list[dict]:
        """List the groups you are a member of. Superusers see every group.

        Args:
            params: Optional GroupListParams object with `name`, `ids`, `offset` and `limit`.
            **kwargs: Alternative way to provide the filtering parameters.

        Returns:
            list[dict]: Group objects.
        """
        params = params or GroupListParams.model_validate(kwargs)
        response = await self._get(
            "/groups/", params=params.model_dump(exclude_none=True, mode="json")
        )
        return response.json()

    async def get_group(self, group_id: int) -> dict:
        """Retrieve a group with its members and the projects it has a role on.

        Visible to members of the group and to superusers.

        Args:
            group_id: The ID of the group.

        Returns:
            dict: The group, with `members` (each with `is_admin`) and `projects`
                (each with the group's `role`).
        """
        response = await self._get(f"/groups/{group_id}")
        return response.json()

    async def update_group(
        self,
        group_id: int,
        *,
        name: str | None = None,
        description: str | None = None,
    ) -> dict:
        """Rename a group or change its description. Group admins only.

        Members and access are not changed.

        Args:
            group_id: The ID of the group.
            name: New name.
            description: New description.

        Returns:
            dict: The updated group.
        """
        update_input = GroupUpdateInput(name=name, description=description)
        response = await self._patch(
            f"/groups/{group_id}",
            json=update_input.model_dump(mode="json", exclude_none=True),
        )
        return response.json()

    async def delete_group(self, group_id: int) -> dict:
        """Delete a group. Group admins only.

        Every member, and every api key that follows the group, loses the access
        it had through the group. This cannot be undone.

        Args:
            group_id: The ID of the group.

        Returns:
            dict: The response confirming deletion.
        """
        response = await self._delete(f"/groups/{group_id}")
        return response.json()

    async def add_group_members(
        self, group_id: int, users: list[UserRef], *, admin: bool = False
    ) -> list[dict]:
        """Add users to a group. They get the group's access. Group admins only.

        Args:
            group_id: The ID of the group.
            users: Users to add, by ID or email address.
            admin: Also make them group admins. Admins can rename and delete the
                group and manage its members; it gives no access to data.

        Returns:
            list[dict]: The added members.
        """
        return await self._add_group_members(
            group_id, await self._resolve_user_ids(users), admin=admin
        )

    async def _add_group_members(
        self, group_id: int, user_ids: list[int], admin: bool
    ) -> list[dict]:
        response = await self._post(
            f"/groups/{group_id}/users",
            json={
                "users": [
                    GroupMemberInput(user_id=user_id, is_admin=admin).model_dump(mode="json")
                    for user_id in user_ids
                ]
            },
        )
        return response.json()

    async def remove_group_member(self, group_id: int, user: UserRef) -> dict:
        """Remove a member from a group. They lose the access the group gave them.

        Group admins can remove anyone; any member can remove themselves to
        leave the group. The last admin cannot be removed.

        Args:
            group_id: The ID of the group.
            user: The member, by ID or email address.

        Returns:
            dict: The response confirming removal.
        """
        user_id = await self._resolve_user_id(user)
        response = await self._delete(f"/groups/{group_id}/users/{user_id}")
        return response.json()

    async def set_group_admin(
        self, group_id: int, user: UserRef, admin: bool = True
    ) -> dict:
        """Make a member a group admin, or remove their admin status. Group admins only.

        Only members can be admins, and a group always keeps at least one.

        Args:
            group_id: The ID of the group.
            user: The member, by ID or email address.
            admin: `True` to make them admin, `False` to remove admin status.

        Returns:
            dict: The updated member.
        """
        user_id = await self._resolve_user_id(user)
        response = await self._put(
            f"/groups/{group_id}/users/{user_id}/admin", params={"is_admin": admin}
        )
        return response.json()

    async def create_api_key(
        self,
        name: str,
        *,
        projects: dict[int, RoleType] | None = None,
        groups: list[int] | None = None,
        base_role: BaseRoleType | None = None,
        expires_in_days: int = 90,
        expires_at: datetime.datetime | None = None,
    ) -> dict:
        """Create an api key for scripts and integrations.

        A key can never do more than you can: its access is always limited to
        your own current access, so it loses access when you do. The scope
        cannot be changed later; use
        [`replace_api_key`][pixel_client.PixelClientAsync.replace_api_key] to
        get a key with a different scope.

        Args:
            name: A name to recognise the key by.
            projects: Mapping of project ID to the role the key gets there,
                e.g. `{42: "editor"}`.
            groups: Groups the key follows, by ID. The key gets whatever each
                group can reach, even as that changes. You must be a member.
            base_role: An organisation-wide role: `"viewer"` reads all projects,
                `"creator"` can create projects and groups, `"superuser"` can do
                everything. You can only give a role you hold yourself.
            expires_in_days: Days until the key expires. Ignored if `expires_at` is given.
            expires_at: When the key expires.

        Returns:
            dict: The key's details under `api_key`, and the secret under `token`.

        Warning:
            The `token` is returned only this once and can never be retrieved
            again. Store it somewhere safe.

        Raises:
            ValueError: If none of `projects`, `groups` or `base_role` is given.

        Note:
            Cannot be called with an api key.
        """
        grants = self._api_key_grants(projects, groups, base_role)
        if not grants:
            raise ValueError(
                "An api key needs at least one of projects, groups or base_role."
            )
        return await self._create_api_key(
            ApiKeyCreateInput(
                name=name,
                expires_at=self._api_key_expiry(expires_in_days, expires_at),
                grants=grants,
            )
        )

    @staticmethod
    def _api_key_grants(
        projects: dict[int, RoleType] | None,
        groups: list[int] | None,
        base_role: BaseRoleType | None,
    ) -> list[ApiKeyGrant]:
        grants: list[ApiKeyGrant] = [
            ApiKeyProjectGrant(project_id=project_id, role=role)
            for project_id, role in (projects or {}).items()
        ]
        grants += [ApiKeyGroupGrant(group_id=group_id) for group_id in groups or []]
        if base_role is not None:
            grants.append(ApiKeyRootGrant(role=base_role))
        return grants

    @staticmethod
    def _api_key_expiry(
        expires_in_days: int, expires_at: datetime.datetime | None
    ) -> datetime.datetime:
        return expires_at or datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
            days=expires_in_days
        )

    async def _create_api_key(self, create_input: ApiKeyCreateInput) -> dict:
        response = await self._post(
            "/api_keys/", json=create_input.model_dump(mode="json")
        )
        return response.json()

    async def list_api_keys(self, all: bool = False) -> list[dict]:
        """List your api keys. Never includes the secret.

        Each key has a `status` (`"active"`, `"expired"`, `"revoked"` or
        `"owner_inactive"`) and is `weakened` when its owner has lost some of the
        access it was given.

        Args:
            all: Superusers only. List every key in the organisation.

        Returns:
            list[dict]: Api key objects, with their `grants`.

        Note:
            Cannot be called with an api key.
        """
        response = await self._get("/api_keys/", params={"all": all})
        return response.json()

    async def rename_api_key(self, api_key_id: str | UUID, name: str) -> dict:
        """Rename an api key. Only the name can change. Owner only.

        Args:
            api_key_id: The ID of the api key.
            name: The new name.

        Returns:
            dict: The updated api key.

        Note:
            Cannot be called with an api key.
        """
        response = await self._patch(f"/api_keys/{api_key_id}", json={"name": name})
        return response.json()

    async def revoke_api_key(self, api_key_id: str | UUID) -> dict:
        """Revoke an api key. Scripts using it stop working within minutes.

        The owner or a superuser can revoke a key. Revoking twice is harmless.

        Args:
            api_key_id: The ID of the api key.

        Returns:
            dict: The response confirming revocation.

        Note:
            Cannot be called with an api key.
        """
        response = await self._delete(f"/api_keys/{api_key_id}")
        return response.json()

    async def replace_api_key(
        self,
        api_key_id: str | UUID,
        *,
        name: str | None = None,
        projects: dict[int, RoleType] | None = None,
        groups: list[int] | None = None,
        base_role: BaseRoleType | None = None,
        expires_in_days: int = 90,
        expires_at: datetime.datetime | None = None,
        take_over: bool = False,
    ) -> dict:
        """Create a new api key with the same scope as an existing one, then revoke the old one.

        Use it to rotate a key, renew an expired one, or change a key's scope:
        each of `projects`, `groups` and `base_role` that you give replaces that
        part of the copied scope. An expired or revoked key is not revoked again.

        Args:
            api_key_id: The ID of the key to replace.
            name: Name of the new key. Defaults to the old name with " (new)" added.
            projects: Replaces the old key's project roles.
            groups: Replaces the groups the old key follows.
            base_role: Replaces the old key's organisation-wide role.
            expires_in_days: Days until the new key expires. Ignored if `expires_at` is given.
            expires_at: When the new key expires.
            take_over: Superusers only. Allow replacing someone else's key. The
                new key belongs to you and is limited to your access, and the
                owner's key is revoked, so their scripts stop working.

        Returns:
            dict: The new key's details under `api_key`, and the secret under `token`.

        Warning:
            The `token` is returned only this once and can never be retrieved again.

        Raises:
            ValueError: If the key is not found, or belongs to someone else and
                `take_over` is not set.

        Note:
            Cannot be called with an api key.
        """
        key = await self._find_api_key(str(api_key_id), take_over)
        copied = [self._api_key_grant_from_response(g) for g in key["grants"]]
        grants: list[ApiKeyGrant] = (
            self._api_key_grants(projects, None, None)
            if projects is not None
            else [g for g in copied if isinstance(g, ApiKeyProjectGrant)]
        )
        grants += (
            self._api_key_grants(None, groups, None)
            if groups is not None
            else [g for g in copied if isinstance(g, ApiKeyGroupGrant)]
        )
        grants += (
            self._api_key_grants(None, None, base_role)
            if base_role is not None
            else [g for g in copied if isinstance(g, ApiKeyRootGrant)]
        )
        new_key = await self._create_api_key(
            ApiKeyCreateInput(
                name=name or f"{key['name']} (new)",
                expires_at=self._api_key_expiry(expires_in_days, expires_at),
                grants=grants,
            )
        )
        if key["status"] not in ("revoked", "expired"):
            await self.revoke_api_key(key["id"])
        return new_key

    async def _find_api_key(self, api_key_id: str, take_over: bool) -> dict:
        for key in await self.list_api_keys():
            if str(key["id"]) == api_key_id:
                return key
        others: list[dict] = []
        try:
            others = await self.list_api_keys(all=True)
        except PixelBadRequestError:
            pass
        for key in others:
            if str(key["id"]) == api_key_id:
                if not take_over:
                    raise ValueError(
                        f"Api key {api_key_id!r} belongs to someone else. Pass "
                        "take_over=True to replace it with a key of your own."
                    )
                return key
        raise ValueError(f"No api key with ID {api_key_id!r}.")

    @staticmethod
    def _api_key_grant_from_response(grant: dict) -> ApiKeyGrant:
        if grant["kind"] == "project":
            return ApiKeyProjectGrant(project_id=grant["project_id"], role=grant["role"])
        if grant["kind"] == "group":
            return ApiKeyGroupGrant(group_id=grant["group_id"])
        return ApiKeyRootGrant(role=grant["role"])

    async def create_geo_service_user(
        self,
        username: str,
        description: str = "",
        password: str | SecretStr | None = None,
    ) -> dict:
        """Create a geo service user: a username and password ArcGIS clients use
        to sign in to the map services.

        A geo service user can only reach the services it is given access to
        (see [`grant_arcgis_service_access`][pixel_client.PixelClientAsync.grant_arcgis_service_access]),
        starts with access to none, and can never sign in to Pixel itself. Only
        works on ArcGIS Server v2. Superusers only.

        Args:
            username: 3-64 characters: lowercase letters, digits, `_`, `.` and
                `-`, starting with a letter or digit. Cannot be changed later.
            description: What the user is for.
            password: Leave out to have a strong password generated. A password
                you choose (12-128 characters) is never shown again.

        Returns:
            dict: The user under `geo_service_user`, and under `password` the
                generated password, or `None` if you chose one.

        Warning:
            A generated password is returned only this once. Store it somewhere safe.

        Note:
            Cannot be called with an api key.
        """
        create_input = GeoServiceUserCreateInput(
            username=username,
            description=description,
            password=password,  # type: ignore[arg-type]
        )
        response = await self._post(
            "/geo_service_users/",
            json=create_input.model_dump(mode="json", exclude_none=True),
            sensitive_content=True,
        )
        return response.json()

    async def list_geo_service_users(self) -> list[dict]:
        """List the geo service users, with the services each can access.

        Returns:
            list[dict]: Geo service user objects, with their `grants`.
        """
        response = await self._get("/geo_service_users/")
        return response.json()

    async def _geo_service_user_id(self, geo_service_user: GeoServiceUserRef) -> int:
        return (await self._resolve_geo_service_user_ids([geo_service_user]))[0]

    async def get_geo_service_user(self, geo_service_user: GeoServiceUserRef) -> dict:
        """Retrieve a geo service user.

        Args:
            geo_service_user: The geo service user's ID or username.

        Returns:
            dict: The geo service user, with the services it can access.
        """
        user_id = await self._geo_service_user_id(geo_service_user)
        response = await self._get(f"/geo_service_users/{user_id}")
        return response.json()

    async def update_geo_service_user(
        self, geo_service_user: GeoServiceUserRef, *, description: str | None = None
    ) -> dict:
        """Change a geo service user's description. Superusers only.

        Args:
            geo_service_user: The geo service user's ID or username.
            description: The new description.

        Returns:
            dict: The updated geo service user.

        Note:
            Cannot be called with an api key.
        """
        return await self._patch_geo_service_user(
            geo_service_user, {"description": description}
        )

    async def disable_geo_service_user(self, geo_service_user: GeoServiceUserRef) -> dict:
        """Stop a geo service user from signing in. Its service access is kept
        and comes back when it is enabled again. Superusers only.

        Args:
            geo_service_user: The geo service user's ID or username.

        Returns:
            dict: The updated geo service user.

        Note:
            Cannot be called with an api key.
        """
        return await self._patch_geo_service_user(geo_service_user, {"disabled": True})

    async def enable_geo_service_user(self, geo_service_user: GeoServiceUserRef) -> dict:
        """Let a disabled geo service user sign in again, with the access it had.
        Superusers only.

        Args:
            geo_service_user: The geo service user's ID or username.

        Returns:
            dict: The updated geo service user.

        Note:
            Cannot be called with an api key.
        """
        return await self._patch_geo_service_user(geo_service_user, {"disabled": False})

    async def _patch_geo_service_user(
        self, geo_service_user: GeoServiceUserRef, fields: dict
    ) -> dict:
        user_id = await self._geo_service_user_id(geo_service_user)
        response = await self._patch(
            f"/geo_service_users/{user_id}",
            json={k: v for k, v in fields.items() if v is not None},
        )
        return response.json()

    async def rotate_geo_service_user_password(
        self,
        geo_service_user: GeoServiceUserRef,
        password: str | SecretStr | None = None,
    ) -> dict:
        """Replace a geo service user's password. The old one stops working
        immediately. Superusers only.

        Args:
            geo_service_user: The geo service user's ID or username.
            password: Leave out to have a strong password generated. A password
                you choose (12-128 characters) is never shown again.

        Returns:
            dict: The user under `geo_service_user`, and under `password` the
                generated password, or `None` if you chose one.

        Warning:
            A generated password is returned only this once. Store it somewhere safe.

        Note:
            Cannot be called with an api key.
        """
        user_id = await self._geo_service_user_id(geo_service_user)
        if isinstance(password, SecretStr):
            password = password.get_secret_value()
        response = await self._post(
            f"/geo_service_users/{user_id}/rotate",
            json={"password": password} if password else {},
            sensitive_content=True,
        )
        return response.json()

    async def delete_geo_service_user(self, geo_service_user: GeoServiceUserRef) -> dict:
        """Delete a geo service user and all its service access. This cannot be
        undone; use [`disable_geo_service_user`][pixel_client.PixelClientAsync.disable_geo_service_user]
        to stop it temporarily. Superusers only.

        Args:
            geo_service_user: The geo service user's ID or username.

        Returns:
            dict: The response confirming deletion.

        Note:
            Cannot be called with an api key.
        """
        user_id = await self._geo_service_user_id(geo_service_user)
        response = await self._delete(f"/geo_service_users/{user_id}")
        return response.json()

    async def grant_arcgis_service_access(
        self,
        service_type: Literal["Feature", "Image"],
        service_id: int,
        geo_service_users: list[GeoServiceUserRef],
    ) -> list[dict]:
        """Give geo service users access to an ArcGIS service. Superusers only.

        Changes reach the map services within a few minutes.

        Args:
            service_type: `"Feature"` or `"Image"`.
            service_id: The ID of the ArcGIS service.
            geo_service_users: Geo service users, by ID or username.

        Returns:
            list[dict]: One entry per geo service user, with `action`
                `"granted"` or `"unchanged"`.
        """
        return await self.grant_access_on(
            ArcgisServiceResource(service_type, service_id),
            geo_service_users=geo_service_users,
        )

    async def revoke_arcgis_service_access(
        self,
        service_type: Literal["Feature", "Image"],
        service_id: int,
        geo_service_users: list[GeoServiceUserRef],
    ) -> list[dict]:
        """Remove geo service users' access to an ArcGIS service. Superusers only.

        Args:
            service_type: `"Feature"` or `"Image"`.
            service_id: The ID of the ArcGIS service.
            geo_service_users: Geo service users, by ID or username.

        Returns:
            list[dict]: One entry per geo service user, with `action`
                `"revoked"` or `"unchanged"`.
        """
        return await self.revoke_access_on(
            ArcgisServiceResource(service_type, service_id),
            geo_service_users=geo_service_users,
        )

    async def list_arcgis_service_access(
        self, service_type: Literal["Feature", "Image"], service_id: int
    ) -> list[dict]:
        """List the geo service users with access to an ArcGIS service.

        Args:
            service_type: `"Feature"` or `"Image"`.
            service_id: The ID of the ArcGIS service.

        Returns:
            list[dict]: Geo service user objects.
        """
        response = await self._get(
            f"/arcgis_services/{service_type}/{service_id}/access"
        )
        return response.json()