`viam.app.app_client`

Module Contents

Classes

`RobotPart`	A class that mirrors the RobotPart proto message.
`LogEntry`	A class that mirrors the LogEntry proto message.
`Fragment`	A class that mirrors the Fragment proto message.
`RobotPartHistoryEntry`	A class that mirrors the RobotPartHistoryEntry proto message.
`APIKeyAuthorization`	A class with the necessary authorization data for creating an API key.
`AppClient`	gRPC client for method calls to app.

Attributes

LOGGER

viam.app.app_client.LOGGER

class viam.app.app_client.RobotPart[source]

A class that mirrors the RobotPart proto message.

Use this class to make the attributes of a viam.proto.app.RobotPart more accessible and easier to read/interpret.

property proto: viam.proto.app.RobotPart

id: str

name: str

dns_name: str

secret: str

robot: str

location_id: str

robot_config: Mapping[str, Any] | None

last_access: datetime.datetime | None

user_supplied_info: Mapping[str, Any] | None

main_part: bool

fqdn: str

local_fqdn: str

created_on: datetime.datetime | None

secrets: List[viam.proto.app.SharedSecret] | None

classmethod from_proto(robot_part: viam.proto.app.RobotPart) → typing_extensions.Self[source]

Create a RobotPart from the .proto defined RobotPart.

Parameters:: robot_part (viam.proto.app.RobotPart) – The object to copy from.
Returns:: The RobotPart.
Return type:: RobotPart

class viam.app.app_client.LogEntry[source]

A class that mirrors the LogEntry proto message.

Use this class to make the attributes of a viam.proto.app.LogEntry more accessible and easier to read/interpret.

property proto: viam.proto.common.LogEntry

host: str

level: str

time: datetime.datetime | None

logger_name: str

message: str

caller: Mapping[str, Any] | None

stack: str

fields: List[Mapping[str, Any]] | None

classmethod from_proto(log_entry: viam.proto.common.LogEntry) → typing_extensions.Self[source]

Create a LogEntry from the .proto defined LogEntry.

Parameters:: log_entry (viam.proto.app.LogEntry) – The object to copy from.
Returns:: The LogEntry.
Return type:: LogEntry

class viam.app.app_client.Fragment[source]

A class that mirrors the Fragment proto message.

Use this class to make the attributes of a viam.proto.app.RobotPart more accessible and easier to read/interpret.

property proto: viam.proto.app.Fragment

id: str

name: str

fragment: Mapping[str, Any] | None

organization_owner: str

public: bool

created_on: datetime.datetime | None

organization_name: str

robot_part_count: int

organization_count: int

only_used_by_owner: bool

classmethod from_proto(fragment: viam.proto.app.Fragment) → typing_extensions.Self[source]

Create a Fragment from the .proto defined Fragment.

Parameters:: fragment (viam.proto.app.Fragment) – The object to copy from.
Returns:: The Fragment.
Return type:: Fragment

class viam.app.app_client.RobotPartHistoryEntry[source]

A class that mirrors the RobotPartHistoryEntry proto message.

Use this class to make the attributes of a viam.proto.app.RobotPartHistoryEntry more accessible and easier to read/interpret.

property proto: viam.proto.app.RobotPartHistoryEntry

part: str

robot: str

when: datetime.datetime | None

old: RobotPart | None

classmethod from_proto(robot_part_history_entry: viam.proto.app.RobotPartHistoryEntry) → typing_extensions.Self[source]

Create a RobotPartHistoryEntry from the .proto defined RobotPartHistoryEntry.

Parameters:: robot_part_history_entry (viam.proto.app.RobotPartHistoryEntry) – The object to copy from.
Returns:: The RobotPartHistoryEntry.
Return type:: RobotPartHistoryEntry

class viam.app.app_client.APIKeyAuthorization(role: Literal[owner] | Literal[operator], resource_type: Literal[organization] | Literal[location] | Literal[robot], resource_id: str)[source]

A class with the necessary authorization data for creating an API key.

Use this class when constructing API key authorizations to minimize the risk of malformed or missing data.

class viam.app.app_client.AppClient(channel: grpclib.client.Channel, metadata: Mapping[str, str], location_id: str | None = None)[source]

gRPC client for method calls to app.

