Add refresh peer token to FishjamClient API (#12)

Author: Karolk99

fishjam/_openapi_client/api/room/refresh_token.py

@@ -0,0 +1,181 @@
+from http import HTTPStatus
+from typing import Any, Dict, Optional, Union
+
+import httpx
+
+from ... import errors
+from ...client import AuthenticatedClient, Client
+from ...models.error import Error
+from ...models.peer_refresh_token_response import PeerRefreshTokenResponse
+from ...types import Response
+
+
+def _get_kwargs(
+    room_id: str,
+    id: str,
+) -> Dict[str, Any]:
+    return {
+        "method": "post",
+        "url": "/room/{room_id}/peer/{id}/refresh_token".format(
+            room_id=room_id,
+            id=id,
+        ),
+    }
+
+
+def _parse_response(
+    *, client: Union[AuthenticatedClient, Client], response: httpx.Response
+) -> Optional[Union[Error, PeerRefreshTokenResponse]]:
+    if response.status_code == HTTPStatus.CREATED:
+        response_201 = PeerRefreshTokenResponse.from_dict(response.json())
+
+        return response_201
+    if response.status_code == HTTPStatus.BAD_REQUEST:
+        response_400 = Error.from_dict(response.json())
+
+        return response_400
+    if response.status_code == HTTPStatus.UNAUTHORIZED:
+        response_401 = Error.from_dict(response.json())
+
+        return response_401
+    if response.status_code == HTTPStatus.NOT_FOUND:
+        response_404 = Error.from_dict(response.json())
+
+        return response_404
+    if response.status_code == HTTPStatus.SERVICE_UNAVAILABLE:
+        response_503 = Error.from_dict(response.json())
+
+        return response_503
+    if client.raise_on_unexpected_status:
+        raise errors.UnexpectedStatus(response.status_code, response.content)
+    else:
+        return None
+
+
+def _build_response(
+    *, client: Union[AuthenticatedClient, Client], response: httpx.Response
+) -> Response[Union[Error, PeerRefreshTokenResponse]]:
+    return Response(
+        status_code=HTTPStatus(response.status_code),
+        content=response.content,
+        headers=response.headers,
+        parsed=_parse_response(client=client, response=response),
+    )
+
+
+def sync_detailed(
+    room_id: str,
+    id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Response[Union[Error, PeerRefreshTokenResponse]]:
+    """Refresh peer token
+
+    Args:
+        room_id (str):
+        id (str):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[Union[Error, PeerRefreshTokenResponse]]
+    """
+
+    kwargs = _get_kwargs(
+        room_id=room_id,
+        id=id,
+    )
+
+    response = client.get_httpx_client().request(
+        **kwargs,
+    )
+
+    return _build_response(client=client, response=response)
+
+
+def sync(
+    room_id: str,
+    id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Optional[Union[Error, PeerRefreshTokenResponse]]:
+    """Refresh peer token
+
+    Args:
+        room_id (str):
+        id (str):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Union[Error, PeerRefreshTokenResponse]
+    """
+
+    return sync_detailed(
+        room_id=room_id,
+        id=id,
+        client=client,
+    ).parsed
+
+
+async def asyncio_detailed(
+    room_id: str,
+    id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Response[Union[Error, PeerRefreshTokenResponse]]:
+    """Refresh peer token
+
+    Args:
+        room_id (str):
+        id (str):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[Union[Error, PeerRefreshTokenResponse]]
+    """
+
+    kwargs = _get_kwargs(
+        room_id=room_id,
+        id=id,
+    )
+
+    response = await client.get_async_httpx_client().request(**kwargs)
+
+    return _build_response(client=client, response=response)
+
+
+async def asyncio(
+    room_id: str,
+    id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Optional[Union[Error, PeerRefreshTokenResponse]]:
+    """Refresh peer token
+
+    Args:
+        room_id (str):
+        id (str):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Union[Error, PeerRefreshTokenResponse]
+    """
+
+    return (
+        await asyncio_detailed(
+            room_id=room_id,
+            id=id,
+            client=client,
+        )
+    ).parsed

fishjam/_openapi_client/models/__init__.py

@@ -45,6 +45,8 @@
 from .peer_details_response_data import PeerDetailsResponseData
 from .peer_options_web_rtc import PeerOptionsWebRTC
 from .peer_options_web_rtc_metadata import PeerOptionsWebRTCMetadata
+from .peer_refresh_token_response import PeerRefreshTokenResponse
+from .peer_refresh_token_response_data import PeerRefreshTokenResponseData
 from .peer_status import PeerStatus
 from .recording_list_response import RecordingListResponse
 from .room import Room
@@ -100,6 +102,8 @@
     "PeerDetailsResponseData",
     "PeerOptionsWebRTC",
     "PeerOptionsWebRTCMetadata",
+    "PeerRefreshTokenResponse",
+    "PeerRefreshTokenResponseData",
     "PeerStatus",
     "RecordingListResponse",
     "Room",

fishjam/_openapi_client/models/peer_refresh_token_response.py

@@ -0,0 +1,68 @@
+from typing import TYPE_CHECKING, Any, Dict, List, Type, TypeVar
+
+from attrs import define as _attrs_define
+from attrs import field as _attrs_field
+
+if TYPE_CHECKING:
+    from ..models.peer_refresh_token_response_data import PeerRefreshTokenResponseData
+
+
+T = TypeVar("T", bound="PeerRefreshTokenResponse")
+
+
+@_attrs_define
+class PeerRefreshTokenResponse:
+    """Response containing new peer token"""
+
+    data: "PeerRefreshTokenResponseData"
+    """"""
+    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
+    """@private"""
+
+    def to_dict(self) -> Dict[str, Any]:
+        """@private"""
+        data = self.data.to_dict()
+
+        field_dict: Dict[str, Any] = {}
+        field_dict.update(self.additional_properties)
+        field_dict.update(
+            {
+                "data": data,
+            }
+        )
+
+        return field_dict
+
+    @classmethod
+    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
+        """@private"""
+        from ..models.peer_refresh_token_response_data import (
+            PeerRefreshTokenResponseData,
+        )
+
+        d = src_dict.copy()
+        data = PeerRefreshTokenResponseData.from_dict(d.pop("data"))
+
+        peer_refresh_token_response = cls(
+            data=data,
+        )
+
+        peer_refresh_token_response.additional_properties = d
+        return peer_refresh_token_response
+
+    @property
+    def additional_keys(self) -> List[str]:
+        """@private"""
+        return list(self.additional_properties.keys())
+
+    def __getitem__(self, key: str) -> Any:
+        return self.additional_properties[key]
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        self.additional_properties[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        del self.additional_properties[key]
+
+    def __contains__(self, key: str) -> bool:
+        return key in self.additional_properties

fishjam/_openapi_client/models/peer_refresh_token_response_data.py

@@ -0,0 +1,60 @@
+from typing import Any, Dict, List, Type, TypeVar
+
+from attrs import define as _attrs_define
+from attrs import field as _attrs_field
+
+T = TypeVar("T", bound="PeerRefreshTokenResponseData")
+
+
+@_attrs_define
+class PeerRefreshTokenResponseData:
+    """ """
+
+    token: str
+    """Token for authorizing websocket connection"""
+    additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)
+    """@private"""
+
+    def to_dict(self) -> Dict[str, Any]:
+        """@private"""
+        token = self.token
+
+        field_dict: Dict[str, Any] = {}
+        field_dict.update(self.additional_properties)
+        field_dict.update(
+            {
+                "token": token,
+            }
+        )
+
+        return field_dict
+
+    @classmethod
+    def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
+        """@private"""
+        d = src_dict.copy()
+        token = d.pop("token")
+
+        peer_refresh_token_response_data = cls(
+            token=token,
+        )
+
+        peer_refresh_token_response_data.additional_properties = d
+        return peer_refresh_token_response_data
+
+    @property
+    def additional_keys(self) -> List[str]:
+        """@private"""
+        return list(self.additional_properties.keys())
+
+    def __getitem__(self, key: str) -> Any:
+        return self.additional_properties[key]
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        self.additional_properties[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        del self.additional_properties[key]
+
+    def __contains__(self, key: str) -> bool:
+        return key in self.additional_properties

fishjam/api/_fishjam_client.py

@@ -11,11 +11,13 @@
 from fishjam._openapi_client.api.room import delete_room as room_delete_room
 from fishjam._openapi_client.api.room import get_all_rooms as room_get_all_rooms
 from fishjam._openapi_client.api.room import get_room as room_get_room
+from fishjam._openapi_client.api.room import refresh_token as room_refresh_token
 from fishjam._openapi_client.models import (
     AddPeerJsonBody,
     Peer,
     PeerDetailsResponse,
     PeerOptionsWebRTC,
+    PeerRefreshTokenResponse,
     RoomConfig,
     RoomCreateDetailsResponse,
     RoomDetailsResponse,
@@ -167,6 +169,16 @@ def delete_room(self, room_id: str) -> None:
 
         return self._request(room_delete_room, room_id=room_id)
 
+    def refresh_peer_token(self, room_id: str, peer_id: str) -> str:
+        """Refreshes peer token"""
+
+        response = cast(
+            PeerRefreshTokenResponse,
+            self._request(room_refresh_token, id=peer_id, room_id=room_id),
+        )
+
+        return response.data.token
+
     def __parse_peer_metadata(self, metadata: dict | None) -> PeerOptionsWebRTCMetadata:
         peer_metadata = PeerOptionsWebRTCMetadata()
 

tests/test_room_api.py

@@ -178,7 +178,7 @@ def test_invalid(self, room_api: FishjamClient):
             room_api.get_room("invalid_id")
 
 
-class TestCreateParticipant:
+class TestCreatePeer:
     def _assert_peer_created(
         self, room_api, webrtc_peer, room_id, server_metadata=None
     ):
@@ -227,7 +227,7 @@ def test_peer_limit_reached(self, room_api: FishjamClient):
             room_api.create_peer(room.id)
 
 
-class TestDeleteParticipant:
+class TestDeletePeer:
     def test_valid(self, room_api: FishjamClient):
         room = room_api.create_room()
         peer, _token = room_api.create_peer(room.id)
@@ -240,3 +240,18 @@ def test_invalid(self, room_api: FishjamClient):
 
         with pytest.raises(NotFoundError):
             room_api.delete_peer(room.id, peer_id="invalid_peer_id")
+
+
+class TestRefreshPeerToken:
+    def test_valid(self, room_api: FishjamClient):
+        room = room_api.create_room()
+        peer, token = room_api.create_peer(room.id)
+        refreshed_token = room_api.refresh_peer_token(room.id, peer.id)
+
+        assert token != refreshed_token
+
+    def test_invalid(self, room_api: FishjamClient):
+        room = room_api.create_room()
+
+        with pytest.raises(NotFoundError):
+            room_api.refresh_peer_token(room.id, peer_id="invalid_peer_id")