Skip to content

API internals

FamilySafetyAPI is the low-level HTTP client. Most applications should use FamilySafety and Account instead.

Advanced usage

Direct use of FamilySafetyAPI requires knowledge of endpoint keys defined in pyfamilysafety.const.ENDPOINTS. See the Endpoint map.

pyfamilysafety.api.FamilySafetyAPI 

FamilySafetyAPI(auth)

The API.

Init API.

Source code in pyfamilysafety/api.py 
21
22
23
24
def __init__(self, auth: Authenticator) -> None:
    """Init API."""
    self._auth: Authenticator = auth
    self.pending_requests = []

pending_requests instance-attribute 

pending_requests = []

send_request async 

send_request(endpoint, body=None, headers=None, platform=None, **kwargs)

Sends a request to a given endpoint.

Source code in pyfamilysafety/api.py 
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
async def send_request(self, endpoint: str, body: object=None, headers: dict=None, platform: str=None, **kwargs):
    """Sends a request to a given endpoint."""
    _LOGGER.debug("Sending request to %s", endpoint)
    # Get the endpoint from the endpoints map
    e_point = ENDPOINTS.get(endpoint, None)
    if e_point is None:
        raise ValueError("Endpoint does not exist")
    # refresh the token if it has expired.
    if self._auth.access_token_expired:
        _LOGGER.debug("Token refresh required before continuing")
        await self._auth.perform_refresh()

    if headers is None:
        headers = {}
    headers["Authorization"] = self._auth.access_token
    headers["User-Agent"] = USER_AGENT
    headers["Content-Type"] = "application/json"
    if platform is not None:
        headers["Plat-Info"] = platform

    # format the URL using the kwargs
    url = e_point.get("url")
    if "{BASE_URL" in url:
        url = url.format(BASE_URL=BASE_URL, **kwargs)
    else:
        url = url.format(**kwargs)
    _LOGGER.debug("Built URL %s", url)
    # now send the HTTP request
    resp: dict = {
        "status": 0,
        "text": "",
        "json": "",
        "headers": ""
    }
    async with self._auth.client_session.request(
        method=e_point.get("method"),
        url=url,
        json=body,
        headers=headers
    ) as response:
        _LOGGER.debug("Request to %s status code %s", url, response.status)
        if _check_http_success(response.status):
            resp["status"] = response.status
            if response.status != 204:
                resp["text"] = await response.text()
                try:
                    resp["json"] = await response.json()
                except aiohttp.client_exceptions.ContentTypeError:
                    _LOGGER.debug("Unable to parse JSON response - invalid content type.")
            resp["headers"] = response.headers
        else:
            text = await response.text()
            if response.status == 500 and AGGREGATOR_ERROR in text:
                raise AggregatorException()
            if response.status == 401:
                raise Unauthorized()
            if response.status == 403:
                raise RequestDenied(text)

            raise HttpException("HTTP Error", response.status, text)

    # now return the resp dict
    return resp

async_get_accounts async 

async_get_accounts()

Retrieve data from endpoint get_accounts.

Source code in pyfamilysafety/api.py 
90
91
92
async def async_get_accounts(self):
    """Retrieve data from endpoint get_accounts."""
    return await self.send_request("get_accounts")

async_get_pending_requests async 

async_get_pending_requests()

Retrieve data from endpoint get_pending_requests.

Source code in pyfamilysafety/api.py 
94
95
96
async def async_get_pending_requests(self):
    """Retrieve data from endpoint get_pending_requests."""
    return await self.send_request("get_pending_requests")

async_get_premium_entitlement async 

async_get_premium_entitlement()

Retrieve data from endpoint get_premium_entitlement.

Source code in pyfamilysafety/api.py 
 98
 99
100
async def async_get_premium_entitlement(self):
    """Retrieve data from endpoint get_premium_entitlement."""
    return await self.send_request("get_premium_entitlement")

async_get_user_app_screentime_usage async 

async_get_user_app_screentime_usage(user_id, begin_time, end_time, platform)

Retrieve data from endpoint get_user_app_screentime_usage.

Source code in pyfamilysafety/api.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def async_get_user_app_screentime_usage(
        self,
        user_id,
        begin_time,
        end_time,
        platform
    ):
    """Retrieve data from endpoint get_user_app_screentime_usage."""
    return await self.send_request(
        "get_user_app_screentime_usage",
        headers={
            "Plat-Info": platform
        },
        USER_ID=user_id,
        BEGIN_TIME=begin_time,
        END_TIME=end_time
    )

async_get_user_device_screentime_usage async 

async_get_user_device_screentime_usage(user_id, begin_time, end_time, device_count, platform)

Retrieve data from endpoint get_user_device_screentime_usage.

Source code in pyfamilysafety/api.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def async_get_user_device_screentime_usage(
        self,
        user_id,
        begin_time,
        end_time,
        device_count,
        platform
    ):
    """Retrieve data from endpoint get_user_device_screentime_usage."""
    return await self.send_request(
        "get_user_device_screentime_usage",
        headers={
            "Plat-Info": platform
        },
        USER_ID=user_id,
        BEGIN_TIME=begin_time,
        END_TIME=end_time,
        DEVICE_COUNT=device_count,
    )

async_get_user_devices async 

async_get_user_devices(user_id)

Retrieve data from endpoint get_user_devices.

Source code in pyfamilysafety/api.py 
140
141
142
async def async_get_user_devices(self, user_id):
    """Retrieve data from endpoint get_user_devices."""
    return await self.send_request("get_user_devices", USER_ID=user_id)