Constructor is used by ViamClient to instantiate relevant service stub. Calls to AppClient methods should be made through ViamClient.

Establish a Connection:

import asyncio

from viam.rpc.dial import DialOptions, Credentials
from viam.app.viam_client import ViamClient


async def connect() -> ViamClient:
    # Replace "<API-KEY>" (including brackets) with your API key and "<API-KEY-ID>" with your API key ID
    dial_options = DialOptions.with_api_key("<API-KEY>", "<API-KEY-ID>")
    return await ViamClient.create_from_dial_options(dial_options)


async def main():

    # Make a ViamClient
    viam_client = await connect()
    # Instantiate an AppClient called "cloud" to run cloud app API methods on
    cloud = viam_client.app_client

    viam_client.close()

if __name__ == '__main__':
    asyncio.run(main())

abstract async create_organization(name: str) → viam.proto.app.Organization[source]

async list_organizations() → List[viam.proto.app.Organization][source]

List the organization(s) the user is an authorized owner of.

org_list = await cloud.list_organizations()

Returns:: The list of organizations.
Return type:: List[viam.proto.app.Organization]

abstract async list_organizations_by_user(user_id: str) → List[viam.proto.app.OrgDetails][source]

async get_organization(org_id: str | None = None) → viam.proto.app.Organization[source]

Return details about the requested organization.

Parameters:: org_id (Optional[str]) – ID of the organization to query. If None, defaults to the currently-authed org.
Raises:: GRPCError – If the provided org_id is invalid, or not currently authed to.
Returns:: The requested organization.
Return type:: viam.proto.app.Organization

async get_organization_namespace_availability(public_namespace: str) → bool[source]

Check the availability of an organization namespace.

available = await cloud.get_organization_namespace_availability(
    public_namespace="my-cool-organization")

Parameters:: public_namespace (str) – Organization namespace to check. Namespaces can only contain lowercase lowercase alphanumeric and dash characters.
Raises:: GRPCError – If an invalid namespace (e.g., “”) is provided.
Returns:: True if the provided namespace is available.
Return type:: bool

async update_organization(name: str | None = None, public_namespace: str | None = None, region: str | None = None, cid: str | None = None) → viam.proto.app.Organization[source]

Updates organization details.

Parameters:

name (Optional[str]) – If provided, updates the org’s name.
public_namespace (Optional[str]) – If provided, sets the org’s namespace if it hasn’t already been set.
region (Optional[str]) – If provided, updates the org’s region.
cid (Optional[str]) – If provided, update’s the org’s CRM ID.

Raises:

GRPCError – If the org’s namespace has already been set, or if the provided namespace is already taken.

Returns:

The updated organization.

Return type:

viam.proto.app.Organization

abstract async delete_organization(org_id: str | None = None) → None[source]

async list_organization_members() → Tuple[List[viam.proto.app.OrganizationMember], List[viam.proto.app.OrganizationInvite]][source]

List the members and invites of the currently authed-to organization.

member_list, invite_list = await cloud.list_organization_members()

Returns:: A tuple containing two lists; the first [0] of organization members, and the second [1] of organization invites.
Return type:: Tuple[List[viam.proto.app.OrganizationMember], List[viam.proto.app.OrganizationInvite]]

async create_organization_invite(email: str, authorizations: List[viam.proto.app.Authorization] | None = None, send_email_invite=True) → viam.proto.app.OrganizationInvite[source]

Creates an organization invite and sends it via email.

await cloud.create_organization_invite("youremail@email.com")

Parameters:

email (str) – The email address to send the invite to.
authorizations (Optional[List[viam.proto.app.Authorization]]) – Specifications of the authorizations to include in the invite. If not provided, full owner permissions will be granted.
send_email_invite (Optional[bool]) – Whether or not an email should be sent to the recipient of an invite. The user must accept the email to be added to the associated authorizations. When set to false, the user automatically receives the associated authorization on the next login of the user with the associated email address.

Raises:

GRPCError – if an invalid email is provided, or if the user is already a member of the org.

async update_organization_invite_authorizations(email: str, add_authorizations: List[viam.proto.app.Authorization] | None = None, remove_authorizations: List[viam.proto.app.Authorization] | None = None) → viam.proto.app.OrganizationInvite[source]

