Skip to content

client

AcrCloud client.

Client

Client class with various methods to call ACRCloud API v2 endpoints.

Examples

Create an instance with configuration.
```python
>>> bearer_token = "bearer-token"
>>> config = Client.Config(retries= 5, backoff_factor= 0.1)
>>> client = Client(bearer_token, config=config)

```
Source code in acrclient/client.py 
 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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class Client:
    """Client class with various methods to call ACRCloud API v2 endpoints.

    Examples
    --------
        Create an instance with configuration.
        ```python
        >>> bearer_token = "bearer-token"
        >>> config = Client.Config(retries= 5, backoff_factor= 0.1)
        >>> client = Client(bearer_token, config=config)

        ```

    """

    class Config:
        """Configuration for acrclient."""

        def __init__(
            self: Self,
            retries: bool | int | None = 5,
            backoff_factor: float = 0.1,
        ) -> None:
            """Parameters

            retries: Total number of retries to allow.

            backoff_factor: A backoff factor to apply between attempts after the
                second try (most errors are resolved immediately by a second try
                without a delay). urllib3 will sleep for::
                    {backoff factor} * (2 ** ({number of total retries} - 1))
                seconds. If the backoff_factor is 0.1, then :func:`Retry.sleep`
                will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. It will
                never be longer than `backoff_max`.
                By default, backoff is set to 0.1.

            """
            self._retries: bool | int | None = retries
            self._backoff_factor: float = backoff_factor

        @property
        def retries(self: Self) -> bool | int | None:
            """Get total number of allowed retries."""
            return self._retries

        @property
        def backoff_factor(self: Self) -> float:
            """Get backoff factor to applied between attempts."""
            return self._backoff_factor

    def __init__(
        self: Self,
        bearer_token: str,
        base_url: str = "https://eu-api-v2.acrcloud.com",
        config: Config | None = None,
    ) -> None:
        """Parameters

        bearer_token: The bearer token for ACRCloud.

        """
        self.base_url: str = base_url

        self._config: Client.Config | None = config or Client.Config()
        self._auth: _Auth = _Auth(bearer_token=bearer_token)
        self._session = Session()
        self._session.mount(
            "https://",
            HTTPAdapter(
                max_retries=Retry(
                    total=self._config.retries,
                    backoff_factor=self._config.backoff_factor,
                ),
            ),
        )

    def get(
        self: Self,
        path: str,
        params: Any = None,  # noqa: ANN401
        **kwargs: Any,  # noqa: ANN401
    ) -> Response:
        """Fetch JSON data from ACRCloud API with set Access Key param.

        Args:
        ----
            path: URL path
            params: Parameters for request (usually used as GET params)
            **kwargs: Get passed to `requests.get`

        Returns:
        -------
            Response object

        """
        url = f"{self.base_url}{path}"
        if not kwargs.get("timeout"):
            kwargs["timeout"] = 60

        # pylint: disable-next=missing-timeout
        response = self._session.get(url=url, auth=self._auth, params=params, **kwargs)
        response.raise_for_status()
        return response

    def json(
        self: Self,
        path: str,
        params: Any = None,  # noqa: ANN401
        **kwargs: Any,  # noqa: ANN401
    ) -> Any:  # noqa: ANN401
        """Get the json results of a get call.

        Args:
        ----
            path: URL path
            params: Parameters for request (usually used as GET params)
            **kwargs: Get passed to `requests.get`

        Returns:
        -------
            Data from API

        """
        response = self.get(path, params=params, **kwargs)
        return response.json()

    def get_bm_cs_projects_results(
        self: Self,
        project_id: int,
        stream_id: str,
        params: GetBmCsProjectsResultsParams | None = None,
        **kwargs: Any,  # noqa: ANN401
    ) -> Any:  # noqa: ANN401
        """Get Custom Broadcast Monitoring Streams Results from ACRCloud.

        Args:
        ----
            project_id: Custom Broadcast Monitoring Project ID
            stream_id: Custom Broadcast Monitoring Stream ID
            params: GET parameters for request
            **kwargs: Get passed to `requests.get`

        Returns:
        -------
            Data from API

        """
        return self.json(
            path=f"/api/bm-cs-projects/{project_id}/streams/{stream_id}/results",
            params=params,
            **kwargs,
        ).get("data")

Config

Configuration for acrclient.

Source code in acrclient/client.py 
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
class Config:
    """Configuration for acrclient."""

    def __init__(
        self: Self,
        retries: bool | int | None = 5,
        backoff_factor: float = 0.1,
    ) -> None:
        """Parameters

        retries: Total number of retries to allow.

        backoff_factor: A backoff factor to apply between attempts after the
            second try (most errors are resolved immediately by a second try
            without a delay). urllib3 will sleep for::
                {backoff factor} * (2 ** ({number of total retries} - 1))
            seconds. If the backoff_factor is 0.1, then :func:`Retry.sleep`
            will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. It will
            never be longer than `backoff_max`.
            By default, backoff is set to 0.1.

        """
        self._retries: bool | int | None = retries
        self._backoff_factor: float = backoff_factor

    @property
    def retries(self: Self) -> bool | int | None:
        """Get total number of allowed retries."""
        return self._retries

    @property
    def backoff_factor(self: Self) -> float:
        """Get backoff factor to applied between attempts."""
        return self._backoff_factor