async_get_user_spending async 

async_get_user_spending(user_id)

Retrieve data from endpoint get_user_spending.

Source code in pyfamilysafety/api.py 
144
145
146
async def async_get_user_spending(self, user_id):
    """Retrieve data from endpoint get_user_spending."""
    return await self.send_request("get_user_spending", USER_ID=user_id)

async_get_user_payment_methods async 

async_get_user_payment_methods(user_id, cid)

Retrieve data from endpoint get_user_payment_methods.

Source code in pyfamilysafety/api.py 
148
149
150
async def async_get_user_payment_methods(self, user_id, cid):
    """Retrieve data from endpoint get_user_payment_methods."""
    return await self.send_request("get_user_payment_methods", USER_ID=user_id, CID=cid)

async_get_user_content_restrictions async 

async_get_user_content_restrictions(user_id)

Retrieve data from endpoint get_user_content_restrictions.

Source code in pyfamilysafety/api.py 
152
153
154
async def async_get_user_content_restrictions(self, user_id):
    """Retrieve data from endpoint get_user_content_restrictions."""
    return await self.send_request("get_user_content_restrictions", USER_ID=user_id)

async_get_user_web_restrictions async 

async_get_user_web_restrictions(user_id)

Retrieve data from endpoint get_user_web_restrictions.

Source code in pyfamilysafety/api.py 
156
157
158
async def async_get_user_web_restrictions(self, user_id):
    """Retrieve data from endpoint get_user_web_restrictions."""
    return await self.send_request("get_user_web_restrictions", USER_ID=user_id)

async_update_web_restrictions async 

async_update_web_restrictions(user_id, body)

Send a PATCH request to update web restrictions.

Source code in pyfamilysafety/api.py 
160
161
162
async def async_update_web_restrictions(self, user_id, body):
    """Send a PATCH request to update web restrictions."""
    return await self.send_request("update_web_restrictions", USER_ID=user_id, body=body)

async_get_override_device_restrictions async 

async_get_override_device_restrictions(user_id)

Send a GET request to override device restrictions.

Source code in pyfamilysafety/api.py 
164
165
166
167
168
169
170
async def async_get_override_device_restrictions(self, user_id):
    """Send a GET request to override device restrictions."""
    return await self.send_request(
        "get_override_device_restrictions",
        USER_ID=user_id,
        platform="ALL",
    )

async_process_pending_request async 

async_process_pending_request(request, approved, extension_time=0)

Process a pending request using the deny and approve pending request method

Source code in pyfamilysafety/api.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
async def async_process_pending_request(
        self,
        request: dict,
        approved: bool,
        extension_time: int=0
):
    """Process a pending request using the deny and approve pending request method"""
    if approved:
        return await self.async_approve_pending_request(
            body={
                "id": request.get("id"),
                "request": {
                    "appId": request.get("id"),
                    "extension": extension_time,
                    "isGlobal": True,
                    "lockTime": request.get("lockTime"),
                    "platform": request.get("platform"),
                    "requestedTime": request.get("requestedTime")
                },
                "type": request.get("type")
            },
            user_id=request["puid"]
        )
    return await self.async_deny_pending_request(
        body={
            "id": request.get("id"),
            "request": {
                "appId": request.get("id"),
                "extension": extension_time,
                "isGlobal": True,
                "lockTime": request.get("lockTime"),
                "platform": request.get("platform"),
                "requestedTime": request.get("requestedTime")
            },
            "type": request.get("type")
        },
        user_id=request["puid"]
    )

async_deny_pending_request async 

async_deny_pending_request(user_id, body)

Send a POST request to deny a pending request.

Source code in pyfamilysafety/api.py 
211
212
213
async def async_deny_pending_request(self, user_id, body):
    """Send a POST request to deny a pending request."""
    return await self.send_request("deny_pending_request", USER_ID=user_id, body=body)

async_approve_pending_request async 

async_approve_pending_request(user_id, body)

Send a POST request to approve a pending request.

Source code in pyfamilysafety/api.py 
215
216
217
async def async_approve_pending_request(self, user_id, body):
    """Send a POST request to approve a pending request."""
    return await self.send_request("approve_pending_request", USER_ID=user_id, body=body)

async_override_device_restriction async 

async_override_device_restriction(user_id, body)

Send a POST request to override device restrictions.

Source code in pyfamilysafety/api.py 
219
220
221
222
223
224
225
226
async def async_override_device_restriction(self, user_id, body):
    """Send a POST request to override device restrictions."""
    return await self.send_request(
        "override_device_restriction",
        USER_ID=user_id,
        body=body,
        platform="ALL",
    )

async_update_schedule async 

async_update_schedule(user_id, body)

Send a PATCH request to update device limits for a platform.

Prefer :meth:~pyfamilysafety.account.Account.set_device_limits with a :class:~pyfamilysafety.schedule.DeviceLimitsSchedule instead of calling this directly.

Source code in pyfamilysafety/api.py 
228
229
230
231
232
233
234
235
async def async_update_schedule(self, user_id, body):
    """Send a PATCH request to update device limits for a platform.

    Prefer :meth:`~pyfamilysafety.account.Account.set_device_limits` with a
    :class:`~pyfamilysafety.schedule.DeviceLimitsSchedule` instead of
    calling this directly.
    """
    return await self.send_request("update_schedule", USER_ID=user_id, body=body)