Update the authorizations attached to an organization invite that has already been created.

Note that an invite can only have one authorization at each resource (e.g., organization, location, robot, etc.) level and must have at least one authorization overall.

from viam.proto.app import Authorization

authorization_to_add = Authorization(
    authorization_type="some type of auth",
    authorization_id="identifier",
    resource_type="abc",
    resource_id="resource-identifier123",
    identity_id="id12345",
    organization_id="org_id_123"
)

update_invite = await cloud.update_organization_invite_authorizations(
    email="notarealemail@viam.com",
    add_authorizations =[authorization_to_add]
)

Parameters:

email (str) – Email of the user the invite was sent to.
add_authorizations (Optional[List[viam.proto.app.Authorization]]) – Optional list of authorizations to add to the invite.
remove_authorizations (Optional[List[viam.proto.app.Authorization]]) – Optional list of authorizations to remove from the invite.

Raises:

GRPCError – If no authorizations are passed or if an invalid combination of authorizations is passed (e.g. an authorization to remove when the invite only contains one authorization).

Returns:

The updated invite.

Return type:

viam.proto.app.OrganizationInvite

async delete_organization_member(user_id: str) → None[source]

Remove a member from the organization.

member_list, invite_list = await cloud.list_organization_members()
first_user_id = member_list[0].user_id

await cloud.delete_organization_member(first_user_id)

Parameters:: user_id (str) – The ID of the user to remove.

async delete_organization_invite(email: str) → None[source]

Deletes a pending organization invite.

await cloud.delete_organization_invite("youremail@email.com")

Parameters:: email (str) – The email address the pending invite was sent to.
Raises:: GRPCError – If no pending invite is associated with the provided email address.

async resend_organization_invite(email: str) → viam.proto.app.OrganizationInvite[source]

Re-sends a pending organization invite email.

await cloud.resend_organization_invite("youremail@email.com")

Parameters:: email (str) – The email address associated with the invite.
Raises:: GRPCError – If no pending invite is associated with the provided email address.

async create_location(name: str, parent_location_id: str | None = None) → viam.proto.app.Location[source]

Create and name a location under the currently authed-to organization and the specified parent location.

my_new_location = await cloud.create_location(name="Robotville",
                                              parent_location_id="111ab12345")

Parameters:

name (str) – Name of the location.
parent_location_id (Optional[str]) – Optional parent location to put the location under. Defaults to a root-level location if no location ID is provided.

Raises:

GRPCError – If either an invalid name (e.g., “”), or parent location ID (e.g., a nonexistent ID) is passed.

Returns:

The newly created location.

Return type:

viam.proto.app.Location

async get_location(location_id: str | None = None) → viam.proto.app.Location[source]

Get a location.

location = await cloud.get_location(location_id="123ab12345")

Parameters:: location_id (Optional[str]) – ID of the location to get. Defaults to the location ID provided at AppClient instantiation.
Raises:: GRPCError – If an invalid location ID is passed or if one isn’t passed and there was no location ID provided at AppClient instantiation.
Returns:: The location.
Return type:: viam.proto.app.Location

async update_location(location_id: str, name: str | None = None, parent_location_id: str | None = None) → viam.proto.app.Location[source]

Change the name of a location and/or assign it a new parent location.

# The following line takes the location with ID "abc12abcde" and moves it to be a
# sub-location of the location with ID "xyz34xxxxx"
my_updated_location = await cloud.update_location(
    location_id="abc12abcde",
    name="",
    parent_location_id="xyz34xxxxx",
)

# The following line changes the name of the location without changing its parent location
my_updated_location = await cloud.update_location(
    location_id="abc12abcde",
    name="Land Before Robots"
)

# The following line moves the location back up to be a top level location without changing its name
my_updated_location = await cloud.update_location(
    location_id="abc12abcde",
    name="",
    parent_location_id=""
)

Parameters:

location_id (str) – ID of the location to update. Must be specified.
name (Optional[str]) – Optional new name to be updated on the location. Defaults to the empty string “” (i.e., the name doesn’t change).
parent_location_id (Optional[str]) – Optional ID of new parent location to move the location under. Defaults to the empty string “” (i.e., no new parent location is assigned).

Raises:

GRPCError – If either an invalid location ID, name, or parent location ID is passed.

