Skip to content

refactor: consolidate Python SDK converters into single _converters module #7226

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
copybara-service wants to merge 1 commit into from
Closed

refactor: consolidate Python SDK converters into single _converters module #7226

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7,537 changes: 7,537 additions & 0 deletions agentplatform/_genai/_converters.py
View file
Open in desktop

Large diffs are not rendered by default.

291 changes: 56 additions & 235 deletions agentplatform/_genai/datasets.py
View file
Open in desktop

Large diffs are not rendered by default.

127 changes: 24 additions & 103 deletions agentplatform/_genai/endpoints.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,112 +17,21 @@

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

from google.genai import _api_module
from google.genai import _common
from google.genai import types as genai_types
from google.genai._common import get_value_by_path as getv
from google.genai._common import set_value_by_path as setv

from . import _converters
from . import _operations_utils
from . import _transformers as t
from . import types

logger = logging.getLogger("agentplatform_genai.endpoints")


def _DeleteEndpointRequestParameters_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"],
t.t_strict_endpoint(getv(from_object, ["name"])),
)

return to_object


def _GetEndpointOperationParameters_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, ["operation_name"]) is not None:
setv(
to_object, ["_url", "operationName"], getv(from_object, ["operation_name"])
)

return to_object


def _GetEndpointParameters_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"],
t.t_strict_endpoint(getv(from_object, ["name"])),
)

return to_object


def _PredictConfig_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, ["parameters"]) is not None:
setv(parent_object, ["parameters"], getv(from_object, ["parameters"]))

return to_object


def _PredictParameters_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", "endpoint"], t.t_endpoint(getv(from_object, ["name"])))

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

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

return to_object


def _UndeployModelRequestParameters_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", "endpoint"],
t.t_strict_endpoint(getv(from_object, ["name"])),
)

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

return to_object


class Endpoints(_api_module.BaseModule):
"""Class for managing Endpoints for prediction, undeployment, and deletion."""

Expand Down Expand Up @@ -150,7 +59,9 @@ def _undeploy(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _UndeployModelRequestParameters_to_vertex(parameter_model)
request_dict = _converters._UndeployModelRequestParameters_to_vertex(
parameter_model
)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{endpoint}:undeployModel".format_map(request_url_dict)
Expand Down Expand Up @@ -224,7 +135,7 @@ def _predict(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _PredictParameters_to_vertex(parameter_model)
request_dict = _converters._PredictParameters_to_vertex(parameter_model)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{endpoint}:predict".format_map(request_url_dict)
Expand Down Expand Up @@ -293,7 +204,9 @@ def _delete(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _DeleteEndpointRequestParameters_to_vertex(parameter_model)
request_dict = _converters._DeleteEndpointRequestParameters_to_vertex(
parameter_model
)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{name}".format_map(request_url_dict)
Expand Down Expand Up @@ -362,7 +275,7 @@ def get(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _GetEndpointParameters_to_vertex(parameter_model)
request_dict = _converters._GetEndpointParameters_to_vertex(parameter_model)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{name}".format_map(request_url_dict)
Expand Down Expand Up @@ -430,7 +343,9 @@ def _get_endpoint_operation(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _GetEndpointOperationParameters_to_vertex(parameter_model)
request_dict = _converters._GetEndpointOperationParameters_to_vertex(
parameter_model
)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{operationName}".format_map(request_url_dict)
Expand Down Expand Up @@ -658,7 +573,9 @@ async def _undeploy(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _UndeployModelRequestParameters_to_vertex(parameter_model)
request_dict = _converters._UndeployModelRequestParameters_to_vertex(
parameter_model
)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{endpoint}:undeployModel".format_map(request_url_dict)
Expand Down Expand Up @@ -734,7 +651,7 @@ async def _predict(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _PredictParameters_to_vertex(parameter_model)
request_dict = _converters._PredictParameters_to_vertex(parameter_model)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{endpoint}:predict".format_map(request_url_dict)
Expand Down Expand Up @@ -805,7 +722,9 @@ async def _delete(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _DeleteEndpointRequestParameters_to_vertex(parameter_model)
request_dict = _converters._DeleteEndpointRequestParameters_to_vertex(
parameter_model
)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{name}".format_map(request_url_dict)
Expand Down Expand Up @@ -876,7 +795,7 @@ async def get(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _GetEndpointParameters_to_vertex(parameter_model)
request_dict = _converters._GetEndpointParameters_to_vertex(parameter_model)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{name}".format_map(request_url_dict)
Expand Down Expand Up @@ -946,7 +865,9 @@ async def _get_endpoint_operation(
"This method is only supported in Gemini Enterprise Agent Platform mode, not in Gemini Developer API mode."
)
else:
request_dict = _GetEndpointOperationParameters_to_vertex(parameter_model)
request_dict = _converters._GetEndpointOperationParameters_to_vertex(
parameter_model
)
request_url_dict = request_dict.get("_url")
if request_url_dict:
path = "{operationName}".format_map(request_url_dict)
Expand Down
Loading
Loading