FamilySafety

pyfamilysafety.FamilySafety

FamilySafety(auth)

Main client for Microsoft Family Safety.

Holds the authenticated API session, family member accounts, and optional pending screen-time request handling.

Attributes:

Name	Type	Description
accounts	list[Account]	Family members with Digital Safety enabled, populated after the first :meth:update.
experimental	bool	When True, :meth:update fetches pending screen-time requests and invokes registered callbacks.
pending_requests		Latest pending request payloads (experimental mode).

Initialize the client.

Parameters:

Name	Type	Description	Default
auth	Authenticator	Authenticated :class:~pyfamilysafety.authenticator.Authenticator session.	required

Source code in pyfamilysafety/__init__.py

def __init__(self, auth: Authenticator) -> None:
    """Initialize the client.

    Args:
        auth: Authenticated :class:`~pyfamilysafety.authenticator.Authenticator`
            session.
    """
    self._api: FamilySafetyAPI = FamilySafetyAPI(auth=auth)
    self.accounts: list[Account] = []
    self.experimental: bool = False
    self.pending_requests = []
    self._pending_request_callbacks = []

accounts instance-attribute

accounts = []

experimental instance-attribute

experimental = False

pending_requests instance-attribute

pending_requests = []

get_account

get_account(user_id)

Return the account with the given member ID.

Parameters:

Name	Type	Description	Default
user_id	str	Microsoft family member ID from the roster.	required

Returns:

Name	Type	Description
Matching	Account	class:~pyfamilysafety.account.Account.

Raises:

Type	Description
IndexError	If no account matches user_id.

Source code in pyfamilysafety/__init__.py

def get_account(self, user_id: str) -> Account:
    """Return the account with the given member ID.

    Args:
        user_id: Microsoft family member ID from the roster.

    Returns:
        Matching :class:`~pyfamilysafety.account.Account`.

    Raises:
        IndexError: If no account matches ``user_id``.
    """
    return [x for x in self.accounts if x.user_id == user_id][0]

get_request

get_request(request_id)

Return a single pending request by ID.

Parameters:

Name	Type	Description	Default
request_id	str	Pending request identifier from the API.	required

Returns:

Type	Description
dict	Raw pending request dictionary.

Raises:

Type	Description
ValueError	If the request is not in :attr:pending_requests.

Source code in pyfamilysafety/__init__.py

def get_request(self, request_id: str) -> dict:
    """Return a single pending request by ID.

    Args:
        request_id: Pending request identifier from the API.

    Returns:
        Raw pending request dictionary.

    Raises:
        ValueError: If the request is not in :attr:`pending_requests`.
    """
    request = [x for x in self.pending_requests if x["id"] == request_id]
    if len(request) > 0:
        return request[0]
    raise ValueError("Pending request not found")

get_account_requests

get_account_requests(user_id)

Return pending requests for a family member.

Parameters:

Name	Type	Description	Default
user_id	str	Member ID (puid field on pending requests).	required

Returns:

Type	Description
list	List of pending request dictionaries for that member.

Source code in pyfamilysafety/__init__.py

def get_account_requests(self, user_id: str) -> list:
    """Return pending requests for a family member.

    Args:
        user_id: Member ID (``puid`` field on pending requests).

    Returns:
        List of pending request dictionaries for that member.
    """
    return [x for x in self.pending_requests if x["puid"] == user_id]

add_pending_request_callback

add_pending_request_callback(callback)

Register a callback invoked after pending requests are refreshed.

Parameters:

Name	Type	Description	Default
callback	Callable	Callable; may be sync or async.	required

Raises:

Type	Description
ValueError	If callback is not callable.

Source code in pyfamilysafety/__init__.py

def add_pending_request_callback(self, callback: Callable) -> None:
    """Register a callback invoked after pending requests are refreshed.

    Args:
        callback: Callable; may be sync or async.

    Raises:
        ValueError: If ``callback`` is not callable.
    """
    if not callable(callback):
        raise ValueError("Object must be callable.")
    if callback not in self._pending_request_callbacks:
        self._pending_request_callbacks.append(callback)

remove_pending_request_callback

remove_pending_request_callback(callback)

Remove a pending request callback.

Parameters:

Name	Type	Description	Default
callback	Callable	Previously registered callable.	required

Raises:

Type	Description
ValueError	If callback is not callable.

Source code in pyfamilysafety/__init__.py

def remove_pending_request_callback(self, callback: Callable) -> None:
    """Remove a pending request callback.

    Args:
        callback: Previously registered callable.

    Raises:
        ValueError: If ``callback`` is not callable.
    """
    if not callable(callback):
        raise ValueError("Object must be callable.")
    if callback in self._pending_request_callbacks:
        self._pending_request_callbacks.remove(callback)

approve_pending_request async

approve_pending_request(request_id, extension_time)

Approve a pending screen-time request.

Parameters:

Name	Type	Description	Default
request_id	str	Pending request identifier.	required
extension_time	int	Extra screen time to grant, in seconds.	required

Returns:

Type	Description
bool	True if the API responded with HTTP 204.

Source code in pyfamilysafety/__init__.py

async def approve_pending_request(self, request_id: str, extension_time: int) -> bool:
    """Approve a pending screen-time request.

    Args:
        request_id: Pending request identifier.
        extension_time: Extra screen time to grant, in **seconds**.

    Returns:
        ``True`` if the API responded with HTTP 204.
    """
    extension_time = extension_time*100 # convert seconds to ms
    request = self.get_request(request_id)
    response = await self._api.async_process_pending_request(
        request,
        True,
        extension_time
    )

    await self._get_pending_requests()
    return response["status"] == 204

deny_pending_request async

deny_pending_request(request_id)

Deny a pending screen-time request.

Parameters:

Name	Type	Description	Default
request_id	str	Pending request identifier.	required

Returns:

Type	Description
bool	True if the API responded with HTTP 204.

Source code in pyfamilysafety/__init__.py

async def deny_pending_request(self, request_id: str) -> bool:
    """Deny a pending screen-time request.

    Args:
        request_id: Pending request identifier.

    Returns:
        ``True`` if the API responded with HTTP 204.
    """
    request = self.get_request(request_id)
    response = await self._api.async_process_pending_request(
        request,
        False
    )
    await self._get_pending_requests()
    return response["status"] == 204

update async

update()

Refresh family roster and all account data.

On the first call, loads the roster and creates :class:Account instances. On every call, runs :meth:Account.update for each member. When :attr:experimental is enabled, also refreshes pending requests.

Raises:

Type	Description
AggregatorException	Not raised directly; transient aggregator errors are logged and ignored so cached data remains available.

Source code in pyfamilysafety/__init__.py

async def update(self):
    """Refresh family roster and all account data.

    On the first call, loads the roster and creates :class:`Account`
    instances. On every call, runs :meth:`Account.update` for each member.
    When :attr:`experimental` is enabled, also refreshes pending requests.

    Raises:
        AggregatorException: Not raised directly; transient aggregator errors
            are logged and ignored so cached data remains available.
    """
    try:
        if len(self.accounts) == 0:
            data = await self._api.send_request("get_accounts")
            self.accounts = await Account.from_dict(
                self._api,
                data["json"],
                self.experimental
            )
        coros = []
        if self.experimental:
            coros.append(self._get_pending_requests())
        for account in self.accounts:
            coros.append(account.update())
        await asyncio.gather(*coros)
    except AggregatorException:
        _LOGGER.warning("Aggregator exception occured, ignoring this update request.")