Returns:

The newly updated location.

Return type:

viam.proto.app.Location

async delete_location(location_id: str) → None[source]

Delete a location.

await cloud.delete_location(location_id="abc12abcde")

Parameters:: location_id (str) – ID of the location to delete. Must be specified.
Raises:: GRPCError – If an invalid location ID is passed.

async list_locations() → List[viam.proto.app.Location][source]

Get a list of all locations under the currently authed-to organization.

locations = await cloud.list_locations()

Returns:: The list of locations.
Return type:: List[viam.proto.app.Location]

abstract async share_location()[source]

abstract async unshare_location()[source]

async location_auth(location_id: str | None = None) → viam.proto.app.LocationAuth[source]

Get a location’s LocationAuth (location secret(s)).

loc_auth = await cloud.location_auth(location_id="123xy12345")

Parameters:: location_id (str) – ID of the location to retrieve LocationAuth from. Defaults to the location ID provided at AppClient instantiation.
Raises:: GRPCError – If an invalid location ID is passed or if one isn’t passed and there was no location ID provided at AppClient instantiation.
Returns:: The LocationAuth containing location secrets.
Return type:: viam.proto.app.LocationAuth

async create_location_secret(location_id: str | None = None) → viam.proto.app.LocationAuth[source]

Create a new location secret.

new_loc_auth = await cloud.create_location_secret()

Parameters:: location_id (Optional[str]) – ID of the location to generate a new secret for. Defaults to the location ID provided at AppClient instantiation.
Raises:: GRPCError – If an invalid location ID is passed or one isn’t passed and there was no location ID provided at AppClient instantiation.
Returns:: The specified location’s LocationAuth containing the newly created secret.
Return type:: viam.proto.app.LocationAuth

async delete_location_secret(secret_id: str, location_id: str | None = None) → None[source]

Delete a location secret.

await cloud.delete_location_secret(
    secret_id="abcd123-456-7890ab-cxyz98-989898xyzxyz")

Parameters:

location_id (str) – ID of the location to delete secret from. Defaults to the location ID provided at AppClient instantiation.
secret_id (str) – ID of the secret to delete.

Raises:

GRPCError – If either an invalid location ID or secret ID is passed or a location ID isn’t passed and there was no location ID provided at AppClient instantiation.

async get_robot(robot_id: str) → viam.proto.app.Robot[source]

Get a robot.

robot = await cloud.get_robot(robot_id="1a123456-x1yz-0ab0-a12xyzabc")

Parameters:: robot_id (str) – ID of the robot to get.
Raises:: GRPCError – If an invalid robot ID is passed.
Returns:: The robot.
Return type:: viam.proto.app.Robot

async get_rover_rental_robots() → List[viam.proto.app.RoverRentalRobot][source]

Returns a list of rover rental robots within an org.

rental_robots = await cloud.get_rover_rental_robots()

Returns:: The list of rover rental robots.
Return type:: List[viam.proto.app.RoverRentalRobot]

async get_robot_parts(robot_id: str) → List[RobotPart][source]

Get a list of all the parts under a specific robot.

list_of_parts = await cloud.get_robot_parts(
    robot_id="1a123456-x1yz-0ab0-a12xyzabc")

Parameters:: robot_id (str) – ID of the robot to get parts from.
Raises:: GRPCError – If an invalid robot ID is passed.
Returns:: The list of robot parts.
Return type:: List[viam.app.app_client.RobotPart]

async get_robot_part(robot_part_id: str, dest: str | None = None, indent: int = 4) → RobotPart[source]

Get a robot part.