backoff_factor: float property

Get backoff factor to applied between attempts.

retries: bool | int | None property

Get total number of allowed retries.

__init__(retries=5, backoff_factor=0.1)

Parameters

retries: Total number of retries to allow.

A backoff factor to apply between attempts after the

second try (most errors are resolved immediately by a second try without a delay). urllib3 will sleep for:: {backoff factor} * (2 ** ({number of total retries} - 1)) seconds. If the backoff_factor is 0.1, then :func:Retry.sleep will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. It will never be longer than backoff_max. By default, backoff is set to 0.1.

Source code in acrclient/client.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self: Self,
    retries: bool | int | None = 5,
    backoff_factor: float = 0.1,
) -> None:
    """Parameters

    retries: Total number of retries to allow.

    backoff_factor: A backoff factor to apply between attempts after the
        second try (most errors are resolved immediately by a second try
        without a delay). urllib3 will sleep for::
            {backoff factor} * (2 ** ({number of total retries} - 1))
        seconds. If the backoff_factor is 0.1, then :func:`Retry.sleep`
        will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. It will
        never be longer than `backoff_max`.
        By default, backoff is set to 0.1.

    """
    self._retries: bool | int | None = retries
    self._backoff_factor: float = backoff_factor

__init__(bearer_token, base_url='https://eu-api-v2.acrcloud.com', config=None)

Parameters

bearer_token: The bearer token for ACRCloud.

Source code in acrclient/client.py 
 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
def __init__(
    self: Self,
    bearer_token: str,
    base_url: str = "https://eu-api-v2.acrcloud.com",
    config: Config | None = None,
) -> None:
    """Parameters

    bearer_token: The bearer token for ACRCloud.

    """
    self.base_url: str = base_url

    self._config: Client.Config | None = config or Client.Config()
    self._auth: _Auth = _Auth(bearer_token=bearer_token)
    self._session = Session()
    self._session.mount(
        "https://",
        HTTPAdapter(
            max_retries=Retry(
                total=self._config.retries,
                backoff_factor=self._config.backoff_factor,
            ),
        ),
    )

get(path, params=None, **kwargs)

Fetch JSON data from ACRCloud API with set Access Key param.

path: URL path
params: Parameters for request (usually used as GET params)
**kwargs: Get passed to `requests.get`

Response object
Source code in acrclient/client.py 
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
def get(
    self: Self,
    path: str,
    params: Any = None,  # noqa: ANN401
    **kwargs: Any,  # noqa: ANN401
) -> Response:
    """Fetch JSON data from ACRCloud API with set Access Key param.

    Args:
    ----
        path: URL path
        params: Parameters for request (usually used as GET params)
        **kwargs: Get passed to `requests.get`

    Returns:
    -------
        Response object

    """
    url = f"{self.base_url}{path}"
    if not kwargs.get("timeout"):
        kwargs["timeout"] = 60

    # pylint: disable-next=missing-timeout
    response = self._session.get(url=url, auth=self._auth, params=params, **kwargs)
    response.raise_for_status()
    return response

get_bm_cs_projects_results(project_id, stream_id, params=None, **kwargs)

Get Custom Broadcast Monitoring Streams Results from ACRCloud.

project_id: Custom Broadcast Monitoring Project ID
stream_id: Custom Broadcast Monitoring Stream ID
params: GET parameters for request
**kwargs: Get passed to `requests.get`

Data from API
Source code in acrclient/client.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def get_bm_cs_projects_results(
    self: Self,
    project_id: int,
    stream_id: str,
    params: GetBmCsProjectsResultsParams | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> Any:  # noqa: ANN401
    """Get Custom Broadcast Monitoring Streams Results from ACRCloud.

    Args:
    ----
        project_id: Custom Broadcast Monitoring Project ID
        stream_id: Custom Broadcast Monitoring Stream ID
        params: GET parameters for request
        **kwargs: Get passed to `requests.get`

    Returns:
    -------
        Data from API

    """
    return self.json(
        path=f"/api/bm-cs-projects/{project_id}/streams/{stream_id}/results",
        params=params,
        **kwargs,
    ).get("data")

json(path, params=None, **kwargs)

Get the json results of a get call.

path: URL path
params: Parameters for request (usually used as GET params)
**kwargs: Get passed to `requests.get`

Data from API
Source code in acrclient/client.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def json(
    self: Self,
    path: str,
    params: Any = None,  # noqa: ANN401
    **kwargs: Any,  # noqa: ANN401
) -> Any:  # noqa: ANN401
    """Get the json results of a get call.

    Args:
    ----
        path: URL path
        params: Parameters for request (usually used as GET params)
        **kwargs: Get passed to `requests.get`

    Returns:
    -------
        Data from API

    """
    response = self.get(path, params=params, **kwargs)
    return response.json()