# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# Code generated by the Google Gen AI SDK generator DO NOT EDIT.

import datetime
import functools
import json
import logging
from typing import Any, Iterator, Optional, Union
from urllib.parse import urlencode

from google.genai import _api_module
from google.genai import _common
from google.genai._common import get_value_by_path as getv
from google.genai._common import set_value_by_path as setv
from google.genai.pagers import AsyncPager, Pager

from . import types


logger = logging.getLogger("vertexai_genai.sessionevents")

logger.setLevel(logging.INFO)


def _AppendAgentEngineSessionEventConfig_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}

    if getv(from_object, ["content"]) is not None:
        setv(parent_object, ["content"], getv(from_object, ["content"]))

    if getv(from_object, ["actions"]) is not None:
        setv(parent_object, ["actions"], getv(from_object, ["actions"]))

    if getv(from_object, ["error_code"]) is not None:
        setv(parent_object, ["errorCode"], getv(from_object, ["error_code"]))

    if getv(from_object, ["error_message"]) is not None:
        setv(parent_object, ["errorMessage"], getv(from_object, ["error_message"]))

    if getv(from_object, ["event_metadata"]) is not None:
        setv(parent_object, ["eventMetadata"], getv(from_object, ["event_metadata"]))

    return to_object


def _AppendAgentEngineSessionEventRequestParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["_url", "name"], getv(from_object, ["name"]))

    if getv(from_object, ["author"]) is not None:
        setv(to_object, ["author"], getv(from_object, ["author"]))

    if getv(from_object, ["invocation_id"]) is not None:
        setv(to_object, ["invocationId"], getv(from_object, ["invocation_id"]))

    if getv(from_object, ["timestamp"]) is not None:
        setv(to_object, ["timestamp"], getv(from_object, ["timestamp"]))

    if getv(from_object, ["config"]) is not None:
        setv(
            to_object,
            ["config"],
            _AppendAgentEngineSessionEventConfig_to_vertex(
                getv(from_object, ["config"]), to_object
            ),
        )

    return to_object


def _ListAgentEngineSessionEventsConfig_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}

    if getv(from_object, ["page_size"]) is not None:
        setv(parent_object, ["_query", "pageSize"], getv(from_object, ["page_size"]))

    if getv(from_object, ["page_token"]) is not None:
        setv(parent_object, ["_query", "pageToken"], getv(from_object, ["page_token"]))

    if getv(from_object, ["filter"]) is not None:
        setv(parent_object, ["_query", "filter"], getv(from_object, ["filter"]))

    return to_object


def _ListAgentEngineSessionEventsRequestParameters_to_vertex(
    from_object: Union[dict[str, Any], object],
    parent_object: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    to_object: dict[str, Any] = {}
    if getv(from_object, ["name"]) is not None:
        setv(to_object, ["_url", "name"], getv(from_object, ["name"]))

    if getv(from_object, ["config"]) is not None:
        setv(
            to_object,
            ["config"],
            _ListAgentEngineSessionEventsConfig_to_vertex(
                getv(from_object, ["config"]), to_object
            ),
        )

    return to_object


class SessionEvents(_api_module.BaseModule):

    def append(
        self,
        *,
        name: str,
        author: str,
        invocation_id: str,
        timestamp: datetime.datetime,
        config: Optional[types.AppendAgentEngineSessionEventConfigOrDict] = None,
    ) -> types.AppendAgentEngineSessionEventResponse:
        """
        Appends Agent Engine session event.

        Args:
            name (str): Required. The name of the Agent Engine session to append the event to. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}/sessions/{session_id}`.
            author (str): Required. The author of the Agent Engine session event.
            invocation_id (str): Required. The invocation ID of the Agent Engine session event.
            timestamp (datetime.datetime): Required. The timestamp of the Agent Engine session event.
            config (AppendAgentEngineSessionEventConfig):
                Optional. Additional configurations for appending the Agent Engine session event.

        Returns:
            AppendAgentEngineSessionEventResponse: The requested Agent Engine session event.

        """

        parameter_model = types._AppendAgentEngineSessionEventRequestParameters(
            name=name,
            author=author,
            invocation_id=invocation_id,
            timestamp=timestamp,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _AppendAgentEngineSessionEventRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}:appendEvent".format_map(request_url_dict)
            else:
                path = "{name}:appendEvent"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("post", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.AppendAgentEngineSessionEventResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSessionEventsConfigOrDict] = None,
    ) -> types.ListAgentEngineSessionEventsResponse:
        """
        Lists Agent Engine session events.

        Args:
            name (str): Required. The name of the Agent Engine session to list events for. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}/sessions/{session_id}`.
            config (ListAgentEngineSessionEventsConfig):
                Optional. Additional configurations for listing the Agent Engine session events.

        Returns:
            ListAgentEngineSessionEventsResponse: The requested Agent Engine session events.

        """

        parameter_model = types._ListAgentEngineSessionEventsRequestParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _ListAgentEngineSessionEventsRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/events".format_map(request_url_dict)
            else:
                path = "{name}/events"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = self._api_client.request("get", path, request_dict, http_options)

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.ListAgentEngineSessionEventsResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSessionEventsConfigOrDict] = None,
    ) -> Iterator[types.SessionEvent]:
        """Lists Agent Engine session events.

        Args:
            name (str): Required. The name of the agent engine to list session
                events for.
            config (ListAgentEngineSessionEventsConfig): Optional. The configuration
                for the session events to list. Currently, the `filter` field in
                `config` only supports filtering by `timestamp`. The timestamp
                value must be enclosed in double quotes and include the time zone
                information. For example:
                `config={'filter': 'timestamp>="2025-08-07T19:44:38.4Z"'}`.

        Returns:
            Iterator[SessionEvent]: An iterable of session events.
        """

        return Pager(
            "session_events",
            functools.partial(self._list, name=name),
            self._list(name=name, config=config),
            config,
        )