my_robot_part = await cloud.get_robot_part(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:

robot_part_id (str) – ID of the robot part to get.
dest (Optional[str]) – Optional filepath to write the robot part’s config file in JSON format to.
indent (int) – Size (in number of spaces) of indent when writing config to dest. Defaults to 4.

Raises:

GRPCError – If an invalid robot part ID is passed.

Returns:

The robot part.

Return type:

viam.app.app_client.RobotPart

async get_robot_part_logs(robot_part_id: str, filter: str | None = None, dest: str | None = None, errors_only: bool = True, num_log_entries: int = 100) → List[LogEntry][source]

Get the logs associated with a robot part.

part_logs = await cloud.get_robot_part_logs(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22", num_log_entries=20)

Parameters:

robot_part_id (str) – ID of the robot part to get logs from.
filter (Optional[str]) – Only include logs with messages that contain the string filter. Defaults to empty string “” (i.e., no filter).
dest (Optional[str]) – Optional filepath to write the log entries to.
errors_only (bool) – Boolean specifying whether or not to only include error logs. Defaults to True.
num_log_entries (int) – Number of log entries to return. Passing 0 returns all logs. Defaults to 100. All logs or the first num_log_entries logs will be returned, whichever comes first.

Raises:

GRPCError – If an invalid robot part ID is passed.

Returns:

The list of log entries.

Return type:

List[viam.app.app_client.LogEntry]

async tail_robot_part_logs(robot_part_id: str, errors_only: bool = True, filter: str | None = None) → viam.app._logs._LogsStream[List[LogEntry]][source]

Get an asynchronous iterator that receives live robot part logs.

logs_stream = await cloud.tail_robot_part_logs(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:

robot_part_id (str) – ID of the robot part to retrieve logs from.
errors_only (bool) – Boolean specifying whether or not to only include error logs. Defaults to True.
filter (Optional[str]) – Only include logs with messages that contain the string filter. Defaults to empty string “” (i.e., no filter).

Returns:

The asynchronous iterator receiving live robot part logs.

Return type:

_LogsStream[List[LogEntry]]

async get_robot_part_history(robot_part_id: str) → List[RobotPartHistoryEntry][source]

Get a list containing the history of a robot part.

part_history = await cloud.get_robot_part_history(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:: robot_part_id (str) – ID of the robot part to retrieve history from.
Raises:: GRPCError – If an invalid robot part ID is provided.
Returns:: The list of the robot part’s history.
Return type:: List[viam.app.app_client.RobotPartHistoryEntry]

async update_robot_part(robot_part_id: str, name: str, robot_config: Mapping[str, Any] | None = None) → RobotPart[source]

Change the name and assign an optional new configuration to a robot part.

my_robot_part = await cloud.update_robot_part(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:

robot_part_id (str) – ID of the robot part to update.
name (str) – New name to be updated on the robot part.
robot_config (Mapping[str, Any]) – Optional new config represented as a dictionary to be updated on the robot part. The robot part’s config will remain as is (no change) if one isn’t passed.

Raises:

GRPCError – If either an invalid robot part ID, name, or config is passed.

Returns:

The newly updated robot part.

Return type:

viam.app.app_client.RobotPart

async new_robot_part(robot_id: str, part_name: str) → str[source]

Create a new robot part.

new_part_id = await cloud.new_robot_part(
    robot_id="1a123456-x1yz-0ab0-a12xyzabc", part_name="myNewSubPart")

Parameters:

robot_id (str) – ID of the the robot to create a new part for.
part_name (str) – Name of the new part.

Raises:

GRPCError – If either an invalid robot ID or name is passed.

Returns:

The new robot part’s ID.

Return type:

str

async delete_robot_part(robot_part_id: str) → None[source]

Delete the specified robot part.

await cloud.delete_robot_part(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:: robot_part_id (str) – ID of the robot part to delete. Must be specified.
Raises:: GRPCError – If an invalid robot part ID is passed.

async mark_part_as_main(robot_part_id: str) → None[source]

Mark a robot part as the main part of a robot.

await cloud.mark_part_as_main(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:: robot_part_id (str) – ID of the robot part to mark as main.
Raises:: GRPCError – If an invalid robot part ID is passed.

async mark_part_for_restart(robot_part_id: str) → None[source]

Mark the specified robot part for restart.

await cloud.mark_part_for_restart(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:: robot_part_id (str) – ID of the robot part to mark for restart.
Raises:: GRPCError – If an invalid robot part ID is passed.

async create_robot_part_secret(robot_part_id: str) → RobotPart[source]

Create a robot part secret.

part_with_new_secret = await cloud.create_robot_part_secret(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22")

Parameters:: robot_part_id (str) – ID of the robot part to create a secret for.
Raises:: GRPCError – If an invalid robot part ID is passed.
Returns:: The robot part the new secret was generated for.
Return type:: viam.app.app_client.RobotPart

async delete_robot_part_secret(robot_part_id: str, secret_id: str) → None[source]

Delete a robot part secret.

await cloud.delete_robot_part_secret(
    robot_part_id="abc12345-1a23-1234-ab12-a22a22a2aa22",
    secret_id="123xyz12-abcd-4321-12ab-12xy1xyz12xy")

Parameters:

robot_part_id (str) – ID of the robot part to delete the secret from.
secret_id (str) – ID of the secret to delete.

Raises:

GRPCError – If an invalid robot part ID or secret ID is passed.

async list_robots(location_id: str | None = None) → List[viam.proto.app.Robot][source]

Get a list of all robots under the specified location.

list_of_machines = await cloud.list_robots(location_id="123ab12345")

Parameters:: location_id (Optional[str]) – ID of the location to retrieve the robots from. Defaults to the location ID provided at AppClient instantiation.
Raises:: GRPCError – If an invalid location ID is passed or one isn’t passed and there was no location ID provided at AppClient instantiation.
Returns:: The list of robots.
Return type:: List[viam.proto.app.Robot]

async new_robot(name: str, location_id: str | None = None) → str[source]

Create a new robot.

new_machine_id = await cloud.new_robot(name="beepboop")

Parameters:

name (str) – Name of the new robot.
location_id (Optional[str]) – ID of the location under which to create the robot. Defaults to the current authorized location.

Raises:

GRPCError – If an invalid location ID is passed or one isn’t passed and there was no location ID provided at AppClient instantiation.

Returns:

The new robot’s ID.

Return type:

str

async update_robot(robot_id: str, name: str, location_id: str | None = None) → viam.proto.app.Robot[source]

Change the name of an existing robot.

updated_robot = await cloud.update_robot(
    robot_id="1a123456-x1yz-0ab0-a12xyzabc",
    name="Orange-Robot")

Parameters:

robot_id (str) – ID of the robot to update.
name (str) – New name to be updated on the robot.
location_id (Optional[str]) – ID of the location under which the robot exists. Defaults to the location ID provided at AppClient instantiation

Raises:

GRPCError – If either an invalid robot ID, name, or location ID is passed or a location ID isn’t passed and there was no location ID provided at AppClient instantiation.

Returns:

The newly updated robot.

Return type:

viam.proto.app.Robot

async delete_robot(robot_id: str) → None[source]

Delete the specified robot.

await cloud.delete_robot(robot_id="1a123456-x1yz-0ab0-a12xyzabc")

Parameters:: robot_id (str) – ID of the robot to delete.
Raises:: GRPCError – If an invalid robot ID is passed.

async list_fragments(show_public: bool = True) → List[Fragment][source]

Get a list of fragments under the currently authed-to organization.

fragments_list = await cloud.list_fragments(show_public=False)

Parameters:: show_public – Optional boolean specifying whether or not to only show public fragments. If True, only public fragments will return. If False, only private fragments will return. Defaults to True.
Returns:: The list of fragments.
Return type:: List[viam.app.app_client.Fragment]

async get_fragment(fragment_id: str) → Fragment[source]

Get a fragment.

# Get a fragment and print its name and when it was created.
the_fragment = await cloud.get_fragment(
    fragment_id="12a12ab1-1234-5678-abcd-abcd01234567")
print("Name: ", the_fragment.name, "\nCreated on: ", the_fragment.created_on)

Parameters:: fragment_id (str) – ID of the fragment to get.
Raises:: GRPCError – If an invalid fragment ID is passed.
Returns:: The fragment.
Return type:: viam.app.app_client.Fragment

async create_fragment(name: str, config: Mapping[str, Any] | None = None) → Fragment[source]

Create a new private fragment.

new_fragment = await cloud.create_fragment(
    name="cool_smart_machine_to_configure_several_of")

Parameters:

name (str) – Name of the fragment.
config (Optional[Mapping[str, Any]]) – Optional Dictionary representation of new config to assign to specified fragment. Can be assigned by updating the fragment.

Raises:

GRPCError – If an invalid name is passed.

Returns:

The newly created fragment.

Return type:

viam.app.app_client.Fragment

async update_fragment(fragment_id: str, name: str, config: Mapping[str, Any] | None = None, public: bool | None = None) → Fragment[source]

Update a fragment name AND its config and/or visibility.

updated_fragment = await cloud.update_fragment(
    fragment_id="12a12ab1-1234-5678-abcd-abcd01234567",
    name="better_name")

Parameters:

fragment_id (str) – ID of the fragment to update.
name (str) – New name to associate with the fragment.
config (Optional[Mapping[str, Any]]) – Optional Dictionary representation of new config to assign to specified fragment. Not passing this parameter will leave the fragment’s config unchanged.
public (bool) – Boolean specifying whether the fragment is public. Not passing this parameter will leave the fragment’s visibility unchanged. A fragment is private by default when created.

Raises:

GRPCError – if an invalid ID, name, or config is passed.

Returns:

The newly updated fragment.

Return type:

viam.app.app_client.Fragment

async delete_fragment(fragment_id: str) → None[source]

Delete a fragment.

await cloud.delete_fragment(
    fragment_id="12a12ab1-1234-5678-abcd-abcd01234567")

Parameters:: fragment_id (str) – ID of the fragment to delete.
Raises:: GRPCError – If an invalid fragment ID is passed.

async add_role(identity_id: str, role: Literal[owner] | Literal[operator], resource_type: Literal[organization] | Literal[location] | Literal[robot], resource_id: str) → None[source]

Add a role under the currently authed-to organization.

await cloud.add_role(
    identity_id="abc01234-0123-4567-ab12-a11a00a2aa22",
    role="owner",
    resource_type="location",
    resource_id="111ab12345")

Parameters:

identity_id (str) – ID of the entity the role belongs to (e.g., a user ID).
role (Union[Literal["owner"], Literal["operator"]]) – The role to add.
resource_type (Union[Literal["organization"], Literal["location"], Literal["robot"]]) – Type of the resource to add role to. Must match resource_id.
resource_id (str) – ID of the resource the role applies to (i.e., either an organization, location, or robot ID).

Raises:

GRPCError – If either an invalid identity ID, role ID, resource type, or resource ID is passed.

async remove_role(identity_id: str, role: Literal[owner] | Literal[operator], resource_type: Literal[organization] | Literal[location] | Literal[robot], resource_id: str) → None[source]

Remove a role under the currently authed-to organization.

await cloud.remove_role(
    identity_id="abc01234-0123-4567-ab12-a11a00a2aa22",
    role="owner",
    resource_type="location",
    resource_id="111ab12345")

Parameters:

identity_id (str) – ID of the entity the role belongs to (e.g., a user ID).
role (Union[Literal["owner"], Literal["operator"]]) – The role to add.
resource_type (Union[Literal["organization"], Literal["location"], Literal["robot"]]) – Type of the resource to add role to. Must match resource_id.
resource_id (str) – ID of the resource the role applies to (i.e., either an organization, location, or robot ID).

Raises:

GRPCError – If either an invalid identity ID, role ID, resource type, or resource ID or is passed.

async list_authorizations(resource_ids: List[str] | None = None) → List[viam.proto.app.Authorization][source]

List all authorizations under a specific resource (or resources) within the currently authed-to organization. If no resource IDs are provided, all resource authorizations within the organizations are returned.

list_of_auths = await cloud.list_authorizations(
    resource_ids=["1a123456-x1yz-0ab0-a12xyzabc"])

Parameters:: resource_ids (Optional[List[str]]) – IDs of the resources to retrieve authorizations from. If None, defaults to all resources.
Raises:: GRPCError – If an invalid resource ID is passed.
Returns:: The list of authorizations.
Return type:: List[viam.proto.app.Authorization]

async check_permissions(permissions: List[viam.proto.app.AuthorizedPermissions]) → List[viam.proto.app.AuthorizedPermissions][source]

Checks validity of a list of permissions.

from viam.proto.app import AuthorizedPermissions

# Check whether the entity you're currently authenticated to has permission to control and/or
# read logs from robots in the "organization-identifier123" org
permissions = [AuthorizedPermissions(resource_type="organization",
                                     resource_id="organization-identifier123",
                                     permissions=["control_robot",
                                                  "read_robot_logs"])]

filtered_permissions = await cloud.check_permissions(permissions)

Parameters:: permissions (List[viam.proto.app.AuthorizedPermissions]) – the permissions to validate (e.g., “read_organization”, “control_robot”)
Raises:: GRPCError – If the list of permissions to validate is empty.
Returns:: The permissions argument, with invalid permissions filtered out.
Return type:: List[viam.proto.app.AuthorizedPermissions]

async create_module(name: str) → Tuple[str, str][source]

Create a module under the currently authed-to organization.

new_module = await cloud.create_module(name="cool_new_hoverboard_module")
print("Module ID:", new_module[0])

Parameters:: name (str) – The name of the module. Must be unique within your organization.
Raises:: GRPCError – If an invalid name (e.g., “”) is passed.
Returns:: A tuple containing the ID [0] of the new module and its URL [1].
Return type:: Tuple[str, str]

async update_module(module_id: str, url: str, description: str, models: List[viam.proto.app.Model] | None, entrypoint: str, organization_id: str | None = None, public: bool = False) → str[source]

Update the documentation URL, description, models, entrypoint, and/or the visibility of a module.

url_of_my_module = await cloud.update_module(
    module_id="my-group:cool_new_hoverboard_module",
    url="https://docsformymodule.viam.com",
    description="A base to support hoverboards.",
    entrypoint="exec")

Parameters:

module_id (str) – ID of the module being updated, containing module name (e.g., “my-module”) or namespace and module name (e.g., “my-org:my-module”).
url (str) – The url to reference for documentation and code (NOT the url of the module itself).
description (str) – A short description of the module that explains its purpose.
models (Optional[List[viam.proto.app.Model]]) – list of models that are available in the module.
entrypoint (str) – The executable to run to start the module program.
public (bool) – The visibility that should be set for the module. Defaults to False (private).

Raises:

GRPCError – If either an invalid module ID, URL, list of models, or organization ID is passed.

Returns:

The URL of the newly updated module.

Return type:

str

async upload_module_file(module_file_info: viam.proto.app.ModuleFileInfo | None, file: bytes) → str[source]

Upload a module file

file_id = await cloud.upload_module_file(file=b"<file>")

Parameters:

module_file_info (Optional[viam.proto.app.ModuleFileInfo]) – Relevant metadata.
file (bytes) – Bytes of file to upload.

Returns:

ID of uploaded file.

Return type:

str

async get_module(module_id: str) → viam.proto.app.Module[source]

Get a module.

the_module = await cloud.get_module(module_id="my-cool-modular-base")

Parameters:: module_id (str) – ID of the module being retrieved, containing module name or namespace and module name.
Raises:: GRPCError – If an invalid module ID is passed.
Returns:: The module.
Return type:: viam.proto.app.Module

async list_modules() → List[viam.proto.app.Module][source]

List the modules under the currently authed-to organization.

modules_list = await cloud.list_modules()

Returns:: The list of modules.
Return type:: List[viam.proto.app.Module]

async create_key(authorizations: List[APIKeyAuthorization], name: str | None = None) → Tuple[str, str][source]

Creates a new API key.

from viam.app.app_client import APIKeyAuthorization

auth = APIKeyAuthorization(
role="owner",
resource_type="robot",
resource_id="your-robot-id123"
)

api_key, api_key_id = cloud.create_key([auth], "my_key")

Parameters:

authorizations (List[viam.proto.app.Authorization]) – A list of authorizations to associate with the key.
name (Optional[str]) – A name for the key. If None, defaults to the current timestamp.

Raises:

GRPCError – If the authorizations list is empty.

Returns:

The api key and api key ID.

Return type:

Tuple[str, str]

async create_key_from_existing_key_authorizations(id: str) → Tuple[str, str][source]

Creates a new API key with an existing key’s authorizations

api_key, api_key_id = cloud.create_key_from_existing_key_authorizations(
    id="INSERT YOUR API KEY ID")

Parameters:: id (str) – the ID of the API key to duplication authorizations from
Returns:: The API key and API key id
Return type:: Tuple[str, str]

async list_keys() → List[viam.proto.app.APIKeyWithAuthorizations][source]

Lists all keys for the currently-authed-to org.

keys = cloud.list_keys()

Returns:: The existing API keys and authorizations.
Return type:: List[viam.proto.app.APIKeyWithAuthorizations]

viam.app.app_client

Module Contents

Classes

Attributes

`viam.app.app_client`