Skip to content

Fleet Module (FM)

Python API reference for the tlm_cli.fleet package.

Use this package to interact with the Fleet Manager programmatically. FMClient gives you typed access to claimed devices and their live status — the same interfaces that power the tlm-cli fm commands.


FMClient

tlm_cli.fleet.client.FMClient

Client interface for accessing Fleet Manager API interfaces.

This class exposes a property-based interface for accessing various Fleet Manager API model interfaces. Each property returns an interface responsible for interacting with a specific Fleet Manager model.

Parameters:

Name Type Description Default
resource_api JsonApiProtocol | None

Optional resources API provider. When omitted, the models keep their lazily built ResourceApi session. Inject a custom implementation for testing. The injected session is bound to every model served by /resources, not only the Fleet Manager ones.

None

Examples:

>>> client = FMClient()
>>> device_interface = client.claimed_device
>>> device = device_interface.get("serial_number", "SN-001")
Source code in tlm_cli/fleet/client.py
class FMClient:
    """Client interface for accessing Fleet Manager API interfaces.

    This class exposes a property-based interface for accessing various Fleet Manager API
    model interfaces. Each property returns an interface responsible for
    interacting with a specific Fleet Manager model.

    Args:
        resource_api: Optional resources API provider. When omitted, the models keep their
            lazily built ``ResourceApi`` session. Inject a custom implementation for testing.
            The injected session is bound to every model served by ``/resources``, not only
            the Fleet Manager ones.

    Examples:
        >>> client = FMClient()
        >>> device_interface = client.claimed_device
        >>> device = device_interface.get("serial_number", "SN-001")
    """

    def __init__(self, resource_api: JsonApiProtocol | None = None) -> None:
        if resource_api is None:
            return

        bind_session(ResourcesApiModel, resource_api.api)

    @property
    def claimed_device(self) -> ClaimedDeviceInterface:
        """Returns the interface for interacting with ClaimedDevice resources.

        Returns:
            ClaimedDeviceInterface: The interface for managing ClaimedDevice models.

        Examples:
            >>> client.claimed_device.get("serial_number", "SN-001")
        """
        interface: ClaimedDeviceInterface = FMInterfaceFactory.get_interface("claimed_device")
        return interface

    @property
    def claimed_device_status(self) -> ClaimedDeviceStatusInterface:
        """Returns the interface for interacting with ClaimedDeviceStatus resources.

        Returns:
            ClaimedDeviceStatusInterface: The interface for managing ClaimedDeviceStatus models.

        Examples:
            >>> client.claimed_device_status.get("connhex_id", "abc-123")
        """
        interface: ClaimedDeviceStatusInterface = FMInterfaceFactory.get_interface("claimed_device_status")
        return interface

    def name_claimed_device(
        self,
        resource_id: str,
        name: str,
        *,
        attempts: int = 5,
        delay: float = 3.0,
        on_retry: Callable[[int, int], None] | None = None,
    ) -> bool:
        """Name a freshly claimed device, retrying while the Fleet Manager has not indexed it yet.

        Args:
            resource_id: Resource ID returned by the claim.
            name: Display name to assign.
            attempts: Maximum number of tries.
            delay: Seconds to wait between tries.
            on_retry: Called as ``on_retry(attempt, attempts)`` before each wait.

        Returns:
            bool: True once the name is saved; False when every attempt failed.
        """
        for attempt in range(1, attempts + 1):
            try:
                device = self.claimed_device.get_id(resource_id)
                self.claimed_device.set_attributes(device, {"name": name})
                device.save()
                return True
            except ApiClientError:
                if attempt == attempts:
                    return False
                if on_retry:
                    on_retry(attempt, attempts)
                time.sleep(delay)
        return False

claimed_device property

Returns the interface for interacting with ClaimedDevice resources.

Returns:

Name Type Description
ClaimedDeviceInterface ClaimedDeviceInterface

The interface for managing ClaimedDevice models.

Examples:

>>> client.claimed_device.get("serial_number", "SN-001")

claimed_device_status property

Returns the interface for interacting with ClaimedDeviceStatus resources.

Returns:

Name Type Description
ClaimedDeviceStatusInterface ClaimedDeviceStatusInterface

The interface for managing ClaimedDeviceStatus models.

Examples:

>>> client.claimed_device_status.get("connhex_id", "abc-123")

name_claimed_device(resource_id, name, *, attempts=5, delay=3.0, on_retry=None)

Name a freshly claimed device, retrying while the Fleet Manager has not indexed it yet.

Parameters:

Name Type Description Default
resource_id str

Resource ID returned by the claim.

required
name str

Display name to assign.

required
attempts int

Maximum number of tries.

5
delay float

Seconds to wait between tries.

3.0
on_retry Callable[[int, int], None] | None

Called as on_retry(attempt, attempts) before each wait.

None

Returns:

Name Type Description
bool bool

True once the name is saved; False when every attempt failed.

Source code in tlm_cli/fleet/client.py
def name_claimed_device(
    self,
    resource_id: str,
    name: str,
    *,
    attempts: int = 5,
    delay: float = 3.0,
    on_retry: Callable[[int, int], None] | None = None,
) -> bool:
    """Name a freshly claimed device, retrying while the Fleet Manager has not indexed it yet.

    Args:
        resource_id: Resource ID returned by the claim.
        name: Display name to assign.
        attempts: Maximum number of tries.
        delay: Seconds to wait between tries.
        on_retry: Called as ``on_retry(attempt, attempts)`` before each wait.

    Returns:
        bool: True once the name is saved; False when every attempt failed.
    """
    for attempt in range(1, attempts + 1):
        try:
            device = self.claimed_device.get_id(resource_id)
            self.claimed_device.set_attributes(device, {"name": name})
            device.save()
            return True
        except ApiClientError:
            if attempt == attempts:
                return False
            if on_retry:
                on_retry(attempt, attempts)
            time.sleep(delay)
    return False

FMInterfaceFactory

tlm_cli.fleet.client.FMInterfaceFactory

Bases: BaseInterfaceFactory

Factory class for retrieving Fleet Manager API interface instances.

This class provides access to interface classes responsible for interacting with various Fleet Manager API models. It also supports data population for those models through their corresponding interfaces.

Attributes:

Name Type Description
_model_map dict

A mapping of model names to their interface classes.

Examples:

>>> interface = FMInterfaceFactory.get_interface("claimed_device")
>>> device = interface.new()
Source code in tlm_cli/fleet/client.py
class FMInterfaceFactory(BaseInterfaceFactory):
    """Factory class for retrieving Fleet Manager API interface instances.

    This class provides access to interface classes responsible for interacting
    with various Fleet Manager API models. It also supports data population for those
    models through their corresponding interfaces.

    Attributes:
        _model_map (dict): A mapping of model names to their interface classes.

    Examples:
        >>> interface = FMInterfaceFactory.get_interface("claimed_device")
        >>> device = interface.new()
    """

    _model_map = {
        "claimed_device": ClaimedDeviceInterface,
        "claimed_device_status": ClaimedDeviceStatusInterface,
    }