Source code for genai.user.user_service

from typing import Optional

from genai._utils.service import (
    BaseService,
    BaseServiceConfig,
    BaseServiceServices,
    get_service_action_metadata,
    set_service_action_metadata,
)
from genai.schema import (
    UserCreateEndpoint,
    UserCreateResponse,
    UserDeleteEndpoint,
    UserPatchEndpoint,
    UserPatchResponse,
    UserRetrieveEndpoint,
    UserRetrieveResponse,
)
from genai.schema._api import (
    _UserCreateParametersQuery,
    _UserCreateRequest,
    _UserDeleteParametersQuery,
    _UserPatchParametersQuery,
    _UserPatchRequest,
    _UserRetrieveParametersQuery,
)

__all__ = ["UserService"]


[docs] class UserService(BaseService[BaseServiceConfig, BaseServiceServices]):
[docs] @set_service_action_metadata(endpoint=UserCreateEndpoint) def create( self, *, first_name: Optional[str] = None, last_name: Optional[str] = None, ) -> UserCreateResponse: """ Raises: ApiResponseException: In case of a known API error. ApiNetworkException: In case of unhandled network error. """ request_body = _UserCreateRequest(first_name=first_name, last_name=last_name).model_dump() self._log_method_execution("User Create", **request_body) with self._get_http_client() as client: metadata = get_service_action_metadata(self.create) response = client.post( url=self._get_endpoint(metadata.endpoint), json=request_body, params=_UserCreateParametersQuery().model_dump(), ) return UserCreateResponse(**response.json())
[docs] @set_service_action_metadata(endpoint=UserRetrieveEndpoint) def retrieve(self) -> UserRetrieveResponse: """ Raises: ApiResponseException: In case of a known API error. ApiNetworkException: In case of unhandled network error. """ self._log_method_execution("User Retrieve") with self._get_http_client() as client: metadata = get_service_action_metadata(self.retrieve) response = client.get( url=self._get_endpoint(metadata.endpoint), params=_UserRetrieveParametersQuery().model_dump() ) return UserRetrieveResponse(**response.json())
[docs] @set_service_action_metadata(endpoint=UserPatchEndpoint) def update(self, *, tou_accepted: Optional[bool] = None) -> UserPatchResponse: """ Raises: ApiResponseException: In case of a known API error. ApiNetworkException: In case of unhandled network error. """ request_body = _UserPatchRequest(tou_accepted=tou_accepted).model_dump() self._log_method_execution("User Update", **request_body) with self._get_http_client() as client: metadata = get_service_action_metadata(self.update) response = client.patch( url=self._get_endpoint(metadata.endpoint), json=request_body, params=_UserPatchParametersQuery().model_dump(), ) return UserPatchResponse(**response.json())
[docs] @set_service_action_metadata(endpoint=UserDeleteEndpoint) def delete(self) -> None: """ Raises: ApiResponseException: In case of a known API error. ApiNetworkException: In case of unhandled network error. """ self._log_method_execution("User Delete") with self._get_http_client() as client: metadata = get_service_action_metadata(self.delete) client.delete(url=self._get_endpoint(metadata.endpoint), params=_UserDeleteParametersQuery().model_dump())