stream-py/getstream/video/rtc/coordinator_api.py at 471b53a149bd5ffaa830a88f07bdf96e7e3c2a7b · GetStream/stream-py · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
"""
Coordinator API functions for Stream Video RTC.
"""

import logging
from typing import Optional

from getstream.base import StreamResponse
from getstream.models import CallRequest
from getstream.video.async_call import Call
from getstream.utils import build_body_dict

# Import the types we need from __init__ without creating circular imports
from getstream.video.rtc.models import JoinCallResponse, FastJoinCallResponse

logger = logging.getLogger("getstream.video.rtc.coordinator")


async def join_call_coordinator_request(
    call: Call,
    user_id: str,
    create: bool = False,
    data: Optional[CallRequest] = None,
    ring: Optional[bool] = None,
    notify: Optional[bool] = None,
    video: Optional[bool] = None,
    location: Optional[str] = None,
) -> StreamResponse[JoinCallResponse]:
    """Make a request to join a call via the coordinator.

    Args:
        call: The call to join
        user_id: The user ID to join the call with
        create: Whether to create the call if it doesn't exist
        data: Additional call data if creating
        ring: Whether to ring other users
        notify: Whether to notify other users
        video: Whether to enable video
        location: The preferred location

    Returns:
        A response containing the call information and credentials
    """
    # Create a token for this user
    token = call.client.stream.create_token(user_id=user_id)

    # Create a new client with this token
    client = call.client.stream.__class__(
        api_key=call.client.stream.api_key,
        api_secret=call.client.stream.api_secret,
        base_url=call.client.stream.base_url,
    )

    # Set up authentication
    client.token = token
    client.headers["Authorization"] = token
    client.client.headers["Authorization"] = token

    # Prepare path parameters for the request
    path_params = {
        "type": call.call_type,
        "id": call.id,
    }

    # Build the request body
    json_body = build_body_dict(
        location=location or "FRA",  # Default to Frankfurt if not specified
        create=create,
        notify=notify,
        ring=ring,
        video=video,
        data=data,
    )

    # Make the POST request to join the call
    return await client.post(
        "/api/v2/video/call/{type}/{id}/join",
        JoinCallResponse,
        path_params=path_params,
        json=json_body,
    )


async def fast_join_call_coordinator_request(
    call: Call,
    user_id: str,
    create: bool = False,
    data: Optional[CallRequest] = None,
    ring: Optional[bool] = None,
    notify: Optional[bool] = None,
    video: Optional[bool] = None,
    location: Optional[str] = None,
) -> StreamResponse[FastJoinCallResponse]:
    """Make a fast join request to get multiple edge credentials from the coordinator.

    Args:
        call: The call to join
        user_id: The user ID to join the call with
        create: Whether to create the call if it doesn't exist
        data: Additional call data if creating
        ring: Whether to ring other users
        notify: Whether to notify other users
        video: Whether to enable video
        location: The preferred location

    Returns:
        A response containing the call information and an array of credentials for multiple edges
    """
    # Create a token for this user
    token = call.client.stream.create_token(user_id=user_id)

    # Create a new client with this token
    client = call.client.stream.__class__(
        api_key=call.client.stream.api_key,
        api_secret=call.client.stream.api_secret,
        base_url=call.client.stream.base_url,
    )

    # Set up authentication
    client.token = token
    client.headers["Authorization"] = token
    client.client.headers["Authorization"] = token

    # Prepare path parameters for the request
    path_params = {
        "type": call.call_type,
        "id": call.id,
    }

    # Build the request body
    json_body = build_body_dict(
        location=location or "FRA",  # Default to Frankfurt if not specified
        create=create,
        notify=notify,
        ring=ring,
        video=video,
        data=data,
    )

    # Make the POST request to fast join the call
    return await client.post(
        "/api/v2/video/call/{type}/{id}/fast_join",
        FastJoinCallResponse,
        path_params=path_params,
        json=json_body,
    )