class AsyncSessionEvents(_api_module.BaseModule):

    async def append(
        self,
        *,
        name: str,
        author: str,
        invocation_id: str,
        timestamp: datetime.datetime,
        config: Optional[types.AppendAgentEngineSessionEventConfigOrDict] = None,
    ) -> types.AppendAgentEngineSessionEventResponse:
        """
        Appends Agent Engine session event.

        Args:
            name (str): Required. The name of the Agent Engine session to append the event to. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}/sessions/{session_id}`.
            author (str): Required. The author of the Agent Engine session event.
            invocation_id (str): Required. The invocation ID of the Agent Engine session event.
            timestamp (datetime.datetime): Required. The timestamp of the Agent Engine session event.
            config (AppendAgentEngineSessionEventConfig):
                Optional. Additional configurations for appending the Agent Engine session event.

        Returns:
            AppendAgentEngineSessionEventResponse: The requested Agent Engine session event.

        """

        parameter_model = types._AppendAgentEngineSessionEventRequestParameters(
            name=name,
            author=author,
            invocation_id=invocation_id,
            timestamp=timestamp,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _AppendAgentEngineSessionEventRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}:appendEvent".format_map(request_url_dict)
            else:
                path = "{name}:appendEvent"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "post", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.AppendAgentEngineSessionEventResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSessionEventsConfigOrDict] = None,
    ) -> types.ListAgentEngineSessionEventsResponse:
        """
        Lists Agent Engine session events.

        Args:
            name (str): Required. The name of the Agent Engine session to list events for. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}/sessions/{session_id}`.
            config (ListAgentEngineSessionEventsConfig):
                Optional. Additional configurations for listing the Agent Engine session events.

        Returns:
            ListAgentEngineSessionEventsResponse: The requested Agent Engine session events.

        """

        parameter_model = types._ListAgentEngineSessionEventsRequestParameters(
            name=name,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _ListAgentEngineSessionEventsRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/events".format_map(request_url_dict)
            else:
                path = "{name}/events"

        query_params = request_dict.get("_query")
        if query_params:
            path = f"{path}?{urlencode(query_params)}"
        # TODO: remove the hack that pops config.
        request_dict.pop("config", None)

        http_options: Optional[types.HttpOptions] = None
        if (
            parameter_model.config is not None
            and parameter_model.config.http_options is not None
        ):
            http_options = parameter_model.config.http_options

        request_dict = _common.convert_to_dict(request_dict)
        request_dict = _common.encode_unserializable_types(request_dict)

        response = await self._api_client.async_request(
            "get", path, request_dict, http_options
        )

        response_dict = {} if not response.body else json.loads(response.body)

        return_value = types.ListAgentEngineSessionEventsResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSessionEventsConfigOrDict] = None,
    ) -> AsyncPager[types.SessionEvent]:
        """Lists Agent Engine session events.

        Args:
            name (str): Required. The name of the agent engine to list session
                events for.
            config (ListAgentEngineSessionEventsConfig): Optional. The configuration
                for the session events to list. Currently, the `filter` field in
                `config` only supports filtering by `timestamp`. The timestamp
                value must be enclosed in double quotes and include the time zone
                information. For example:
                `config={'filter': 'timestamp>="2025-08-07T19:44:38.4Z"'}`.

        Returns:
            AsyncPager[SessionEvent]: An async pager of session events.
        """

        return AsyncPager(
            "session_events",
            functools.partial(self._list, name=name),
            await self._list(name=name, config=config),
            config,
        )