Client API reference

Importing the Client API 

The Client API is exported by the radkit_client.sync module. All the useful functions and classes can be directly imported from that module; there should be no need to import from specific submodules of radkit_client.sync, except in specific situations which are documented below.

Note

radkit_client exposes a synchronous user-facing API, but internally all its APIs are asynchronous, hence the sync/async terminology. The async API is no longer exposed or documented as of RADKit 1.7.0 due to its complexity and the absence of user-facing use cases that actually require it.

The radkit_client top-level module also exposes the synchronous API for ease of use and for backwards compatibility. In other words, the following two imports are equivalent:

>>> from radkit_client import Client as a
>>> from radkit_client.sync import Client as b
>>> a is b
True

Basic settings 

These functions are meant to be used in stand-alone scripts. Do not use them in the Client REPL.

radkit_common.settings.init_settings( settings_file: Path | None = None, *, domain: str | None = None, log_level: LogLevel | None = None, extra_settings: Iterable[tuple[str, str]] | None = None, ) → None

Initialize the Service settings context variable.

Merge the settings from the file with CLI-provided.

radkit_client.sync.change_settings( settings: dict[str, Any] | str, ) → None

Changes one or more Client settings (see Settings reference for details).

Some settings may only take effect if they are changed before any other RADKit Client API functions are called. It is recommended to change settings to the desired values at the very start of your script.

Parameters:: settings – dictionary, JSON string or TOML string in one of the formats shown below.

Python dict or JSON string:

{
    "client": {
        "log_file": "file.log"
    },
    "cloud_client": {
        "key1": "value1",
        "key2": "value2"
    }
}

TOML syntax (see Settings management):

[client]
log_file = "file.log"

[cloud_client]
key1 = "value1"
key2 = "value2"

Creation of a Client 

Almost anything we want to do starts with the creation of a Client. This is done automatically in the radkit-client REPL, but for stand-alone scripts, we would have to do this first:

from radkit_client.sync import Client

with Client.create() as client:
    ...

Many operations, like logging in, connecting to services, enrolling, etc… are started by calling a method on this Client instance.

In the radkit-client REPL, the client is instantiated automatically, and exposed using the client variable. Most methods of this client instance are also directly exposed. E.g., in the REPL, we can simply type sso_login() instead of client.sso_login().

Basic types 

class radkit_client.sync.Identity

Identity can be imported from radkit_client.sync. It is a typing.NewType that is in fact an alias for str, i.e. you can simply pass a string instead and it will work, but if you are using mypy, ruff or similar tools to typecheck your code, then you need to import and use Identity to comply with the API’s type annotations, for example:

from radkit_client.sync import Identity, Client
with Client.create() as client:
    client.sso_login(Identity("id@example.com"))

If you are not sure what mypy is or how type checking works in Python, then you can safely treat Identity as a str in your code.

class radkit_client.sync.CustomSecretStr

Bases: SecretStr

Similar to Pydantic’s SecretStr, except that it doesn’t leak information when a password is empty.

>>> from radkit_client import CustomSecretStr
>>> from pydantic import SecretStr

>>> CustomSecretStr('test')
SecretStr('**********')

>>> CustomSecretStr('')
SecretStr('**********')

>>> SecretStr('test')
SecretStr('**********')

>>> SecretStr('')
SecretStr('')

class radkit_client.sync.Domain

Bases: BaseModel

Set of parameters for connecting to a RADKit Cloud instance (for connecting to a radkit-access or a set of radkit-forwarder servers).

(This is a BaseModel so that we can deserialize it from a RADKIT_DOMAIN_DEFINITION. This is frozen, because Domain needs to be hashable.)

forwarder_address_objects() → list[WebSocketForwarderAddress]: Get forwarder addresses objects as required by the forwadrer client.

Login 

Logging in to a cloud domain is done by calling a login method of the Client instance.

In the radkit-client REPL, these login functions are directly exposed under their names, but under the hood, they are actually methods of a Client instance that was automatically instantiated.

A client can be connected to multiple domains at the same time and reach services from all connected domains.

>>> sso_login('user@cisco.com', domain='PROD')
...

>>> sso_login('user@cisco.com', domain='DEVEL')
...

>>> client
[CONNECTED] Client(status='CONNECTED')
-----------------  -------------------------------------------------------------------------------------------------
connected_domains  PROD (...@cisco.com; OIDC auth; authenticated ); DEVEL (...@cisco.com; OIDC auth; authenticated )
services           not connected to any service
running_proxies    no local proxy running
-----------------  -------------------------------------------------------------------------------------------------

Log in to RADKit Cloud using SSO.

Parameters:

identity – (optional) the Client ID (owner email address) to use as SSO identity
domain – (optional) the RADKit Cloud domain to connect to
open_browser – (optional) open the SSO login page in a browser instead of displaying it (default: False on Linux, True on other platforms)
provider – (optional) specify a different OAuth provider name than the one set in the domain
oauth_connect_response – (optional) OAuthConnectResponse object obtained separately to perform OAuth
admin_level – (optional) the admin level to use for the session, defaults to 0.
silent – (optional) if True, suppresses all output from the SSO login process

Log in to RADKit Cloud using a Client certificate. This requires to have previously enrolled the Client.

Parameters:

identity – (optional) the Client ID (owner email address) present in the certificate
ca_path – (optional) the path to the issuer chain for the identity certificate
cert_path – (optional) the path to the identity certificate
key_path – (optional) the path to the private key for the identity certificate
private_key_password – (optional) the password that unlocks the private key
domain – (optional) the RADKit Cloud domain to connect to
admin_level – (optional) the admin level to use for the session, defaults to 0.

Client.access_token_login( access_token: str, domain: Domain | str | None = None, provider: OAuthProvider | None = None, admin_level: int = 0, ) → Self

Log in to RADKit Cloud using an Access Token from a supported OAuth provider.

Parameters:

access_token – an externally obtained OAuth Token
domain – (optional) the RADKit Cloud domain to connect to
provider – (optional) the token provider name
admin_level – (optional) the admin level to use for the session, defaults to 0.

As a result of calling any of these login functions, a cloud connection will appear in the client.cloud_connections attribute.

Connecting to a service 

Once we have a Client instance, we can use that to connect it to services. If the client is authenticated to a cloud domain, we can connect services from that cloud domain. But we can also directly connect to services or start an integrated service and connect to that.

Following connect methods are methods of the Client class, but all of them are also exposed directly under their name in the radkit-client REPL. All of these connect methods return a Service instance.

Parameters:

service_id – the Service ID to connect to.
connection – Any connection from the client.cloud_connections dictionary. If not specified, the default connection will be used.
e2ee_fingerprint – Sha256 fingerprint for verifying the E2EE certificate. If given, E2EE will always be used.
access_token – Token to be passed for the service (within an E2EE) connection so that the service can verify the end-to-end encrypted connection.
name – Register the service using the given name in the .services dictionary, or when not specified, use the service_id instead. When a name is already taken, then a new name is generated by adding a suffix.

Establish a direct (cloud-less) connection to a service.

Parameters:: rpc_url – When passed, connect to the legacy RPC endpoint for direct RPC. E.g., “https://localhost:8081/rpc”. In this case the host and port parameters are ignored. Also note that for legacy direct RPC, an admin username/password is required instead of a “remote user”.

Client.service_direct_with_sso( service_id: ServiceID | str, *, connection: CloudConnection | None = None, e2ee_fingerprint: str | None = None, host: str = localhost, port: int = 8181, name: str | None = None, ) → Service

Connect directly to a service using the cloud first to authenticate.

This will first establish a connection to the service through the cloud, requiring SSO for authorization. Then using cloud-RPC we will obtain an access token from the service that’s going to be used for a direct RPC connection that’s going to be used for all consecutive requests.

Parameters:

service_id – the Service ID to connect to.
connection – Any connection from the client.cloud_connections dictionary. If not specified, the default connection will be used.
e2ee_fingerprint – Sha256 fingerprint for verifying the E2EE certificate. If given, E2EE will always be used.
host – Hostname or IP address for the direct connection.
port – Port number of the RPC server of the service.
name – If given, register the service using this name in the .services dictionary. Otherwise, use the service_id instead, or generate a name if this is already taken.

Client.service_integrated(

*_: object,

**__: object,

) → None

>>> sso_login('user@cisco.com', domain='PROD')
>>> service = service_cloud('serial').wait()

Methods like service_cloud() take additional arguments like the domain and parameters for things like end-to-end encryption.

All the connected services can be looked up in the services dictionary where the key corresponds to the name attribute of the connected service.

OAuth helper function 

In situations where the client application using the RADKit Client API needs to drive the OAuth exchange instead of simply delegating it to sso_login(), use the oauth_connect_only() function alongside sso_login() to complete the OAuth exchange. It returns an OAuthConnectResponse object which contains the sso_url the user needs to visit in order to complete the OAuth exchange.

The client application should handle redirecting the user to sso_url which would put them through the login flow. Note that the at the moment, the login flow does not redirect back to the client application, i.e., the client application needs to handle redirecting the user back to itself.

The OAuthConnectResponse object itself can be passed to sso_login() to complete the OAuth exchange and obtain a Client object that is capable of automatically refreshing the token when it expires, if the OAuth backend supports it.

Requests the data needed to perform a RADKit Cloud OAuth login for the specified identity (user email address).

Example script:

CLIENT_ID = "id@example.com"
SERVICE_ID = "abcd-ddsa-zb12"

with Client.create() as client:
    connect_data = client.oauth_connect_only(CLIENT_ID)

    # The `print` here is an example, the intent is to use `sso_url` to point the
    # user to the authentication flow and simultaneously open the web socket awaiting
    # the token to be set.
    print("Open in a browser:", connect_data.sso_url)

    client = client.sso_login(CLIENT_ID, oauth_connect_response=connect_data)
    service = client.service(SERVICE_ID).wait()

    # All operations on the Client and Service objects must be performed inside
    # the `with` block; when it is exited, the context will terminate and
    # invalidate all active Client and Service objects.

The user must be asked to open the sso_url in a browser and follow the OAuth/SSO authentication process. In parallel, call sso_login() passing the same parameters that were passed to oauth_connect_only(), plus the OAuthConnectResponse object returned by oauth_connect_only(). Then sso_login() will open a websocket awaiting the SSO result to be confirmed by RADKit Cloud.

Parameters:

identity – optional – the Client ID (owner email address) to use as SSO identity.
domain – optional – a string or object representing the RADKit domain to connect to (default: PROD).
provider – optional – use a different OAuth provider than the one set for the domain. The provider must be known to the RADKit Access backend.

Returns:

an OAuthConnectResponse instance with sso_url and token_url attributes.

pydantic model radkit_client.sync.OAuthConnectResponse

Bases: BaseModel

Represents the data returned by the RADKit Cloud in response to an OAuth connect request, which is a request by a client to authenticate using OAuth.

field sso_url: HttpUrl [Required]: The HTTPS URL that must be opened in a browser to perform OAuth authentication.

field token_url: WebsocketUrl [Required]: The WSS (secure websocket) URL that must be accessed to retrieve the token resulting from a successful OAuth exchange.

field max_session_ttl: int [Required]

field provider: OAuthProvider [Required]

property ttl: float: The timeout value (in seconds) that should be set on the websocket awaiting the token on token_url.

Client 

class radkit_client.sync.Client

Bases: SyncWrapper[AsyncClient]

The main RADKit client class.

Usage:

async with Client.create() as client:
    ...

forwarders

Property type: ForwardersDict

The status of all Forwarder connections for this Client instance.

socks_proxy

Property type: ProxyPortForwarder | None

Returns the active SOCKS proxy for this Client instance, or None if there is no active proxy.

http_proxy

Property type: ProxyPortForwarder | None

Returns the active HTTP proxy for this Client instance, or None if there is no active proxy.

status

Property type: ClientStatus

Returns the status of this Client instance, as a ClientStatus enum value.

cloud_connections

Property type: CloudConnectionsDict

All the cloud connections for this client.

services

Property type: ServicesDict

All Service instances for this Client instance, indexed by serial.

requests

Property type: RequestsDict

All Request instances for this Client instance, enumerated in order.

session_logs

Property type: SessionLogsDict

All terminal/SCP session logs that were created for all the sessions for the devices in this Client instance. (This includes sessions that are terminated.)

sessions

Property type: SessionsDict

Returns all terminal/SCP/SFTP sessions that were established for this Client instance. This includes sessions that are terminated.

port_forwards

Property type: TCPPortForwardsDict

Returns all open direct port forwardings for this Client instance (SOCKS/HTTP proxy excluded).

set_default_domain( domain: Domain | str, ) → None

Sets the default RADKit domain for all the login functions. This can also be changed through the client.default_domain setting.

Parameters:: domain – a string or object representing the RADKit domain to connect to.

get_default_domain() → Domain: Get the default RADKit domain for all the login functions.

start_http_proxy( local_port: int, local_address: str = localhost, username: str | None = None, password: str | None = None, max_connections: int | None = 50, server_startup_timeout: float = 2.0, destroy_previous: bool = False, ) → ProxyPortForwarder

Starts an HTTP proxy. By default, the listener will bind to localhost on the specified port. A username and password can be set to protect the proxy (this is strongly recommended, particularly if the proxy listener is reachable through the network, or if the local host is a multi-user machine).

The proxy will run until either stop_http_proxy() is called or the returned ProxyPortForwarder object is explicitly stopped.

If a HTTP proxy already exists for this client, calling this method will raise ProxyAlreadyStartedError. To start a fresh HTTP proxy even if one was already running, pass destroy_previous=True.

Parameters:

local_port – the local port for the proxy listener (you may need elevated permissions in order to open ports 1023 and below).
local_address – optional – the local address for the proxy listener (default: localhost).
username – optional – set a username to authenticate to the proxy locally.
password – optional – set a password to authenticate to the proxy locally.
max_connections – optional – the maximum number of concurrent connections to the proxy (default: 50).
server_startup_timeout – optional – how many seconds to wait for the listener to start before timing out (default: 2).
destroy_previous – optional – specifies if an existing proxy instance should be destroyed (default: False).

Raises:

ProxyAlreadyStartedError – if a proxy of the same type is already started.

stop_http_proxy() → None: Stops the active HTTP proxy, if there is one. This is equivalent to calling stop() on the ProxyPortForwarder object returned by start_http_proxy().

start_socks_proxy( local_port: int, local_address: str = localhost, username: str | None = None, password: str | None = None, max_connections: int | None = 50, server_startup_timeout: float = 2.0, destroy_previous: bool = False, ) → ProxyPortForwarder

Starts a SOCKS v5 proxy. By default, the listener will bind to localhost on the specified port. A username and password can be set to protect the proxy (this is strongly recommended, particularly if the proxy listener is reachable through the network, or if the local host is a multi-user machine).

The proxy will run until either stop_socks_proxy() is called or the returned ProxyPortForwarder object is explicitly stopped.

If a SOCKS proxy already exists for this client, calling this method will raise ProxyAlreadyStartedError. To start a fresh SOCKS proxy even if one was already running, pass destroy_previous=True.

Parameters:

local_port – the local port for the proxy listener (you may need elevated permissions in order to open ports 1023 and below).
local_address – optional – the local address for the proxy listener (default: localhost).
username – optional – set a username to authenticate to the proxy locally.
password – optional – set a password to authenticate to the proxy locally.
max_connections – optional – the maximum number of concurrent connections to the proxy (default: 50).
server_startup_timeout – optional – how many seconds to wait for the listener to start before timing out (default: 2).
destroy_previous – optional – specifies if an existing proxy instance should be destroyed (default: False).

Raises:

ProxyAlreadyStartedError – if a proxy of the same type is already started.

stop_socks_proxy() → None: Stops the active SOCKS v5 proxy, if there is one. This is equivalent to calling stop() on the ProxyPortForwarder object returned by start_socks_proxy().

start_ssh_proxy( local_port: int, local_address: str = localhost, password: str | None = None, host_key: bytes | None = None, destroy_previous: bool = False, ) → ProxySshForwarder: TODO

stop_ssh_proxy() → None: Stops the active SSH proxy, if there is one. This is equivalent to calling stop() on the ProxySshForwarder object returned by start_ssh_proxy().

Log in to RADKit Cloud using SSO.

Parameters:

identity – (optional) the Client ID (owner email address) to use as SSO identity
domain – (optional) the RADKit Cloud domain to connect to
open_browser – (optional) open the SSO login page in a browser instead of displaying it (default: False on Linux, True on other platforms)
provider – (optional) specify a different OAuth provider name than the one set in the domain
oauth_connect_response – (optional) OAuthConnectResponse object obtained separately to perform OAuth
admin_level – (optional) the admin level to use for the session, defaults to 0.
silent – (optional) if True, suppresses all output from the SSO login process

Log in to RADKit Cloud using SSO.

Parameters:

identity – (optional) the Client ID (owner email address) to use as SSO identity
domain – (optional) the RADKit Cloud domain to connect to
open_browser – (optional) open the SSO login page in a browser instead of displaying it (default: False on Linux, True on other platforms)
provider – (optional) specify a different OAuth provider name than the one set in the domain
oauth_connect_response – (optional) OAuthConnectResponse object obtained separately to perform OAuth
admin_level – (optional) the admin level to use for the session, defaults to 0.
silent – (optional) if True, suppresses all output from the SSO login process

basic_login( identity: EndpointID | str | None = None, password: str | None = None, domain: Domain | str | None = None, admin_level: int = 0, ) → Self: Basic authentication. For dev/test setups only. Cannot be enabled on AWS.

access_token_login( access_token: str, domain: Domain | str | None = None, provider: OAuthProvider | None = None, admin_level: int = 0, ) → Self

Log in to RADKit Cloud using an Access Token from a supported OAuth provider.

Parameters:

access_token – an externally obtained OAuth Token
domain – (optional) the RADKit Cloud domain to connect to
provider – (optional) the token provider name
admin_level – (optional) the admin level to use for the session, defaults to 0.

Log in to RADKit Cloud using a Client certificate. This requires to have previously enrolled the Client.

Parameters:

identity – (optional) the Client ID (owner email address) present in the certificate
ca_path – (optional) the path to the issuer chain for the identity certificate
cert_path – (optional) the path to the identity certificate
key_path – (optional) the path to the private key for the identity certificate
private_key_password – (optional) the password that unlocks the private key
domain – (optional) the RADKit Cloud domain to connect to
admin_level – (optional) the admin level to use for the session, defaults to 0.

logout( domain: Domain | str | None = None, client_id: ClientID | str | None = None, ) → None

Do a logout from the given domain.

Parameters:: client_id – If given, only log out for the given domain/client_id combination. If None log out for all client_ids for this domain.

Requests the data needed to perform a RADKit Cloud OAuth login for the specified identity (user email address).

Example script:

CLIENT_ID = "id@example.com"
SERVICE_ID = "abcd-ddsa-zb12"

with Client.create() as client:
    connect_data = client.oauth_connect_only(CLIENT_ID)

    # The `print` here is an example, the intent is to use `sso_url` to point the
    # user to the authentication flow and simultaneously open the web socket awaiting
    # the token to be set.
    print("Open in a browser:", connect_data.sso_url)

    client = client.sso_login(CLIENT_ID, oauth_connect_response=connect_data)
    service = client.service(SERVICE_ID).wait()

    # All operations on the Client and Service objects must be performed inside
    # the `with` block; when it is exited, the context will terminate and
    # invalidate all active Client and Service objects.

The user must be asked to open the sso_url in a browser and follow the OAuth/SSO authentication process. In parallel, call sso_login() passing the same parameters that were passed to oauth_connect_only(), plus the OAuthConnectResponse object returned by oauth_connect_only(). Then sso_login() will open a websocket awaiting the SSO result to be confirmed by RADKit Cloud.

Parameters:

identity – optional – the Client ID (owner email address) to use as SSO identity.
domain – optional – a string or object representing the RADKit domain to connect to (default: PROD).
provider – optional – use a different OAuth provider than the one set for the domain. The provider must be known to the RADKit Access backend.

Returns:

an OAuthConnectResponse instance with sso_url and token_url attributes.

Get the OAuth provider information for a given domain.

Parameters:

user_domain – the domain to query.
provider – the provider to query.
tool – the tool to query.
domain – the domain to connect to.

service_from_rpc_target( rpc_target: RPCTarget, name: str | None = None, ) → Service: Connect a service using a given RPC target.

Establish a direct (cloud-less) connection to a service.

Parameters:: rpc_url – When passed, connect to the legacy RPC endpoint for direct RPC. E.g., “https://localhost:8081/rpc”. In this case the host and port parameters are ignored. Also note that for legacy direct RPC, an admin username/password is required instead of a “remote user”.

enroll_client_from_otp( otp_access_token: str | None = None, private_key_password: str | None = None, overwrite_certificate: bool | None = None, ) → None | PromptInterrupted

Enrolls a RADKit Client using a one-time password (OTP) issued by a privileged RADKit user. This is only for specific situations; if you don’t know whether you need this, then you most likely don’t – just use the .enroll() method on an authenticated Client object and you should be fine.

Parameters:

otp_access_token – optional – the OTP received from the RADKit admin (default: prompt the user).
private_key_password – optional – the password to protect the private key generated during enrollment (it is not recommended to provide the password as an argument, as it may find its way into the history or scrollback buffer; please favor user interaction, which is the default).
overwrite_certificate – optional – if True, any existing certificate/keypair for this client identity will be overwritten without prompting; if False, enrollment will abort if a certificate/keypair already exist; if None, the user will be prompted whether to overwrite if needed.

read_identity_certificate( identity: EndpointID | str, domain: Domain | str | None = None, ) → tuple[str, str]

Reads and returns the contents of the Client identity certificate and keypair files for the given identity and domain. The private key remains encrypted with the passphrase chosen by the user.

This is only supposed to be used in conjunction with write_identity_certificate().

Parameters:

identity – the Client ID (email address).
domain – optional – the RADKit Cloud domain as a string or object. If omitted, the domain is chosen based on the current Client settings.

Returns:

a 2-element tuple of the certificate and keypair as PEM strings.

write_identity_certificate( identity: EndpointID | str, cert: str, keypair: str, domain: Domain | str | None = None, ) → None

Writes a Client identity certificate and associated keypair to the local identities store for the given identity and domain. The private key should already be encrypted with the passphrase chosen by the user.

This function does not perform any checks, in particular, it does not verify that the cert and keypair actually match, or that the identity in the certificate matches the one passed as argument, or even that the PEM data is valid.

This is only supposed to be used in conjunction with read_identity_certificate().

Parameters:

identity – the Client ID (email address).
cert – the certificate as a PEM string.
keypair – the keypair as a PEM string.
domain – optional – the RADKit Cloud domain as a string or object. If omitted, the domain is chosen based on the current Client settings.

integrations: Property type: Integrations

support_package() → ClientSupportPackage: Generate support information for troubleshooting activity on this client.

Connects to a RADKit Service, given its Service ID (aka serial number). This returns a Service instance. If there is already a connection to the requested Service ID, the existing instance is returned.

Parameters:

service_id – the Service ID to connect to.
connection – Any connection from the client.cloud_connections dictionary. If not specified, the default connection will be used.
e2ee_fingerprint – Sha256 fingerprint for verifying the E2EE certificate. If given, E2EE will always be used.
access_token – Token to be passed for the service (within an E2EE) connection so that the service can verify the end-to-end encrypted connection.
name – Register the service using the given name in the .services dictionary, or when not specified, use the service_id instead. When a name is already taken, then a new name is generated by adding a suffix.

Parameters:

service_id – the Service ID to connect to.
connection – Any connection from the client.cloud_connections dictionary. If not specified, the default connection will be used.
e2ee_fingerprint – Sha256 fingerprint for verifying the E2EE certificate. If given, E2EE will always be used.
access_token – Token to be passed for the service (within an E2EE) connection so that the service can verify the end-to-end encrypted connection.
name – Register the service using the given name in the .services dictionary, or when not specified, use the service_id instead. When a name is already taken, then a new name is generated by adding a suffix.

service_direct_with_sso( service_id: ServiceID | str, *, connection: CloudConnection | None = None, e2ee_fingerprint: str | None = None, host: str = localhost, port: int = 8181, name: str | None = None, ) → Service

Connect directly to a service using the cloud first to authenticate.

This will first establish a connection to the service through the cloud, requiring SSO for authorization. Then using cloud-RPC we will obtain an access token from the service that’s going to be used for a direct RPC connection that’s going to be used for all consecutive requests.

Parameters:

service_id – the Service ID to connect to.
connection – Any connection from the client.cloud_connections dictionary. If not specified, the default connection will be used.
e2ee_fingerprint – Sha256 fingerprint for verifying the E2EE certificate. If given, E2EE will always be used.
host – Hostname or IP address for the direct connection.
port – Port number of the RPC server of the service.
name – If given, register the service using this name in the .services dictionary. Otherwise, use the service_id instead, or generate a name if this is already taken.

create_service( *, suppress_logs: bool = False, headless: bool = False, superadmin_password: CustomSecretStr | str | None = None, create_extra_devices_backend: Callable[[], AbstractContextManager[DevicesBackend]] | None = None, ) → IntegratedService

Create a running service within this process, and connect to it with this client using an in-memory RPC transport.

The integrated service will run a service in the same way that radkit-service would run a service. However, keep in mind that the settings are those that were loaded within this radkit-client process, which is usually: ~/.radkit/client/settings.toml.

Example usage:

# Create the service.
integrated_service = create_service(...)

# Start connections.
integrated_service.start_cloud_connection()
integrated_service.start_webserver()

# Use the control API for managing the service.
print(integrated_service.control_api.list_admins())

# Interact with the service like usually from radkit-client:
print(integrated_service.service.wait())

Parameters:

headless – Don’t ask for a superadmin password interactively.
extra_devices_backend – Pass an extra implementation of a devices backend for the service to use. The service will combine the users/devices from the built-in database with those that are exposed through this devices backend.

Grants a one-time password to a RADKit Client to let it enroll with the RADKit Access CA. If a Client owner email is not provided, the current Client identity will be used. If a Client ID is not provided, the Client owner email will be used.

This operation may require specific privileges in RADKit Access.

Parameters:

client_id – optional – the email address that will be used as the identity of the Client being enrolled (this is the identity that will be found in the resulting certificate).
client_owner_email – optional – the email address of the owner of the Client being enrolled (this is not necessarily the identity that will be found in the resulting certificate).
connection – optional – the connection to use for enrollment. If not specified, the default connection will be used.
description – optional – a description for the client the otp is requested for.

Raises:

GrantingFailedError – if the OTP could not be granted.

Grants a one-time password to a RADKit Service to let it enroll with the RADKit Access CA. If a Service ID is not provided, one will be generated automatically by RADKit Access. If a Service owner email is not provided, the current Client identity will be used.

This operation may require specific privileges in RADKit Access.

Parameters:

service_id – optional – the ID of the Service to be enrolled.
service_owner_email – optional – the owner of the Service to be enrolled.
connection – optional – the connection to use for enrollment. If not specified, the default connection will be used.
description – optional – a description for the service the otp is requested for.

Raises:

GrantingFailedError – if the OTP could not be granted.

Enrolls this Client to make certificate-based authentication possible. By default, the Client ID and Client owner in the certificate will both be set to the currently active Client ID. If another Client ID is specified, specific permissions in RADKit Cloud may be necessary for the enrollment to succeed.

Warning

The private key and its passphrase are highly sensitive information. Their usage is strictly restricted to the owner of the keys, and CANNOT BE SHARED UNDER ANY CIRCUMSTANCES. Sharing those credentials exposes the owner to impersonation attacks. Please refer to Enrolling the Client for more information.

Parameters:

client_id – optional – the email address that will be used as the identity of the Client being enrolled.
private_key_password – optional – the password to protect the keypair (it is not recommended to provide the password as an argument, as it may find its way into the history or scrollback buffer; please favor user interaction, which is the default).
overwrite_certificate – optional – if True, any existing certificate/keypair for this client identity will be overwritten without prompting; if False, enrollment will abort if a certificate/keypair already exist; if None, the user will be prompted whether to overwrite if needed.
connection – optional – the connection to use for enrollment. If not specified, the default connection will be used.
description – optional – a description for the certificate endpoint.

Raises:

GrantingFailedError – if an OTP could not be granted to authorize the enrollment.
EnrollmentFailedError – if an error occurred during enrollment.
ValueError – if one of the parameters failed validation.

reauthenticate( access_token: str | None = None, basic_auth_password: str | None = None, oauth_connect_response: OAuthConnectResponse | None = None, connection: CloudConnection | None = None, ) → None

Re-authenticates this Client with the RADKit Cloud. Use this method when the authentication token has expired or got invalidated for some reason (e.g. the client was disconnected for too long). Unlike calling the login() function again, this makes it possible to keep all active Service and Device objects bound to this instance along with their state.

Parameters:

access_token – optional – a valid OAuth access token.
basic_auth_password – optional – a valid Basic HTTP authentication password (only usable in dev/test environments).
oauth_connect_response – optional – a valid OAuthConnectResponse object.

admin_client( connection: CloudConnection | None = None, ) → AdminClient

Create an ‘admin’ client.

Parameters:: connection – Any connection from the client.cloud_connections dictionary. If not specified, the default connection will be used.

start_background_job( func: Callable[[], Coroutine[Any, Any, None]], ) → None

Schedule an async background job in the event loop thread.

This background job will automatically be cancelled when this client terminates.

property authenticator: AuthFlow | None

Returns the authenticator object used to create this client.

This is useful for re-authenticating the client, checking the type of authenticator used to create the client, and for other introspection purposes.

Returns:: The authenticator object used to create this client, as an object of one of the following types: AccessTokenAuthFlow, BasicAuthFlow, CertificateAuthFlow, or OIDCAuthFlow.

class radkit_client.sync.ClientStatus

Bases: Enum

Enumeration of the possible Client statuses.

NOT_CONNECTED = 'NOT_CONNECTED': Not connected to any domain or service yet.

CONNECTED = 'CONNECTED': Connected to at least one domain or service. (Check domain/service state for details.)

TERMINATED = 'TERMINATED': Client got terminated. No new RPC requests can be instantiated.

class radkit_client.sync.ForwardersDict

Bases: SyncDictWrapper[AsyncForwardersDict, int, ForwarderInfo, ForwarderInfo]

RADKit Forwarder connections for a given Client instance. Keys are simple indices.

client

Property type: Client

The Client object that the connections in this mapping are bound to.

class radkit_client.sync.ServiceEnrollInfo

Bases: object

Service enrollment information for OTP-based enrollment.

email: Email: The Service owner email address.

serial: ServiceID: The Service serial number (this is now referred to as the Service ID).

otp: str: The enrollment one-time-password.

domain: Domain: The RADKit domain in which this OTP is valid, as a Domain object.

domain_name() → str: Returns the name of the RADKit domain in which this OTP is valid, as a str.

class radkit_client.sync.ClientEnrollInfo

Bases: object

Client enrollment information for OTP-based enrollment.

email: ClientID: The Client identity email address.

otp: str: The enrollment one-time-password.

domain: Domain: The RADKit domain in which this OTP is valid, as a Domain object.

domain_name() → str: Returns the name of the RADKit domain in which this OTP is valid, as a str.

class radkit_client.sync.ProxyAlreadyStartedError

Bases: ClientError

Raised when trying to start a proxy that is already started.

class radkit_client.sync.GrantingFailedError

Bases: ClientError

Something went wrong while granting the OTP.

class radkit_client.sync.EnrollmentFailedError

Bases: ClientError

An error occurred during enrollment.

Certificates 

Enrollment using OTP 

In most situations, the standard procedure to enroll a RADKit Client with the RADKit Access CA will be sufficient: first log in with sso_login(), then call enroll_client() on the Client object. For situations where this is not feasible however, it is also possible to enroll using a one-time password (OTP). This can be done without an authenticated Client object.

radkit_client.sync.Client.enroll_client_from_otp( otp_access_token: str | None = None, private_key_password: str | None = None, overwrite_certificate: bool | None = None, ) → None | PromptInterrupted

Enrolls a RADKit Client using a one-time password (OTP) issued by a privileged RADKit user. This is only for specific situations; if you don’t know whether you need this, then you most likely don’t – just use the .enroll() method on an authenticated Client object and you should be fine.

Parameters:

otp_access_token – optional – the OTP received from the RADKit admin (default: prompt the user).
private_key_password – optional – the password to protect the private key generated during enrollment (it is not recommended to provide the password as an argument, as it may find its way into the history or scrollback buffer; please favor user interaction, which is the default).
overwrite_certificate – optional – if True, any existing certificate/keypair for this client identity will be overwritten without prompting; if False, enrollment will abort if a certificate/keypair already exist; if None, the user will be prompted whether to overwrite if needed.

Export and import 

In scenarios where the RADKit Client is instantiated in a non-persistent environment such as a container, it may be useful to export and import the identity certificate and keypair using the two functions below.

radkit_client.sync.Client.read_identity_certificate( identity: EndpointID | str, domain: Domain | str | None = None, ) → tuple[str, str]

Reads and returns the contents of the Client identity certificate and keypair files for the given identity and domain. The private key remains encrypted with the passphrase chosen by the user.

This is only supposed to be used in conjunction with write_identity_certificate().

Parameters:

identity – the Client ID (email address).
domain – optional – the RADKit Cloud domain as a string or object. If omitted, the domain is chosen based on the current Client settings.

Returns:

a 2-element tuple of the certificate and keypair as PEM strings.

radkit_client.sync.Client.write_identity_certificate( identity: EndpointID | str, cert: str, keypair: str, domain: Domain | str | None = None, ) → None

Writes a Client identity certificate and associated keypair to the local identities store for the given identity and domain. The private key should already be encrypted with the passphrase chosen by the user.

This function does not perform any checks, in particular, it does not verify that the cert and keypair actually match, or that the identity in the certificate matches the one passed as argument, or even that the PEM data is valid.

This is only supposed to be used in conjunction with read_identity_certificate().

Parameters:

identity – the Client ID (email address).
cert – the certificate as a PEM string.
keypair – the keypair as a PEM string.
domain – optional – the RADKit Cloud domain as a string or object. If omitted, the domain is chosen based on the current Client settings.

Service 

class radkit_client.sync.Service

Bases: SyncWrapper[AsyncService]

Represents a connection to a RADKit Service via the RADKit Cloud.

reload( *, username: str =, e2ee_fingerprint: str | None | Keep = <keep>, access_token: str | None | Keep = <keep>, ) → Self

Reload both the capabilities and the inventory. Use the new E2EE fingerprint or access token if given.

Parameters:

e2ee_fingerprint – Sha256 fingerprint for verifying the E2EE certificate. If given, E2EE will always be used.
access_token – Access token for E2EE session verification to the remote service. This requires passing an e2ee_fingerprint as well.

status

Property type: ServiceStatus

The connection status of this Service. See ServiceStatus for details.

connection

Property type: CloudConnection | None

The cloud connection (client_id/domain) that’s used for reaching out to this service. Return None when this service is not connected through the cloud.

domain

Property type: Domain | None

The cloud domain through which this service is connected (if connected through the cloud, otherwise None).

domain_name

Property type: str | None

The name of the cloud domain through which this service is connected (if connected through the cloud, otherwise None).

target

Property type: RPCTarget

The RPC target for this Service.

name: Property type: str

client_id

Property type: ClientID | None

The identity of the Client instance that this Service instance is bound to.

service_id

Property type: ServiceID | None

The service ID of this Service.

loading_error

Property type: BaseException | None

Return the exception that was raised while loading the capabilities or inventory for this service, if any. Return None if no exception was raised.

For instance:

RPCCertificateVerificationError(“Incorrect e2ee certificate fingerprint found, rejecting the connection.”)
RequestError(“End-to-end session verification required.”).
RPCVerificationError(“E2EE session verification failed. Incorrect access token.”)

inventory

Property type: DeviceDict

The cached Device inventory of this Service. Update the cache using update_inventory().

logout() → None: Perform a logout for this service.

version

Property type: str | None

Returns the software version of the remote RADKit Service if known, or None otherwise. The capabilities need to be loaded in order to know this.

capabilities

Property type: ServiceCapabilities

The capabilities (list of RPC calls) supported by the remote Service. Update using update_capabilities().

e2ee_supported

Property type: bool

Returns True if this Service supports end-to-end encryption (E2EE). Returns False if E2EE is not supported, or if capabilities have not yet been loaded.

e2ee_active

Property type: bool

Returns True if end-to-end encryption (E2EE) is currently in use when communicating with this Service.

set_e2ee_credentials( *, username: str =, access_token: str | CustomSecretStr, ) → None

Set E2EE credentials for this service. If the provided credentials are different, then the service is invalidated and reloaded.

If this is a cloud connected service, the username can be empty, and will be taken from the cloud connection itself.

set_e2ee_fingerprint( e2ee_fingerprint: str | None, ) → None: Set E2EE fingerprint for this service. If the provided fingerprint is different, then the service is invalidated and reloaded.

supported_compression_methods

Property type: list[str] | None

Returns the RPC compression methods supported by this Service. Returns None if we did not yet get a capabilities response.

direct_rpc_url

Property type: str | None

The RPC URL that is used, in the case of a direct connection for the Client to this Service.

requests

Property type: RequestsDict

All RPC requests that have been performed against this Service so far.

session_logs

Property type: SessionLogsDict

All terminal/SCP session logs that were created for all the sessions for the devices in this Service so far. (This includes sessions that are terminated.)

sessions

Property type: SessionsDict

All Terminal/SCP/SFTP sessions that were established to this Service instance so far. (This includes sessions that are terminated.)

port_forwards

Property type: TCPPortForwardsDict

All open direct port forwardings for this Service instance (SOCKS/HTTP proxy excluded).

client

Property type: Client

Returns the Client object that this Service is bound to.

ping( *, count: int = 4, payload_size: int = 56, concurrent: int = 1, interval: float = 1.0, ) → None

Sends ping RPC messages to the remote Service. Useful as a liveness check and for measuring round-trip time (RTT).

Parameters:

count – optional – the number of pings to send (can be cancelled using ^C; default: 4).
payload_size – optional – size in bytes of the random payload that will be echoed (default: 56).
concurrent – optional – the number of concurrent threads sending pings (default: 1).
interval – optional – for every concurrent thread, the interval at which ping messages will be sent (default: 1.0).

get_e2ee_information() → E2EEInformation: Gets the E2EE information (including the X.509 certificate) from the Service by doing a ping RPC call.

wait( timeout: float | None = None, ) → Self

Wait for the remote Service capabilities and inventory to be loaded.

Parameters:: timeout – optional – timeout in seconds (default: wait forever).

update_inventory( update_capabilities_timeout: float | None = None, ) → SimpleRequest[p.GetBasicInventory, p.BasicInventoryResponse, InventoryUpdateResult]

Updates the inventory for this Service. This will also refresh the Service capabilities.

Parameters:: update_capabilities_timeout – optional – how long (in seconds) to wait for capabilities retrieval to complete (default: wait forever).

update_capabilities() → SimpleRequest[c.GetCapabilities, c.GetCapabilitiesResponse, ServiceCapabilities]: Updates the capabilities for this Service. The capabilities may change if the remote Service is upgraded/downgraded.

property serial: ServiceID | None: Deprecated – please use service_id instead.

class radkit_client.sync.ServiceCapabilities

Bases: SyncDictWrapper[AsyncServiceCapabilities, int, ServiceCapability, ServiceCapability]

Represents the RPC capabilities of a remote RADKit Service. Keys are indices, values are capability objects.

service_id

Property type: ServiceID | None

The service ID of the remote Service.

service_version

Property type: str | None

The version of the remote Service, or None if unknown.

status

Property type: ServiceLoadingStatus

The loading status of the remote Service capabilities.

wait( timeout: float | None = None, ) → Self

Wait for the remote Service capabilities to be loaded.

Parameters:: timeout – optional – timeout in seconds (default: wait forever).

property serial: ServiceID | None: Deprecated – please use service_id instead.

class radkit_client.sync.ServiceLoadingStatus

Bases: Enum

An enum of the successive states for a Service inventory or capabilities request.

NOT_YET_STARTED = 'NOT_YET_STARTED'

REQUEST_CREATED = 'REQUEST_CREATED'

IN_PROGRESS = 'IN_PROGRESS'

SUCCESS = 'SUCCESS'

FAILED = 'FAILED'

class radkit_client.sync.E2EEInformation

Bases: SyncWrapper[AsyncE2EEInformation]

E2EE (end-to-end encryption) information for a Service.

service_id

Property type: ServiceID | None

The Service ID used for E2EE.

client_id

Property type: ClientID | None

The Client ID used for E2EE.

e2ee_used

Property type: bool

True if E2EE was used for this request, False otherwise.

domain

Property type: Domain | None

The RADKit Cloud domain that this Client and Service are connected to, as an object.

domain_name

Property type: str | None

The name of the RADKit Cloud domain that this Client and Service are connected to.

sha256_hash_hexdigest

Property type: str | None

The SHA256 hash of the Service E2EE certificate, as a hexadecimal string, or None if no E2EE information is available.

x509_certificate

Property type: x509.Certificate | None

The Service X.509 E2EE certificate, as an object, or None if no E2EE information is available.

class radkit_client.sync.ServicesDict

Bases: SyncDictWrapper[AsyncServicesDict, int, AsyncService, Service]

Service instances that are currently known to a given Client instance. Keys are Service names and values are Service instances.

client

Property type: Client

The Client object that the services in this mapping are bound to.

class radkit_client.sync.RequestsDict

Bases: SyncDictWrapper[AsyncRequestsDict, int, AsyncBaseRequest[BaseModel, BaseModel, BaseModel], SimpleRequest[BaseModel, BaseModel, BaseModel] | StreamRequest[BaseModel, BaseModel, BaseModel] | FillerRequest[BaseModel, BaseModel, BaseModel] | BinaryStreamRequest[BaseModel, BaseModel, BaseModel]]

Mapping (non-mutable dict) for displaying requests associated with an object.

client

Property type: Client

The Client object that the requests in this mapping are bound to.

Integrated 

If RADKit Service is installed, it can be started within the RADKit Client process and interacted with locally without the need for a connection to the RADKit Cloud (see Integrated mode for a full explanation and some examples).

with Client.create() as client:
    service = client.service_integrated()

service_integrated is directly exposed as a function in the radkit-client REPL.

Client.service_integrated(

*_: object,

**__: object,

) → None

Device 

class radkit_client.sync.Device

Bases: SyncWrapper[AsyncDevice]

Represents a single network device reachable through a RADKit Service. Device objects are part of the inventory of a Service instance, which takes the form of a DeviceDict through which devices can be looked up:

>>> device = service.inventory["device-name"]

This class gives access to information about the device, as well as a number of actions such as remote command execution, Swagger/Netconf queries or port forwarding. Some of those actions can also be carried out on multiple devices at the same time (see DeviceDict for details).

service

Property type: Service

Returns the Service object that this Device is bound to.

client

Property type: Client

Returns the Client object that this Device is bound to.

name

Property type: str

The canonical (unique) name of this device. This name is normalized, guaranteed to be unique and is used as the device name in DeviceDict as well as SOCKS5 proxy FQDNs (see SOCKS5 proxy for details).

service_display_name

Property type: str

The display name of this device, as defined on the RADKit Service. This name is not guaranteed to be unique. The canonical name is derived from the display name.

attributes

Property type: DeviceAttributes

Attributes for this device, as returned by the RADKit Service. These can be updated through update_attributes().

host

Property type: str

The hostname or IP address of this device. Shortcut for attributes.internal["host"].

device_type

Property type: str

The device type string for this device. Shortcut for attributes.internal["device_type"].

forwarded_tcp_ports

Property type: PortRanges

The range(s) of allowed forwarded TCP ports for this device. Shortcut for attributes.internal["forwarded_tcp_ports"].

failed

Property type: bool

When True, indicates that this device has failed to process a request. Can be used to quickly filter out devices from which complete sets of output will not be available (see DeviceDict.exclude_failed()).

session_logs

Property type: SessionLogsDict

All terminal/SCP session logs that were created for this device. (This includes sessions that are terminated.)

sessions

Property type: SessionsDict

All terminal/SCP sessions that were established for this device. (This includes sessions that are terminated.)

requests

Property type: RequestsDict

All requests performed against this device so far.

port_forwards

Property type: TCPPortForwardsDict

All open direct port forwardings for this device (SOCKS/HTTP proxy excluded).

swagger

Property type: swagger.SwaggerAPI

The cached Swagger capabilities for this device. Can be updated by calling update_swagger().

See Swagger/OpenAPI for details.

netconf

Property type: SingleDeviceNetconfAPI

The cached Netconf capabilities for this device. Can be updated by calling update_netconf().

See Netconf/YANG for details.

snmp

Property type: SingleDeviceSNMP_API

API for sending SNMP requests to this device.

See SNMP for details.

http

Property type: HttpApi

Sends an HTTP request to this device. This can be used in cases where a Swagger API is not available. This requires much more coding, particularly around authentication.

See Raw HTTP API for details.

singleton() → DeviceDict: Returns a DeviceDict containing only this device.

update_attributes() → SimpleRequest[GetDetailedInventory, DetailedInventoryResponse, None]: Retrieves detailed information about this device from the Service. See attributes for details.

interactive( command: str | None = None, term: str | None = None, initial_terminal_size: os.terminal_size | None = None, request_pty: bool = True, include_initialization: bool = False, ) → None

Opens a session using terminal(), then attaches to the session, and automatically closes the session after the user disconnects/detaches.

The parameters are the same as for terminal().

terminal( command: str | None = None, term: str | None = None, initial_terminal_size: os.terminal_size | None = None, request_pty: bool = True, include_initialization: bool = False, ) → InteractiveConnection

Opens a terminal (SSH/telnet) session to this device and returns a connection object.

Parameters:

command – optional – if specified, run this command instead of an interactive shell. This may not work on all devices; for instance, it is not supported on IOS and related, but it should work on most Linux-based platforms.
term – optional – the $TERM environment variable for the subshell; if not specified, pass on $TERM from the Client environment.
initial_terminal_size – optional – the terminal size as a (columns, rows) tuple; if not specified, use the terminal size of the controlling terminal if there is one, or (80, 40) otherwise.
request_pty – optional – tells the remote system to allocate a PTY (pseudo-terminal) for the command to run in (usually this affects only processes that run within an SSH channel on the remote side; default: False).
include_initialization – optional – show the output of device provisioning and jumphost initialization, if applicable, at the start of the session (default: False).

forward_tcp_port( local_port: int, destination_port: int, local_address: str = localhost, server_startup_timeout: float = 2.0, ) → TCPPortForwarder

Starts listening on local_port on the host running the RADKit Client and forwards every incoming connection to destination_port on this device through the RADKit Service. The device must have port forwarding enabled on the RADKit Service and the destination port must be part of an allowed range.

This method waits until we’re actually listening and accepting incoming connections before returning.

Parameters:

local_port – the port to bind the local listener to.
destination_port – the port to forward connections to on the device.
local_address – optional – the address to bind the local listener to (default: localhost).
server_startup_timeout – optional – how long (in seconds) to wait for the local listener to start (default: 2).

update_swagger( timeout: float | None = None, ) → TransformedFillerRequest[DeviceAction[GetSwaggerPaths], DeviceActionPartialResponse[SwaggerPaths], DeviceDictCapabilitiesUpdate, DeviceCapabilitiesUpdate]

Updates the cached Swagger capabilities for this device.

Parameters:: timeout – the time allowed for the GetSwaggerPaths request to complete on the Service side.

update_netconf() → TransformedFillerRequest[DeviceAction[GetNetconfCapabilities], DeviceActionPartialResponse[dict[str, str]], DeviceDictCapabilitiesUpdate, DeviceCapabilitiesUpdate]: Updates the cached Netconf capabilities for this device.

scp_download_to_stream( remote_path: str, ) → FileReadConnection

Downloads a file from this device via SCP into local memory, to be read as a stream. Data is read from the stream using FileReadConnection.read(), FileReadConnection.readline() etc.

Parameters:: remote_path – the source path on the device.

>>> device.scp_download_to_stream("hello.txt").wait().read()
b"Hello world!"

sftp_download_to_stream( remote_path: str, ) → FileReadConnection

Downloads a file from this device via SFTP into local memory, to be read as a stream. Data is read from the stream using FileReadConnection.read(), FileReadConnection.readline() etc.

Parameters:: remote_path – the source path on the device.

>>> device.sftp_download_to_stream("hello.txt").wait().read()
b"Hello world!"

scp_download_to_file( remote_path: str, local_path: str | None = None, ) → FileReadConnection

Downloads a file from this device via SCP. If the destination file exists, it will be overwritten. Progress can be tracked using FileReadConnection.show_progress().

Parameters:

remote_path – the source path on the remote device.
local_path – optional – the destination path on the local host; if not given, use the current directory; if a directory is given, use the same filename as remote_path.

>>> result = device.scp_download_to_file("hello.txt", "./").wait()
11:59:33.801Z INFO  | download complete [local_path='./hello.txt' size=12]

sftp_download_to_file( remote_path: str, local_path: str | None = None, ) → FileReadConnection

Downloads a file from this device via SFTP. If the destination file exists, it will be overwritten. Progress can be tracked using FileReadConnection.show_progress().

Parameters:

remote_path – the source path on the remote device.
local_path – optional – the destination path on the local host; if not given, use the current directory; if a directory is given, use the same filename as remote_path.

>>> result = device.sftp_download_to_file("hello.txt", "./").wait()
11:59:33.801Z INFO  | download complete [local_path='./hello.txt' size=12]

scp_to_destination( remote_path: str, upload_parameters: UploadParameters, ) → FillerRequest[DeviceAction[ReadAndUploadFile], DeviceActionPartialResponse[NOTHING], None]

Retrieves a file from this device via SCP and uploads it to a chosen destination. The data does not flow through the Client: the Service establishes the SCP connection to the device, retrieves the data and forwards it to a destination determined by the UploadParameters object (see Upload parameters for more information).

Parameters:

remote_path – the source path on the device for the SCP transfer.
upload_parameters – an UploadParameters object.

sftp_to_destination( remote_path: str, upload_parameters: UploadParameters, ) → FillerRequest[DeviceAction[ReadAndUploadFile], DeviceActionPartialResponse[NOTHING], None]

Retrieves a file from this device via SFTP and uploads it to a chosen destination. The data does not flow through the Client: the Service establishes the SFTP connection to the device, retrieves the data and forwards it to a destination determined by the UploadParameters object (see Upload parameters for more information).

Parameters:

remote_path – the source path on the device for the SFTP transfer.
upload_parameters – an UploadParameters object.

scp_upload_from_stream( remote_path: str, size: int, chmod: str = 644, ) → FileWriteConnection

Uploads a file to this device via SCP by writing to a stream. The SCP protocol mandates that the total number of bytes be provided in advance. If the destination file exists, it will be overwritten.

Parameters:

remote_path – the destination path on the device.
size – the size in bytes of the data to be written (has to be accurate).
chmod – optional – the octal mode mask for the remote file (default: 644).

>>> data = b"Hello world!"
>>> con = device.scp_upload_from_stream("hello.txt", len(data)).wait()
>>> con.write(data)
>>> con.wait_closed()

sftp_upload_from_stream( remote_path: str, size: int, chmod: str = 644, ) → FileWriteConnection

Uploads a file to this device via SFTP by writing to a stream. The SFTP protocol mandates that the total number of bytes be provided in advance. If the destination file exists, it will be overwritten.

Parameters:

remote_path – the destination path on the device.
size – the size in bytes of the data to be written (has to be accurate).
chmod – optional – the octal mode mask for the remote file (default: 644).

>>> data = b"Hello world!"
>>> con = device.sftp_upload_from_stream("hello.txt", len(data)).wait()
>>> con.write(data)
>>> con.close()

scp_upload_from_file( local_path: str, remote_path: str, ) → FileWriteConnection

Uploads a file to this device via SCP. If the destination file exists, it will be overwritten. Progress can be tracked using FileWriteConnection.show_progress().

Parameters:

local_path – the source path on the local host.
remote_path – the destination path on the remote device; if a directory is given, uses the same filename as local_path.

>>> result = device.scp_upload_from_file("hello.txt", "./").wait()
12:17:38.622Z INFO  | upload complete [local_path='hello.txt']

sftp_upload_from_file( local_path: str, remote_path: str, ) → FileWriteConnection

Uploads a file to this device via SFTP. If the destination file exists, it will be overwritten. Progress can be tracked using FileWriteConnection.show_progress().

Parameters:

local_path – the source path on the local host.
remote_path – the destination path on the remote device; if a directory is given, uses the same filename as local_path.

>>> result = device.scp_upload_from_file("hello.txt", "./").wait()
12:17:38.622Z INFO  | upload complete [local_path='hello.txt']

exec( commands: str, timeout: int = 0, upload_to: HTTPUploadParameters | GenericUploadParameters | DEVNULLUploadParameters | None = None, reset_before: bool = False, reset_after: bool = False, sudo: bool = False, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → SingleExecResponse[str]

exec( commands: list[str], timeout: int = 0, upload_to: HTTPUploadParameters | GenericUploadParameters | DEVNULLUploadParameters | None = None, reset_before: bool = False, reset_after: bool = False, sudo: bool = False, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecResponse_ByCommand_ToSingle[str]

Executes one or more commands on this device.

See Command execution for more information.

Parameters:

commands – either a non-empty string or a list of strings representing the command(s) to run.
timeout – optional – how many seconds the Service should wait for the command(s) to complete (default: 0 = use the timeout value configured on the Service).
upload_to – optional – if present, upload the command output to the destination specified in the UploadParameters object and return a status message instead of the command output (see Upload parameters for more information).
reset_before – optional – close and re-open any cached connection to the device before running the command(s) (default: False = reuse the cached connection).
reset_after – optional – close the cached connection to the device after running the command(s) (default: False = keep the cached connection alive).
sudo – optional – execute the command(s) with sudo (default: False).
prompt_detection_strategy – optional – chooses prompt detection strategy, if not set prompt detection strategy is chosen by service

create_device_flow( flow_mode: DeviceFlowMode | None = None, exec_error_patterns: Sequence[Pattern[str]] | None = None, ) → DeviceFlow

Creates an DeviceFlow helper based on this device.

Parameters:

flow_mode – optional – the Device Flow mode (default: DeviceFlowMode.BEST_EFFORT).
exec_error_patterns – optional – a list of error patterns (default: use the built-in ones).

class radkit_client.sync.DeviceDict

Bases: SyncDictWrapper[AsyncDeviceDict, str, AsyncDevice, Device]

A collection of Device objects (a subset of the Service inventory, or the Service inventory as a whole).

Even though this class is called “device dict”, it behaves mostly like a (mutable) set, but the elements (the devices) are indexable by their name, similar to a dict. Devices can be added or removed from the collection. There are methods like .exec() to apply the .exec() command to all devices in the set at once.

Iterating over the DeviceDict object will produce the keys (device names) that can be used to look up individual devices.

The following in-place mutations are supported:

add(): adds devices or other device dicts;
remove(): removes devices or other device dicts;
these two methods accept either: a single Device object; a DeviceDict with any number of devices; a UUID; a name; or a list of any of those.

The following operations produce a new DeviceDict:

filter(): filters by field name and value;
subset(): selects a subset or take an intersection with another device dict;
exclude_failed(): returns a new device dict excluding the devices that were marked as “failed” (= had an error during a previous operation).

Python set operators are supported and produce a new device dict, for example:

union: dict1 | dict2 or dict1.union(dict2)
intersection: dict1 & dict2 or dict1.intersection(dict2)
difference: dict1 - dict2 or dict1.difference(dict2)

Other set-like functionality:

len(device_dict): returns the number of devices in the dict;
device_object in device_dict (membership): tests for the presence of the given UUID, name or device object in the dict;
device_dict1 > device_dict2 (comparison): tests if one dict is a subset or superset of another;
.discard() and .update() are not implemented because .remove() and .add() do exactly the same thing.

Dict-like functionality:

device_dict[key]: gets an item from the device dict by key, where key can be a device name or a UUID;
.get(key): gets an item from the device dict by name or UUID.

Iterating over all devices:

for device in device_dict: iterates over the device names in the dict, similar to iterating over a dictionary;
for device in device_dict.items(): iterates pairwise over device names and device objects at the same time;
for name in device_dict.device_names(): iterates over the device names.

status

Property type: DeviceDictStatus

The status of this DeviceDict. See DeviceDictStatus for details.

service

Property type: Service

Returns the Service object that this DeviceDict is bound to.

client

Property type: Client

Returns the Client object that this DeviceDict is bound to.

touched

Property type: bool

If False, this device dict represents the whole inventory of the Service. It will track the contents of the inventory, i.e. newly added devices in the inventory will also show up in this dict.

If True, this device dict contains a defined set of devices. Devices that are removed from the inventory will be hidden, but newly added devices will not show up.

This attribute goes from False to True if this device dict is modified in place using add() or remove(), or if it is created as a result of an operation such as filter() or difference().

Removes one or more devices from this device dict (modifies it in place). Does not complain if some of those devices are not present.

Parameters:: obj – a single Device object, device name, device UUID, or a (possibly mixed) list of those.

Adds one or more devices to this device dict (modifies it in place). Does not complain if some of those devices are already present.

Parameters:: obj – a single Device object, device name, device UUID, or a (possibly mixed) list of those.

clone() → DeviceDict: Alias for copy(), retained for backwards compatibility. This method will be removed in the future.

copy() → DeviceDict: Returns a shallow copy of this DeviceDict.

filter( attr: str, pattern: str, ) → DeviceDict

Returns a subset of the devices in this dict for which the specified attribute matches the specified pattern (see Python regular expressions). First the attribute name is looked for in the device parameters, then in the internal attributes, then in the metadata.

Parameters:

attr – the name of the attribute to look at.
pattern – the regular expression to search for, using re.search().

exclude_failed() → DeviceDict

Returns a subset of the devices in this dict that excludes failed devices (see Device.failed).

This makes it possible to run a sequence of commands, where each following command will only be executed on the devices for which we had successful responses, for example:

device_dict = service.inventory.filter(...)
device_dict.exclude_failed().exec('show version').wait().result.data
device_dict.exclude_failed().exec('show clock').wait().result.data

subset( devices: Iterable[str | UUID | Device], ) → DeviceDict

Returns a subset of the devices in this dict corresponding to the list of devices, names or UUIDs passed as argument. (In other words, compute the intersection between our DeviceDict and the given devices.)

Parameters:: devices – mixed list of device names, device UUIDs, or Device objects.

union( other: object, ) → DeviceDict: Set union (OR) of two device dicts (same as the | operator).

difference( other: object, ) → DeviceDict: Set difference of two device dicts (same as the - operator).

intersection( other: object, ) → DeviceDict

Set intersection (AND) of two device dicts (same as the & operator).

See also subset() which does the same, but takes more arguments.

symmetric_difference( other: object, ) → DeviceDict: Set symmetric difference (XOR) of two device dicts (same as the ^ operator).

issubset( other: DeviceDict, ) → bool: Tests whether every element in this dict is in other.

issuperset( other: DeviceDict, ) → bool: Tests whether every element in other is in this dict.

update_attributes() → SimpleRequest[GetDetailedInventory, DetailedInventoryResponse, None]: Retrieves detailed information from the Service for all the devices in this dict.

update_swagger( timeout: float | None = None, ) → FillerRequest[DeviceAction[GetSwaggerPaths], DeviceActionPartialResponse[SwaggerPaths], DeviceDictCapabilitiesUpdate]

Updates the cached Swagger capabilities for the devices in this dict.

Parameters:: timeout – the time allowed for the GetSwaggerPaths RPC request to complete on the Service side.

update_netconf() → FillerRequest[DeviceAction[GetNetconfCapabilities], DeviceActionPartialResponse[dict[str, str]], DeviceDictCapabilitiesUpdate]: Updates the cached Netconf capabilities for the devices in this dict.

netconf

Property type: NetconfAPI

The combined cached Netconf capabilities for the devices in this dict. Can be updated by calling update_netconf().

See Netconf/YANG for details.

snmp

Property type: SNMP_API

API for sending SNMP requests to the devices in this dict.

See SNMP for details.

session_logs

Property type: SessionLogsDict

All terminal/SCP session logs that were created for all the sessions for the devices in this dict. (This includes sessions that are terminated.)

sessions

Property type: SessionsDict

All terminal/SCP sessions that were established for the devices in this dict. (This includes sessions that are terminated.)

requests

Property type: RequestsDict

All requests performed so far against the devices in this dict.

port_forwards

Property type: TCPPortForwardsDict

All open direct port forwardings for this device (SOCKS/HTTP proxy excluded).

create_device_flow( flow_mode: DeviceFlowMode | None = None, exec_error_patterns: Sequence[Pattern[str]] | None = None, ) → DeviceFlow

Creates a DeviceFlow helper based on the devices in this dict.

Parameters:

flow_mode – optional – the Device Flow mode (default: DeviceFlowMode.BEST_EFFORT).
exec_error_patterns – optional – a list of error patterns (default: use the built-in ones).

exec( commands: str, *, timeout: int = 0, synced: bool = ..., upload_to: HTTPUploadParameters | GenericUploadParameters | DEVNULLUploadParameters | None = None, reset_before: bool = False, sudo: bool = False, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecResponse_ByDevice_ToSingle[str]

exec( commands: list[str], *, timeout: int = 0, synced: bool = ..., upload_to: HTTPUploadParameters | GenericUploadParameters | DEVNULLUploadParameters | None = None, reset_before: bool = False, sudo: bool = False, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecResponse_ByDevice_ByCommand[str]

Executes one or more commands on the devices in this dict.

See Command execution for more information.

Parameters:

commands – either a non-empty string or a list of strings representing the command(s) to run.
timeout – optional – how many seconds the Service should wait for the command(s) to complete (default: 0 = use the timeout value configured on the Service).
upload_to – optional – if present, upload each command output to the destination specified in the UploadParameters object and return a status message instead of the command output (see Upload parameters for more information).
reset_before – optional – close and re-open any cached connection to each device before running the command(s) (default: False = reuse the cached connection).
reset_after – optional – close the cached connection to each device after running the command(s) (default: False = keep the cached connection alive).
sudo – optional – execute the command(s) with sudo (default: False).
prompt_detection_strategy – optional – chooses prompt detection strategy, if not set prompt detection strategy is chosen by service

scp_to_destination( remote_path: str, upload_parameters: UploadParameters, ) → FillerRequest[DeviceAction[ReadAndUploadFile], DeviceActionPartialResponse[NOTHING], None]

Retrieves a file from each device via SCP and uploads it to a chosen destination. The data does not flow through the Client: the Service establishes the SCP connection to the device, retrieves the data and forwards it to a destination determined by the UploadParameters object (see Upload parameters for more information).

Parameters:

remote_path – the source path on the device for the SCP transfer.
upload_parameters – an UploadParameters object.

sftp_to_destination( remote_path: str, upload_parameters: UploadParameters, ) → FillerRequest[DeviceAction[ReadAndUploadFile], DeviceActionPartialResponse[NOTHING], None]

Retrieves a file from each device via SFTP and uploads it to a chosen destination. The data does not flow through the Client: the Service establishes the SFTP connection to the device, retrieves the data and forwards it to a destination determined by the UploadParameters object (see Upload parameters for more information).

Parameters:

remote_path – source path on the device for the SFTP transfer.
upload_parameters – an UploadParameters object.

class radkit_client.sync.DeviceAttributes

Bases: SyncWrapper[AsyncDeviceAttributes]

Container for the three categories of device attributes (internal, metadata and ephemeral).

internal

Property type: DeviceAttributesDict

Internal attributes come from the Service internal inventory. Those include the display name, hostname, SSH/telnet port number, TCP forwarded ports, etc.

metadata

Property type: DeviceAttributesDict

Metadata comes from the device import source (e.g. Catalyst Center). If the device was added to RADKit Service manually or through the API rather than imported from an external source, the metadata will be empty.

ephemeral

Property type: EphemeralDeviceAttributesDict

Ephemeral attributes can be filled in by the RADKit Client user. These attributes are locally significant to the Client only, and as the name implies, they are discarded when the device object is destroyed.

device_name

Property type: str

A shortcut to the device’s canonical name.

device

Property type: Device

The Device object for the device that has these attributes.

class radkit_client.sync.DeviceAttributesDict

Bases: SyncDictWrapper[AsyncDeviceAttributesDict, str, object, object]

A mapping (non-mutable dict) that holds device internal attributes or metadata.

device_name

Property type: str

A shortcut to the device’s canonical name.

device

Property type: Device

The Device object for the device that has these attributes.

class radkit_client.sync.EphemeralDeviceAttributesDict

Bases: SyncDictWrapper[AsyncEphemeralDeviceAttributesDict, str, object, object]

A mutable dictionary that the user can use to add ephemeral attributes to a device. The contents of this dict remain in memory even if the device is renamed or temporarily disabled, and are discarded once the Device object is deleted from memory.

device_name

Property type: str

A shortcut to the device’s canonical name.

device

Property type: Device

The Device object for the device that has these attributes.

Command execution 

These are the result and status objects for Device.exec() and DeviceDict.exec(). The chain of objects returned depends on the parameters of the request (single or multiple devices, and single or multiple commands):

single_device.exec(single_command): SingleExecResponse
- status: ExecStatus
- data: str
multiple_devices.exec(single_command): ExecResponse_ByDevice_ToSingle
- status: ExecStatus
- [device]: SingleExecResponse
  
  status: ExecStatus
  
  data: str
single_device.exec(multiple_commands): ExecResponse_ByCommand_ToSingle
- status: ExecStatus
- [command]: SingleExecResponse
  
  status: ExecStatus
  
  data: str
multiple_devices.exec(multiple_commands): ExecResponse_ByDevice_ByCommand
- status: ExecStatus
- [device]: ExecResponse_ByCommand_ToSingle
  
  status: ExecStatus
  
  [command]: SingleExecResponse
  
  status: ExecStatus
  
  data: str

class radkit_client.sync.ExecResponseBase

Bases: SyncWrapper[AsyncExecResponseBase[_D]], Generic[_D]

status

Property type: ExecStatus

The execution status for this exec result. See ExecStatus for details.

success

Property type: bool

True if we have a SUCCESS status.

sudo

Property type: bool

Returns True if sudo was requested for this command.

devices

Property type: DeviceDict

DeviceDict that corresponds to the associated devices for this exec result.

commands

Property type: Sequence[str]

Returns the canonical name of the device that these commands were run on.

device_names

Property type: Sequence[str]

List of associated device names for this exec result.

requests: Property type: Sequence[SimpleExecRequest]

client

Property type: Client

Returns the Client object that this exec response is bound to.

property unfiltered: AsyncExecResponse_ByIndex[str]: All entries, without any filtering (or mapping/sorting) by index.

property by_index: AsyncExecResponse_ByIndex[_DataType]

Produce a view that allows for iteration over all entries in this view by a numeric index.

Usage:

narrowed_response = response.by_index[12]

property by_status: AsyncExecResponse_ByStatus[_DataType]

Produce a view that allows for filtering by state (success/failure).

Usage:

narrowed_response = response.by_status['SUCCESS']

property by_command: AsyncExecResponse_ByCommand[_DataType]

Produce a mapping for selecting the responses for a single command.

Usage:

narrowed_response = response.by_command['my-command-name']

property by_device: AsyncExecResponse_ByDevice[_DataType]

Produce a mapping for selecting the responses for a single device name.

Usage:

narrowed_response = response.by_device['my-device-name']

property by_device_type: AsyncExecResponse_ByDeviceType[_DataType]

Produce a mapping for selecting the responses for a device type.

Usage:

narrowed_response = response.by_device_type['LINUX']

property by_device_by_command: AsyncExecResponse_ByDeviceType[_DataType]

by_device_type() -> ExecResponse_ByDeviceType[_DataType]

Produce a mapping for selecting the responses for a device type.

Usage:

narrowed_response = response.by_device_type['LINUX']

filter( filter_func: ExecFilterFunction[_DataType], ) → ExecResponse_ByIndex[_DataType]

Filter exec response by applying a filter function.

Example:

response.filter(lambda single_response: 'iosxr' in single_response.device_name)

map( map_func: ExecMapFunction[_DataType, _DataType2], ) → ExecResponse_ByIndex[_DataType2]

Transform exec response by applying a map function to the data column. This is useful for applying a parser to the returned response.

Example:

response.map(lambda exec_record, data: data.lower())

The map function should accept two arguments. First, the ExecRecord which holds all information about the individual entry, like a device type, which can be used for lookup up the right parser function. Secondly, it receives the actual data that should be transformed.

sort( sort_key_func: ExecSortFunction[_DataType], ) → ExecResponse_ByIndex[_DataType]

Sort exec response by applying a sort function.

Example:

response.sort(lambda single_response: single_response.status)

property sort_by_device: AsyncExecResponse_ByIndex[_DataType]: Sort records by device name.

property sort_by_device_type: AsyncExecResponse_ByIndex[_DataType]: Sort records by device type.

property sort_by_status: AsyncExecResponse_ByIndex[_DataType]: Sort records by status .

property sort_by_command: AsyncExecResponse_ByIndex[_DataType]: Sort records by command.

excluded_commands

Property type: frozenset[str]

Commands that have been filtered out from this response object.

excluded_device_names

Property type: frozenset[str]

Devices that have been filtered out from this response object.

excluded_devices

Property type: DeviceDict

Devices that were part of the full response, but filtered out in this object.

cancel() → None: Cancel all underlying RPC requests to the service for this command execution.

wait( timeout: float | None = None, ) → Self: Wait until all underlying RPC requests are completed.

show_progress() → None: Show progress bar of the completion progress for this view.

property result: Self

property full_result: ExecResponse_ByDevice_ByCommand[str]

class radkit_client.sync.ExecResponse_ByIndex: Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByIndex[_D], int, AsyncSingleExecResponse[_D] | AsyncExecResponse_ByIndex[_D], SingleExecResponse[_D] | ExecResponse_ByIndex[_D]], Generic[_D]

class radkit_client.sync.ExecResponse_ByDevice_ByCommand: Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByDevice_ByCommand[_D], str, AsyncExecResponse_ByCommand_ToSingle[_D], ExecResponse_ByCommand_ToSingle[_D]], Generic[_D]

class radkit_client.sync.ExecResponse_ByDevice_ToSingle

Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByDevice_ToSingle[_D], str, AsyncSingleExecResponse[_D], SingleExecResponse[_D]], Generic[_D]

command

Property type: str

Returns the command that was executed.

class radkit_client.sync.ExecResponse_ByCommand_ToSingle

Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByCommand_ToSingle[_D], str, AsyncSingleExecResponse[_D], SingleExecResponse[_D]], Generic[_D]

device

Property type: Device

Returns the Device object for the device that these commands were run on.

class radkit_client.sync.ExecResponse_ByCommand: Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByCommand[_D], str, AsyncExecResponse_ByIndex[_D], ExecResponse_ByIndex[_D]], Generic[_D]

class radkit_client.sync.ExecResponse_ByDevice: Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByDevice[_D], str, AsyncExecResponse_ByIndex[_D], ExecResponse_ByIndex[_D]], Generic[_D]

class radkit_client.sync.ExecResponse_ByDeviceType: Bases: ExecResponseBase[_D], SyncDictWrapper[AsyncExecResponse_ByDeviceType[_D], str, AsyncExecResponse_ByIndex[_D], ExecResponse_ByIndex[_D]], Generic[_D]

class radkit_client.sync.SingleExecResponse

Bases: ExecResponseBase[_D], SyncWrapper[AsyncSingleExecResponse[_D]], Generic[_D]

request

Property type: SimpleExecRequest

The RPC request that was used for this exec() call (mostly for debugging).

device_name

Property type: str

Returns the canonical name of the device that these commands were run on.

device_uuid: Property type: UUID

device_type

Property type: str

Returns the device type of the device that these commands were run on.

device

Property type: Device

Returns the Device object for the device that these commands were run on.

command

Property type: str

Returns the command that was executed.

client_id

Property type: ClientID | None

The identity that is connected to the remote Service for this request.

service_id

Property type: ServiceID | None

The ID of the Service that handled this request.

property data: _DataType

The text output, or in case .map() was called, the transformed output for this single response.

Raises:

ExecError when something went wrong
or the subclass ExecPendingError if this request is still in progress.
or the subclass ExecMapError if something went wrong while applying a given map function.

raw_data

Property type: str

The text output for this single response.

This can raise ExecError when something went wrong, or the subclass ExecPendingError if this request is still in progress.

errors

Property type: Sequence[str]

List of associated error messages.

property status_message: str

class radkit_client.sync.ExecStatus

Bases: Enum

The status of the execution of a single or multiple commands on one or more devices.

Note that this status does not indicate whether the commands succeeded on a functional level, only that those were successfully delivered at the prompt and that some output was received. It is still possible, for example, that the device rejected a command with a syntax error or that a command was not allowed due to insufficient privileges.

FAILURE = 'FAILURE': None of the commands could be delivered.

SUCCESS = 'SUCCESS': The commands were delivered successfully.

PROCESSING = 'PROCESSING': The commands are still in the process of being delivered.

PARTIAL_SUCCESS = 'PARTIAL_SUCCESS': Some of the commands could be delivered and some could not.

classmethod from_multiple( statuses: Iterable[ExecStatus], ) → ExecStatus

class radkit_client.sync.ExecError

Bases: ClientError

Raised when the execution of a command failed, i.e. it could not be successfully delivered to the device and/or its output could not be collected.

class radkit_client.sync.ExecRecord

Bases: Generic[_DataType]

Single output for usage in ExecFilterFunction, ExecMapFunction and ExecSortFunction.

command: str

sudo: bool

service_id: ServiceID | None

device_uuid: UUID

device_name: str

device_type: str

device_attributes_internal: Mapping[str, object]

device_attributes_metadata: Mapping[str, object]

device_attributes_ephemeral: Mapping[str, object]

property raw_data: str

property data: _DataType

property status: ExecStatus

Terminal and SCP/SFTP 

class radkit_client.sync.InteractiveConnection

Bases: _TerminalConnection[AsyncInteractiveConnection]

A terminal session to a remote device that can be interacted with in multiple ways:

interactively, by attaching to the controlling terminal;
programmatically, using raw read/write APIs;
programmatically, using RADKit’s terminal interactions and exec sequences.

status

Property type: InteractiveConnectionStatus

The status of this interactive connection. See InteractiveConnectionStatus for details.

attach() → None

Starts interactive mode by attaching the current terminal to the session.

To detach or control the session, type the escape character ~ (tilde), followed by one of these characters:

. (period): detach from the connection;
? (question mark): show a help message;
~ (tilde): send a single ~ character to the remote host.

If connecting to RADKit over SSH, ~ is probably the escape sequence of your SSH client as well. Press ~ twice to send it to RADKit rather than your SSH client. This also works the other way around, if you need to disconnect an SSH session initiated from the remote device, without disconnecting RADKit from the device.

Note

Support for ^C^C^C^C, ~c and ~z as alternative ways to detach was removed in RADKit 1.6.0.

write( data: str | bytes, ) → None

Writes bytes to this connection.

Parameters:: data – the bytes to be written.

read( n: int = -1, *, timeout: float | None = None, ) → bytes

Reads bytes from this connection.

Parameters:

n – optional – the number of bytes to read. Defaults to -1, which means: read until EOF (similar to what StreamReader.read does).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.

readline( *, timeout: float | None = None, ) → bytes

Reads a single line from this connection (returns bytes).

Parameters:: timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).
Raises:: ValueError – if the timeout is lower than or equal to zero.

readexactly( n: int, *, timeout: float | None = None, ) → bytes

Reads exactly the specified number of bytes from this connection.

Parameters:

n – the number of bytes to read.
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before n bytes could be read. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil( separator: bytes = b\n, *, timeout: float | None = None, ) → bytes

Reads from this connection until a separator is found. The separator is included as part of the output. It may be several bytes long.

Parameters:

separator – optional – the separator that signals the end of the data to be read (expressed as bytes, defaults to \n = single newline).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before the separator could be found. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil_regex( pattern: bytes, *, timeout: float | None = None, ) → bytes

Reads from this connection until a match for the specified regular expression is found. The data that matched the regex is included as part of the output.

Note

Since the data to be searched is read as bytes, the regular expression also has to be expressed as bytes. For example: rb"[\r\n]".

Parameters:

pattern – the regular expression that detects the end of the data to be read (expressed as bytes).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before the regex could be matched. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil_timeout( timeout: float, ) → bytes

Reads as much data as possible from this connection until no data is received for the specified amount of time.

Parameters:: timeout – how long to wait (in seconds) while no data is coming in, before returning.

exec( command: str, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → str

Performs command execution and reads the output until it gets to the next prompt. Returns the command output as a string.

Parameters:

command – the command to be executed.
prompt_detection_strategy – optional – chooses prompt detection strategy, if not set prompt detection strategy is chosen by service

run_exec_sequence( exec_sequence: ExecSequence, device_information: DeviceInformation | None = None, ) → bytes

Runs an ExecSequence locally on this connection.

Parameters:

exec_sequence – an ExecSequence object to run locally.
device_information – optional – this parameter is required if the ExecSequence requires device information like credentials to operate. Usually, this information is only available on the Service, and for those situations it is advised to use run_remote_exec_sequence() instead of storing secrets on the Client.

run_remote_exec_sequence( exec_sequence: ExecSequence, upload_to: UploadParameters | None = None, ) → bytes

Sends an ExecSequence to the Service and runs it on this connection remotely (i.e. on the Service side). This is usually needed when Service-side device details/secrets are used in the sequence.

The captured result will be returned as a bytes object, or uploaded if an upload_to parameter is given. In the latter case, the captured data won’t be transferred to the Client, but directly uploaded from within the Service, and a status message will be returned to the Client instead of the actual output.

Parameters:

exec_sequence – an ExecSequence object to run on the Service side.
upload_to – optional – if present, upload the output to the destination specified in the UploadParameters object (see Upload parameters for more information).

run_remote_exec_sequence_as_connection( exec_sequence: ExecSequence, upload_to: UploadParameters | None = None, ) → RemoteExecSequenceConnection

Executes an ExecSequence on the Service side.

This returns a new connection object of type RemoteExecSequenceConnection. The output captured by the Service can be streamed by calling its .read() method.

Warning

All data from this connection must be read before the main connection will be unblocked.

Usually run_remote_exec_sequence() is recommended instead.

run_procedure( name: str, device_information: DeviceInformation | None = None, ) → bytes

Executes a stored procedure locally (on the Client) on this connection.

Parameters:

name – the name of the stored procedure.
device_information – optional – this parameter is required if the stored procedure requires device information like credentials to operate. Usually, this information is only available on the Service, and for those situations it is advised to use run_remote_procedure() instead of storing secrets on the Client.

run_remote_procedure( name: str, ) → bytes

Executes a stored procedure remotely (on the Service) on this connection.

Parameters:: name – the name of the stored procedure on the remote Service.

read_line( *, timeout: float | None = None, ) → bytes: Deprecated. Please use readline() instead.

at_eof() → bool: Returns True when there is no more data to read. (This can return False even if the connection is closed, if there is still data in the buffer.)

attach_socket() → SocketForwarder: Attaches a local socket that forwards bytes read from and written to this connection. This can be used e.g. for attaching a pexpect instance.

bytes_read

Property type: int

Returns the number of bytes read so far on this connection.

bytes_written

Property type: int

Returns the number of bytes written so far on this connection.

client_id

Property type: ClientID | None

Returns the Client identity that is used to connect to the remote Service.

close() → None: Closes this connection. This call does not block; call wait_closed() afterwards to wait for the connection closure to complete.

closed

Property type: bool

Returns True if the connection is closed (the request completed successfully or unsuccessfully), False otherwise.

detach_socket() → None: Detaches and closes the local socket linked to this connection.

device

Property type: Device

Returns the Device object for the device that this connection is established with.

device_name

Property type: str

Returns the canonical name of the device that this connection is established with.

request

Property type: BinaryStreamRequest[_T_RequestModel, _T_ResponseModel, _T_UploadModel]

The underlying request object that is used to perform this connection.

property serial: ServiceID | None: Deprecated.

service

Property type: Service

Returns the Service object through which this connection is established.

service_id

Property type: ServiceID | None

Returns the Service ID for the remote Service.

session_log_filename

Property type: str | None

Returns the file name of the current session log file if session logging is enabled, otherwise returns None.

session_log_filepath

Property type: Path | None

Returns the full path to the current session log file if session logging is enabled, otherwise returns None.

session_log_rotate( new_filepath: str | Path | None = None, ) → None: Closes existing session log and starts logging to a new log file on new_filepath, or a default one if not provided.

wait() → Self: Waits until the connection is established. Propagates exceptions.

wait_closed() → Self: Waits for the connection to be completely closed.

write_eof() → None: Writes EOF to the underlying stream. This closes the sockets on both sides.

class radkit_client.sync.SocketForwarder

Bases: SyncWrapper[AsyncSocketForwarder]

Forwards data between a bidirectional socket and a pair of asyncio read/write streams.

status

Property type: SocketForwarderStatus

Returns the status of the socket forwarder.

device_name

Property type: str

Returns the name of the device to which the streams are connected.

device

Property type: Device

Returns the Device object corresponding to the device to which the streams are connected.

socket

Property type: _socket.socket

Returns the socket object associated with this forwarder.

Raises:: RuntimeError – if the forwarder is not running and there is no active socket.

start() → _socket.socket

Starts the forwarder. Returns the socket that is attached to the read/write streams.

Raises:: RuntimeError – if the forwarder is already running or there is no active client.

stop() → None

Stops the forwarder. Waits until the forwarder is actually stopped before returning.

Raises:: RuntimeError – if the forwarder is not running.

spawn_pexpect( args: Any = None, timeout: int = 30, maxread: int = 2000, searchwindowsize: Any = None, logfile: Any = None, encoding: Any = None, codec_errors: str = 'strict', use_poll: bool = False, ) → SocketSpawn[AnyStr]

Convenience wrapper to pexpect. Uses SocketSpawn rather than pexpect.fdspawn, because the latter is UNIX-only.

Example usage:

terminal = device.terminal().wait()
with terminal.attach_socket() as forwarder:
    child = forwarder.spawn_pexpect()
    child.sendline("show clock")
    child.expect("show clock")
    child.expect("#")
    print(child.before)
terminal.close()
terminal.wait_closed()

Or, without context managers:

terminal = device.terminal().wait()
forwarder = terminal.attach_socket()
child = forwarder.spawn_pexpect()

child.sendline("show clock")
child.expect("show clock")
child.expect("#")
print(child.before)

forwarder.stop()
terminal.close()
terminal.wait_closed()

The arguments for this function are the same as for pexpect.socket_pexpect.SocketSpawn.

See https://pexpect.readthedocs.io/ for more information.

class radkit_client.sync.SocketForwarderStatus

Bases: Enum

An enum of the possible statuses for a socket forwarder.

RUNNING = 'RUNNING': Socket forwarder is running.

FAILED = 'FAILED': Socket forwarder failed with an exception.

STOPPED = 'STOPPED': Socket forwarder is not running or was stopped.

class radkit_client.sync.FileReadConnection

Bases: _TerminalConnection[AsyncFileReadConnection]

Connection for reading the contents of a file from the remote peer. Depending on how this object is created, the contents of the remote file will either be written to a local file, or stored in memory and made available for reading.

status

Property type: FileTransferStatus

The status of this file transfer connection. See FileTransferStatus for details.

remote_path() → str | None: Returns the remote path for this connection.

local_path() → str | None: Returns the local path for this connection.

abs_local_path() → str | None: Returns the normalized/absolutized local path for this connection (see os.path.abspath for details).

show_progress() → None: Displays a prompt_toolkit progress bar. Hitting Ctrl-C will stop the display of the bar but not the transfer, which will continue in the background until it completes or the session is terminated.

write( data: str | bytes, ) → None

Writes bytes to this connection.

Parameters:: data – the bytes to be written.

read( n: int = -1, *, timeout: float | None = None, ) → bytes

Reads bytes from this connection.

Parameters:

n – optional – the number of bytes to read. Defaults to -1, which means: read until EOF (similar to what StreamReader.read does).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.

readline( *, timeout: float | None = None, ) → bytes

Reads a single line from this connection (returns bytes).

Parameters:: timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).
Raises:: ValueError – if the timeout is lower than or equal to zero.

readexactly( n: int, *, timeout: float | None = None, ) → bytes

Reads exactly the specified number of bytes from this connection.

Parameters:

n – the number of bytes to read.
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before n bytes could be read. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil( separator: bytes = b\n, *, timeout: float | None = None, ) → bytes

Reads from this connection until a separator is found. The separator is included as part of the output. It may be several bytes long.

Parameters:

separator – optional – the separator that signals the end of the data to be read (expressed as bytes, defaults to \n = single newline).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before the separator could be found. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil_regex( pattern: bytes = b\n, *, timeout: float | None = None, ) → bytes

Reads from this connection until a match for the specified regular expression is found. The data that matched the regex is included as part of the output.

Note

Since the data to be searched is read as bytes, the regular expression also has to be expressed as bytes. For example: rb"[\r\n]".

Parameters:

pattern – the regular expression that detects the end of the data to be read (expressed as bytes).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before the regex could be matched. Bytes read so far are stored in the IncompleteReadError.partial attribute.

wait_transfer_done() → Self: Deprecated. Use wait_closed() instead.

read_line( *, timeout: float | None = None, ) → bytes: Deprecated. Please use readline() instead.

at_eof() → bool: Returns True when there is no more data to read. (This can return False even if the connection is closed, if there is still data in the buffer.)

attach_socket() → SocketForwarder: Attaches a local socket that forwards bytes read from and written to this connection. This can be used e.g. for attaching a pexpect instance.

bytes_read

Property type: int

Returns the number of bytes read so far on this connection.

bytes_written

Property type: int

Returns the number of bytes written so far on this connection.

client_id

Property type: ClientID | None

Returns the Client identity that is used to connect to the remote Service.

close() → None: Closes this connection. This call does not block; call wait_closed() afterwards to wait for the connection closure to complete.

closed

Property type: bool

Returns True if the connection is closed (the request completed successfully or unsuccessfully), False otherwise.

detach_socket() → None: Detaches and closes the local socket linked to this connection.

device

Property type: Device

Returns the Device object for the device that this connection is established with.

device_name

Property type: str

Returns the canonical name of the device that this connection is established with.

request

Property type: BinaryStreamRequest[_T_RequestModel, _T_ResponseModel, _T_UploadModel]

The underlying request object that is used to perform this connection.

property result: Self

property serial: ServiceID | None: Deprecated.

service

Property type: Service

Returns the Service object through which this connection is established.

service_id

Property type: ServiceID | None

Returns the Service ID for the remote Service.

session_log_filename

Property type: str | None

Returns the file name of the current session log file if session logging is enabled, otherwise returns None.

session_log_filepath

Property type: Path | None

Returns the full path to the current session log file if session logging is enabled, otherwise returns None.

session_log_rotate( new_filepath: str | Path | None = None, ) → None: Closes existing session log and starts logging to a new log file on new_filepath, or a default one if not provided.

wait() → Self: Waits until the connection is established. Propagates exceptions.

wait_closed() → Self: Waits for the connection to be completely closed.

write_eof() → None: Writes EOF to the underlying stream. This closes the sockets on both sides.

class radkit_client.sync.FileWriteConnection

Bases: _TerminalConnection[AsyncFileWriteConnection]

Connection for writing to a file on the remote peer. Depending on how this object is created, the data to be sent will be read from a local file, or from a buffer in memory that can be written to.

status

Property type: FileTransferStatus

The status of this file transfer connection. See FileTransferStatus for details.

remote_path() → str | None: Returns the remote path for this connection.

local_path() → str | None: Returns the local path for this connection.

abs_local_path() → str | None: Returns the normalized/absolutized local path for this connection (see os.path.abspath for details).

show_progress() → None: Displays a prompt_toolkit progress bar. Hitting Ctrl-C will stop the display of the bar but not the transfer, which will continue in the background until it completes or the session is terminated.

write( data: str | bytes, ) → None

Writes bytes to this connection.

The writer gets closed automatically after writing the expected number of bytes to the SCP/SFTP stream.

Parameters:: data – the bytes to be written.

read( n: int = -1, *, timeout: float | None = None, ) → bytes

Reads bytes from this connection.

Parameters:

n – optional – the number of bytes to read. Defaults to -1, which means: read until EOF (similar to what StreamReader.read does).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.

readline( *, timeout: float | None = None, ) → bytes

Reads a single line from this connection (returns bytes).

Parameters:: timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).
Raises:: ValueError – if the timeout is lower than or equal to zero.

readexactly( n: int, *, timeout: float | None = None, ) → bytes

Reads exactly the specified number of bytes from this connection.

Parameters:

n – the number of bytes to read.
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before n bytes could be read. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil( separator: bytes = b\n, *, timeout: float | None = None, ) → bytes

Reads from this connection until a separator is found. The separator is included as part of the output. It may be several bytes long.

Parameters:

separator – optional – the separator that signals the end of the data to be read (expressed as bytes, defaults to \n = single newline).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before the separator could be found. Bytes read so far are stored in the IncompleteReadError.partial attribute.

readuntil_regex( pattern: bytes, *, timeout: float | None = None, ) → bytes

Reads from this connection until a match for the specified regular expression is found. The data that matched the regex is included as part of the output.

Note

Since the data to be searched is read as bytes, the regular expression also has to be expressed as bytes. For example: rb"[\r\n]".

Parameters:

pattern – the regular expression that detects the end of the data to be read (expressed as bytes).
timeout – optional – the time (in seconds) after which reading will be interrupted. Defaults to None (no timeout).

Raises:

ValueError – if the timeout is lower than or equal to zero.
IncompleteReadError – when EOF is reached before the regex could be matched. Bytes read so far are stored in the IncompleteReadError.partial attribute.

wait_transfer_done() → Self: Deprecated. Use wait_closed() instead.

start_upload( local_path: str, ) → Future[None]: Deprecated. Please use device.scp_upload_from_file() or device.sftp_upload_from_file() instead.

at_eof() → bool: Returns True when there is no more data to read. (This can return False even if the connection is closed, if there is still data in the buffer.)

attach_socket() → SocketForwarder: Attaches a local socket that forwards bytes read from and written to this connection. This can be used e.g. for attaching a pexpect instance.

bytes_read

Property type: int

Returns the number of bytes read so far on this connection.

bytes_written

Property type: int

Returns the number of bytes written so far on this connection.

client_id

Property type: ClientID | None

Returns the Client identity that is used to connect to the remote Service.

close() → None: Closes this connection. This call does not block; call wait_closed() afterwards to wait for the connection closure to complete.

closed

Property type: bool

Returns True if the connection is closed (the request completed successfully or unsuccessfully), False otherwise.

detach_socket() → None: Detaches and closes the local socket linked to this connection.

device

Property type: Device

Returns the Device object for the device that this connection is established with.

device_name

Property type: str

Returns the canonical name of the device that this connection is established with.

request

Property type: BinaryStreamRequest[_T_RequestModel, _T_ResponseModel, _T_UploadModel]

The underlying request object that is used to perform this connection.

property result: Self

property serial: ServiceID | None: Deprecated.

service

Property type: Service

Returns the Service object through which this connection is established.

service_id

Property type: ServiceID | None

Returns the Service ID for the remote Service.

session_log_filename

Property type: str | None

Returns the file name of the current session log file if session logging is enabled, otherwise returns None.

session_log_filepath

Property type: Path | None

Returns the full path to the current session log file if session logging is enabled, otherwise returns None.

session_log_rotate( new_filepath: str | Path | None = None, ) → None: Closes existing session log and starts logging to a new log file on new_filepath, or a default one if not provided.

wait() → Self: Waits until the connection is established. Propagates exceptions.

wait_closed() → Self: Waits for the connection to be completely closed.

write_eof() → None: Writes EOF to the underlying stream. This closes the sockets on both sides.

class radkit_client.sync.FileTransferStatus

Bases: Enum

An enum of the possible statuses for a file transfer connection.

PREPARING = 'PREPARING': Connection setup in progress.

CONNECTED = 'CONNECTED': Connection open (connected).

UPLOADING = 'UPLOADING': Upload to device in progress.

DOWNLOADING = 'DOWNLOADING': Download from device in progress.

TRANSFER_DONE = 'TRANSFER_DONE': File transfer completed successfully.

ABORTED = 'ABORTED': Transfer aborted by user.

CLOSED = 'CLOSED': Connection closed (disconnected).

FAILURE = 'FAILURE': File transfer failed.

class radkit_client.sync.SessionsDict

Mapping (non-mutable dict) for displaying Terminal (interactive/SCP/SFTP) sessions.

client

Property type: Client

The Client object that the sessions in this mapping are bound to.

class radkit_client.sync.TerminalConnectionError

Bases: ClientError

Terminal (interactive/SCP/SFTP) connection error.

ExecSequence 

class radkit_client.sync.ExecSequence

Bases: object

Serializable definition of an execution flow that can be sent from radkit_client to radkit_service and executed over there.

This can be executed as part of a “terminal interaction” in radkit_client, and will be translated into a “terminal interaction” in radkit_service.

def run_command(command: str) -> ExecSequence:
    return (
        ExecSequence()
        .readuntil_prompt()
        .write(f'{command}\n')
        .start_capture()
        .readuntil_prompt()
    )

exec_sequence = (
    ExecSequence()
    .enter_privileged_mode()
    .include(run_command("show version"))
    .include(run_command("show ip"))
)

connection.do_remote_exec_sequence(exec_sequence, upload_to=upload_parameters)

# This is serialized so that we can pass it to the Service.

# On the service, we translates it into an interaction for execution on
# top of a PTY stream.
# It is possible to translate the exec sequence into an `Interaction`
# that runs on the client too, if the `Device` is not required (so,
# `enter_privileged_mode()` can't be used in that case.)

def interaction(device: Device) -> Interaction[None]:
    # Privileged mode.
    yield from enter_privileged_mode(device)

    # show version.
    yield from readuntil_prompt()
    data = yield from readuntil_prompt('show version')
    yield from capture(data)

    # Show IP.
    yield from readuntil_prompt()
    data = yield from readuntil_prompt('show ip')
    yield from capture(data)

actions: tuple[_Write | _WriteEOF | _Read | _ReadExactly | _ReadUntil | _ReadUntilRegex | _ReadLine | _ReadUntilTimeout | _ReadUntilPrompt | _Sleep | _Exec | _SetTerminalSize | _StartCapture | _EndCapture | _Capture | _CallProcedure | ExecSequence, ...] = ()

read( count: int = -1, *, timeout: float | None = None, on_timeout: TimeoutAction = TimeoutAction.FAIL, ) → ExecSequence

Read up to the given amount of bytes, or read until EOF if count is -1.

This will return an empty byte string (b'') when reaching EOF, similar to .read() on a file.

Parameters:: timeout – When given, stop waiting after this timeout. Either raise an exception (if on_timeout is TimeoutAction.FAIL), or ignore and proceed with the next action (if on_timeout is TimeoutAction.PROCEED).

readexactly( count: int, *, timeout: float | None = None, on_timeout: TimeoutAction = TimeoutAction.FAIL, ) → ExecSequence: Read exactly the given amount of bytes.

readuntil( pattern: bytes, *, timeout: float | None = None, on_timeout: TimeoutAction = TimeoutAction.FAIL, ) → ExecSequence: Read until the given substring is found.

readuntil_regex( pattern: bytes, *, timeout: float | None = None, on_timeout: TimeoutAction = TimeoutAction.FAIL, ) → ExecSequence: Read until the given regular expression pattern is found.

readline( *, timeout: float | None = None, on_timeout: TimeoutAction = TimeoutAction.FAIL, ) → ExecSequence: Read until the end of the current line.

readuntil_timeout( timeout: float, ) → ExecSequence: Read and keep reading until we don’t receive any more data for a timeout interval.

readuntil_prompt( timeout: float | None = None, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecSequence: Read and keep reading until we arrive at the next prompt (using automatic prompt detection).

write( data: str | bytes | CustomSecretStr, ) → ExecSequence: Write data into the input stream.

write_eof() → ExecSequence: Terminate the input stream by writing an “end of file”.

exec( command: str, sudo: bool = False, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecSequence: Perform a command execution and read the output until we arrive at the next prompt (using automatic prompt detection).

set_terminal_size( *, columns: int, rows: int, ) → ExecSequence: Set terminal size (only for connections that support it).

sleep( seconds: float, ) → ExecSequence: Sleep for a given number of seconds.

start_capture() → ExecSequence: Start automatic capturing of all data that’s read from the terminal using any of the read methods the rest of this ExecSequence (including nested sequences). We will stop capturing either at the end of this sequence or if end_capture() is called.

end_capture() → ExecSequence: Stop automatic capturing.

capture( data: str | bytes, ) → ExecSequence

Send the given string into the capture stream.

This is mainly useful for debugging, or for adding annotations into the captured stream.

call_procedure( name: str, ) → ExecSequence

Call any of the available procedures.

For instance:

.call_procedure("enable-privileged-mode")
.call_procedure("disable-pager")
.call_procedure("maglev-login")

enter_privileged_mode() → ExecSequence

Enter privileged mode, using the “enable password”.

Identical to .call_procedure("enable-privileged-mode").

include( exec_sequence: ExecSequence, ) → ExecSequence: Include another ExecSequence.

classmethod from_serialized( serialized_exec_sequence: SerializedExecSequence, ) → Self

class radkit_client.sync.DeviceInformation

Bases: object

Device specific information passed to an execution sequence or procedures for execution.

This is information that’s typically only available on the RADKit service. So, while the ExecSequence is sent from the client to the service for execution, this device information is filled in from what we have in our inventory to enable the execution.

All fields are optional here. Because potentially we don’t have some information available. Or we are actually execution in radkit_client where none of this is available.

device_type: DeviceType | None = None

terminal_username: str | None = None

terminal_password: CustomSecretStr | None = None

enable_set: bool | None = None

enable_password: CustomSecretStr | None = None

http_username: str | None = None

http_password: CustomSecretStr | None = None

enable_timeout: float | None = None

provisioning_variant: ProvisioningVariant = 'DEFAULT'

class radkit_client.sync.TimeoutAction

Bases: str, Enum

What to do when we get a timeout during an ExecSequence execution.

FAIL = 'FAIL'

PROCEED = 'PROCEED'

RETURN_SUCCESSFUL = 'RETURN_SUCCESSFUL'

class radkit_client.sync.PromptDetectionStrategy

Bases: object

Prompt detection strategy definition that can be used with:

Device.exec()
DeviceDict.exec()
ExecSequence.exec()
ExecSequence.readuntil_prompt()

Note

This API is experimental. The available prompt detection strategies and their respective options will be documented in a future release.

Example usage:

prompt_detection_strategy = PromptDetectionStrategy(
    name="test-probe-exponential-backoff"
    options={
        "max_probes": 41
        "backoff_algorithm": {
            "base": 1.15
        }
    }
)
device.exec("show clock", prompt_detection_strategy=prompt_detection_strategy)

Or, with the ExecSequence:

prompt_detection_strategy = PromptDetectionStrategy(
    name="test-probe-exponential-backoff"
    options={
        "max_probes": 41
        "backoff_algorithm": {
            "base": 1.15
        }
    }
)
seq = (
    ExecSequence()
        .start_capture()
        .exec("show clock", prompt_detection_strategy=prompt_detection_strategy)
)
result = terminal.run_remote_exec_sequence(seq)

name: str: Name of the prompt detection strategy.

options: dict[str, Any] | None = None: Prompt detection strategy options.

Upload parameters 

UploadParameters objects allow RADKit Client to instruct RADKit Service to send the results of command execution, file transfers or other operations directly to a remote destination, rather than back to the Client; this is especially useful for large outputs that need to be stored as attachments to a support case.

At the moment, the only supported upload destination is Cisco’s CX Drive (CXD), a cloud service that enables the authenticated and encrypted upload of data directly to a TAC case. See CXD integration for more information.

Note

Possible destinations for HTTP uploads are restricted by default. This restriction can be further fine-tuned on the Service through the cloud_client.upload_allowed_hosts setting.

radkit_client.sync.UploadParameters: A Union between DEVNULLUploadParameters, HTTPUploadParameters and GenericUploadParameters.

pydantic model radkit_client.sync.DEVNULLUploadParameters

Bases: BaseModel

“Uploader” that doesn’t upload anything anywhere. Useful if a command has a lot of output, but we don’t actually need the result.

field type: Literal['DEVNULLUploadParameters/1.0'] = 'DEVNULLUploadParameters/1.0'

pydantic model radkit_client.sync.HTTPUploadParameters

Bases: BaseModel

Upload parameters for HTTP(S).

field type: Literal['HTTPUploadParameters/1.0'] = 'HTTPUploadParameters/1.0'

field url: str = ''

The HTTP URL to upload to.

Validated by:

validate_url

field verb: str = 'PUT': The HTTP verb to use.

field headers: dict[str, str] [Optional]: A dict of HTTP headers to add to the request.

pydantic model radkit_client.sync.GenericUploadParameters

Bases: BaseModel

Upload parameters for non-HTTP protocols. Currently not supported.

field type: Literal['GenericUploadParameters/1.0'] = 'GenericUploadParameters/1.0'

field url: str = ''

CXD integration 

RADKit Client and Service integrate with Cisco’s CX Drive (CXD), a cloud service that enables the authenticated and encrypted upload of data directly to a TAC case. This integration can be leveraged from RADKit Client in two ways: either by remotely instructing the Service to upload the results/outputs of Client-initiated operations to CXD, or by uploading directly to CXD from the Client itself. Both are made possible by a CXD object, which must be properly authenticated before data can be uploaded.

The RADKit Client REPL provides an instance of the CXD class below called cxd as part of the initial scope.

For more information, see CX Drive (CXD).

class radkit_client.sync.CXD

Bases: SyncWrapper[AsyncCXD]

The primary class for CXD support. This allows uploading of data to a TAC SR number, CS engagement ID or a BCS Collector ID.

Typical usage starts by calling sso_login(), and then navigating to the CXD object like this:

with radkit_client.sync.Client.create() as client:
    # Login.
    client.sso_login("username@example.com")
    # Navigate to the CXD object.
    cxd = client.integrations.cxd
    # Call `exec` on a device (from a service), passing `upload_to=`.
    device.exec(command, upload_to=cxd(target, filename))

Data or files can be uploaded directly from here using the upload_to_cxd() method. But it’s also possible to obtain HTTPUploadParameters using get_upload_parameters() for sending to a RADKit service as part of an RPC call. That way, the service can start an upload to that destination. (HTTPUploadParameters is a CXD-independent concept from RADKit that represents an upload-destination that the radkit-service understands.)

Both upload_to_cxd() and get_upload_parameters() derive the HTTPUploadParameters from a ‘CXD target’. A CXD target is identified as a string, and is a TAC SR number, a CS engagement ID or a BCS Collector ID.

In order to be able to upload to a CXD target, we have to authenticate to CXD. If the target is not added explicitly, then the default authenticator is used, which is by default the cloud connection established through sso_login().

Targets can be added manually, through any of these functions:

add_target_from_upload_token(): Add a target using a CXD upload token for that target specifically. This allows uploading to one CXD target without requiring authentication to CXD.
add_target_from_cloud_connection(): Add a target, and use an authenticated cloud connection from sso_login().
add_target_from_oauth_token(): Add a target using an OAuth bearer token.
add_target_from_client_id_secret(): (Cisco internal only).
add_target_from_default_authenticator(): Add a target using the default authenticator.

The default authenticator can be set explicitly using either: set_default_authenticator_from_cloud_connection(), set_default_authenticator_from_oauth_token() or set_default_authenticator_from_client_id_secret(). But if those were not called, then the default current cloud connection will be used.

set_default_authenticator_from_cloud_connection( connection: CloudConnection, ) → None: Create a new CXD authenticator using an already established cloud connection, that was established using client.sso_login() (or any of the other radkit-client login functions.)

set_default_authenticator_from_oauth_token( token: str, domain: Domain | str | None = None, ) → None: Create a new CXD authenticator using an oauth bearer token.

set_default_authenticator_from_client_id_secret( client_id: Identity, client_secret: str, user_email: Email, domain: Domain | str | None = None, ) → None: Create a new CXD authenticator using client id/secret, and set it as the default authenticator for new targets (Cisco internal only).

default_authenticator

Property type: AnyCXDAuthenticator | None

Return the default authenticator.

If a default authenticator is set, return that, otherwise, create one from the current default radkit-client cloud connection. If not yet connected to the cloud, this will raise an exception.

add_target_from_upload_token( target: str, token: str, hostname: str = cxd.cisco.com, domain: Domain | str | None = None, set_default: bool = False, ) → None

Add a target manually, using a CXD upload token for this target. (A CXD token is only valid for one target.) This is useful when the user does not have the rights needed to retrieve the token, but someone issued them an upload token.

Parameters:

token – The upload token to use for this target.
hostname – The hostname to be used in the upload URL. If not set, the default hostname will be used. This is recommended for add_target_from_token, because tokens might be issued to specific hosts.
set_default – Set this target as the default target for uploads.

add_target_from_client_id_secret( target: str, client_id: Identity, client_secret: str, user_email: Email, domain: Domain | str | None = None, set_default: bool = False, ) → None

Add a CXD target, authenticated using a dedicated client ID and secret, provisioned by the CXD team (Cisco internal only).

Parameters:

target – a TAC SR number, CS engagement ID or BCS Collector ID.
client_id – The client ID to use for this target.
client_secret – The client secret to use for this target.
user_email – The email address of the user to use for this target.
domain – The domain to use for this target. If not set, the default domain will be used.
set_default – Set this target as the default target for uploads.

add_target_from_oauth_token( target: str, token: str, domain: Domain | str | None = None, set_default: bool = False, ) → None

Add a CXD target, authenticated using an oauth token.

Parameters:

target – a TAC SR number, CS engagement ID or BCS Collector ID.
token – OAuth bearer token.
set_default – Set this target as the default target for uploads.

add_target_from_default_authenticator( target: str, set_default: bool = False, ) → None

Add a target using the default authenticator.

To set a default authenticator, use: set_default_authenticator_from_cloud_connection(), set_default_authenticator_from_oauth_token() or set_default_authenticator_from_client_id_secret().

Parameters:

target – a TAC SR number, CS engagement ID or BCS Collector ID.
set_default – Set this target as the default target for uploads.

add_target_from_cloud_connection( target: str, connection: CloudConnection | None = None, set_default: bool = False, ) → None

Add a CXD target, authenticated using an already authenticated radkit-client connection. (using sso_login). This is the easiest and recommended way to add a target.

Parameters:

target – a TAC SR number, CS engagement ID or BCS Collector ID.
connection – An from the current radkit-client cloud connections. If None, then the default cloud connection will be used.
set_default – Set this target as the default target for uploads.

targets

Property type: CXDTargetsDict

Returns a dictionary of targets and their associated upload parameters.

remove_target( target: str, ) → None: Remove a target from targets.

set_default_target( target: str, ) → None

Set the default CXD target.

Raises an CXDTargetNotFoundError if this target was not yet added.

Parameters:: target – a TAC SR number, CS engagement ID or BCS Collector ID.

default_target

Property type: AnyCXDTarget | None

Return the default CXD target, or None if no default target was set.

get_upload_parameters( target: str | None, file_name: str, *, prefix: str = radkit, notify_applications: list[str] | str | None = None, ) → HTTPUploadParameters

Get upload parameters for a CXD target. These can be send over from radkit-client to a radkit-service, so that the service can perform the upload from there. The returned HTTPUploadParameters can be passed to e.g., exec() or scp_to_destination.

Parameters:

target – A TAC SR number, CS engagement ID or BCS Collector ID. When None, the default target for the CXD object is used.
file_name – The destination attachment name on CXD.
prefix – A string to prefix the destination filename with (default: radkit).
notify_applications – A list of application names that the CXD backend should notify when the upload is complete.

Raises:

CXDError – (multiple sub-classes)

Uploads local data or a local file to CXD.

Either local_file_name or data should be given.

Parameters:

target – A TAC SR number, CS engagement ID or BCS Collector ID. When None, the default target for the CXD object is used.
local_file_name – The path to a local file to be uploaded. If omitted, data must be provided.
data – A str or bytes containing the data to upload to CXD. If omitted, local_file_name must be provided.
destination_file_name – The destination attachment name on CXD. If omitted, and local_file_name was provided, the destination file will be named after the source file.
prefix – A string to prefix the destination filename with (default: radkit).
notify_applications – A list of application names that the CXD backend should notify when the upload is complete.

class radkit_client.sync.CXDError

Bases: ClientError

The base class for all CXD exceptions.

Netconf 

Similar to exec(), the chain of objects returned for Netconf queries depends on the parameters of the request (single or multiple devices, and single or multiple XPaths):

single_device.netconf.yang[...].get()
- status: RequestStatus
- result: GetSingleXPathResult
  
  status: NetconfResultStatus
  
  yang: YangDataMapping
multiple_devices.netconf.yang[...].get()
- status: RequestStatus
- result: DeviceToSingleXPathResultDict
  
  [device]: GetSingleXPathResult
  
  status: NetconfResultStatus
  
  yang: YangDataMapping
single_device.netconf.get_xpaths(multiple_xpaths)
- status: RequestStatus
- result: GetXPathsResult
  
  [xpath]: GetSingleXPathResult
  
  status: NetconfResultStatus
  
  yang: YangDataMapping
multiple_devices.netconf.get_xpaths(multiple_xpaths)
- status: RequestStatus
- result: DeviceToXPathResultsDict
  
  [device]: GetXPathsResult
  
  [xpath]: GetSingleXPathResult
  
  status: NetconfResultStatus
  
  yang: YangDataMapping

class radkit_client.sync.SingleDeviceNetconfAPI

Bases: SyncWrapper[AsyncSingleDeviceNetconfAPI]

Represents the Netconf API of a single device.

status

Property type: NetconfAPIStatus

The status of the Netconf capabilities retrieval for this device.

yang

Property type: SingleDeviceYangNode

The Netconf data model for this device.

capabilities

Property type: NetconfCapabilities

The Netconf capabilities for this device.

get_xpaths( xpaths: str | list[str], namespaces: dict[str, str] | None = None, ) → AsyncTransformedFillerRequest[DeviceAction[GetNetconfXpaths], DeviceActionPartialResponse[dict[str, Any]], AsyncDeviceToXPathResultsDict, AsyncGetXPathsResult] | AsyncTransformedFillerRequest[DeviceAction[GetNetconfXpaths], DeviceActionPartialResponse[dict[str, Any]], AsyncDeviceToXPathResultsDict, AsyncGetSingleXPathResult]

Runs a Netconf GET query on one or more XPaths in a single request.

Takes either a single XPath or a list of XPaths as the first argument. The second (optional) argument is a dict of namespace-to-URI mappings.

This function attempts to automatically gather namespace URIs based on what it finds in the queried XPaths and the device capabilities, if known.

Any additional namespaces that need to be provided with the Netconf query may be passed as the second argument, and will be added to the Netconf query for each individual XPath. If a namespace is found in both the device capabilities and the list of extra namespaces, the latter takes precedence.

Possible reasons for having to provide extra namespaces include:

performing a query without first retrieving the device capabilities;
mentioning a namespace within an XPath predicate or function call.

Examples:

# Single XPath, capabilities known
get_xpaths("/ietf-interfaces:interfaces")

# Multiple XPaths, capabilities known
get_xpaths([ "/ietf-interfaces:interfaces", "/openconfig-system:system" ])

# Single XPath, namespace specified explicitly
get_xpaths("/foo:bar", {"foo": "http://foo.bar.baz"})

Parameters:

xpaths – a single (str) or multiple (list[str]) XPath(s).
namespaces – optional – the namespaces corresponding to the XPaths (see above).

class radkit_client.sync.NetconfAPI

Bases: SyncWrapper[AsyncNetconfAPI]

Represents the collective Netconf API for a set of devices. This API can only be used if the devices share the same capabilities.

status

Property type: NetconfAPIStatus

The status of the Netconf capabilities retrieval for these devices.

yang

Property type: YangNode

The Netconf data model for these devices.

capabilities

Property type: NetconfCapabilities

The Netconf capabilities for these devices.

get_xpaths( xpaths: str | list[str], namespaces: dict[str, str] | None = None, synced: bool = False, ) → AsyncFillerRequest[DeviceAction[GetNetconfXpaths], DeviceActionPartialResponse[dict[str, Any]], AsyncDeviceToXPathResultsDict] | AsyncTransformedFillerRequest[DeviceAction[GetNetconfXpaths], DeviceActionPartialResponse[dict[str, Any]], AsyncDeviceToXPathResultsDict, AsyncDeviceToSingleXPathResultDict]

Runs a Netconf GET query on one or more XPaths for all devices in the set.

The parameters are the same as for SingleDeviceNetconfAPI.get_xpaths(), with the addition of synced.

Parameters:: synced – optional – True if the query must be performed simultaneously on all devices in the set (default: False).

class radkit_client.sync.NetconfAPIStatus

Bases: Enum

The possible statuses for a single- or multi-device Netconf API.

UNKNOWN = 'UNKNOWN': Netconf capabilities not known.

NO_CONFIG = 'NO_CONFIG': Netconf credentials/config ARE NOT available on Service.

CONFIGURED = 'CONFIGURED': Netconf credentials/config ARE available on Service.

AVAILABLE = 'AVAILABLE': Netconf capabilities downloaded successfully.

UNAVAILABLE = 'UNAVAILABLE': Error downloading Netconf capabilities.

HETEROGENOUS = 'HETEROGENOUS': Devices in the set have different or missing Netconf capabilities.

class radkit_client.sync.NetconfCapabilities

Bases: SyncWrapper[AsyncNetconfCapabilities]

Dataclass that holds the Netconf capabilities for a single device.

Warning

The only attribute that can be safely displayed in its entirety is hash, the others are usually way too large.

raw

Property type: dict[str, Any] | None

The raw Netconf capabilities, as received from the device, returned as a tree structure.

>>> device.netconf.capabilities.raw['ietf-interfaces:']
    {
        "interfaces": {"interface": {"name": None, "description": None, "type": None, "enabled": None, "link-up-down-trap-enable": None}},
        "interfaces-state": {
            "interface": {
                "name": None,
                "type": None,
                "admin-status": None,
                "oper-status": None,
                ...

namespaces

Property type: dict[str, str] | None

The Netconf namespaces, as received from the device. Returned as a dict where the key is the namespace name and the value is the URL/URN.

{
    "openconfig-mpls-sr": "http://openconfig.net/yang/mpls-sr",
    "Cisco-IOS-XE-policy": "http://cisco.com/ns/yang/Cisco-IOS-XE-policy",
    "common-mpls-types": "urn:ietf:params:xml:ns:yang:common-mpls-types",
    ...
}

hash

Property type: str | None

A hash of the device’s Netconf capabilities. This value is used to determine if a set of devices share the same Netconf capabilities and can be queried together.

class radkit_client.sync.SingleDeviceYangNode

Bases: SyncDictWrapper[AsyncSingleDeviceYangNode, str, AsyncSingleDeviceYangNode, SingleDeviceYangNode]

Represents a node in the dict representation of the device’s entire YANG data model, corresponding to a specific XPath that can be queried.

>>> device.netconf.yang['ietf-interfaces']['interfaces']
    SingleDeviceYangNode(device_name='device', xpath='/ietf-interfaces:interfaces')
    {
        "interface": {
            "name": {},
            "description": {},
            "type": {},
            "enabled": {},
            "link-up-down-trap-enable": {}
        }
    }

name

Property type: str

The name of this node.

>>> device.netconf.yang['ietf-interfaces']['interfaces'].name
"interfaces"

device_name

Property type: str

The canonical name of the device whose YANG model this node belongs to.

xpath

Property type: str

The full XPath leading to this node.

>>> device.netconf.yang['ietf-interfaces']['interfaces'].xpath
"/ietf-interfaces:interfaces"

namespaces

Property type: dict[str, str]

The namespaces relevant to this node in the YANG model.

>>> device.netconf.yang['ietf-interfaces']['interfaces'].namespaces
{"ietf-interfaces": "urn:ietf:params:xml:ns:yang:ietf-interfaces"}

device

Property type: Device

The Device object whose YANG model this node belongs to.

get() → TransformedFillerRequest[DeviceAction[GetNetconfXpaths], DeviceActionPartialResponse[dict[str, Any]], DeviceToXPathResultsDict, GetSingleXPathResult]: Performs a Netconf query on the XPath corresponding to this node in the data model.

class radkit_client.sync.YangNode

Bases: SyncDictWrapper[AsyncYangNode, str, AsyncYangNode, YangNode]

Represents a node in the dict representation of the entire YANG data model for a homogeneous set of devices, corresponding to a specific XPath that can be queried on all devices.

>>> devices.netconf.yang['ietf-interfaces']['interfaces']
    YangNode(xpath='/ietf-interfaces:interfaces')
    {
        "interface": {
            "name": {},
            "description": {},
            "type": {},
            "enabled": {},
            "link-up-down-trap-enable": {}
        }
    }

name

Property type: str

The name of this node.

>>> devices.netconf.yang['ietf-interfaces']['interfaces'].name
"interfaces"

devices

Property type: DeviceDict

The DeviceDict object whose YANG model this node belongs to.

xpath

Property type: str

The full XPath leading to this node.

>>> device.netconf.yang['ietf-interfaces']['interfaces'].xpath
"/ietf-interfaces:interfaces"

namespaces

Property type: dict[str, str]

The namespaces relevant to this node in the YANG model.

>>> devices.netconf.yang['ietf-interfaces']['interfaces'].namespaces
{"ietf-interfaces": "urn:ietf:params:xml:ns:yang:ietf-interfaces"}

get( synced: bool = False, ) → TransformedFillerRequest[DeviceAction[GetNetconfXpaths], DeviceActionPartialResponse[dict[str, Any]], DeviceToXPathResultsDict, DeviceToSingleXPathResultDict]

Performs a Netconf query on the XPath corresponding to this node in the data model.

Parameters:: synced – optional – True if the query must be performed simultaneously on all devices in the set (default: False).

class radkit_client.sync.GetSingleXPathResult

Bases: SyncWrapper[AsyncGetSingleXPathResult]

The result of a single XPath query on a single device.

raw

Property type: dict[str, Any]

The unprocessed YANG representation of the query results, in the form of a structured Python object:

>>> req.result.raw
{
    "interfaces": {
        "@xmlns": "urn:ietf:params:xml:ns:yang:ietf-interfaces",
        "interface": [
            {
                "name": "GigabitEthernet1",
                ...

yang

Property type: YangDataMapping

The processed YANG representation of the query result, in the form of a tree of YangDataMapping and YangDataSequence:

>>> req.result.yang
{
    "interfaces": {
        "@xmlns": "urn:ietf:params:xml:ns:yang:ietf-interfaces",
        "interface": {
            "GigabitEthernet1": {
                "name": "GigabitEthernet1",
                ...

In the output above, the interface node has been turned from a list into a dict by using the name field of each interface entry in the list as a key. This makes the data easier to process as everything can be subscripted using square brackets, for example:

>>> req.result.yang['interfaces']['interface']['GigabitEthernet1']['type']
{"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:ethernetCsmacd"}

See YangDataMapping for details about the logic and possible pitfalls.

status

Property type: NetconfResultStatus

Returns the status of this Netconf query. See NetconfResultStatus for values.

device_name

Property type: str

The canonical name of the device that this query was performed on.

device

Property type: Device

The Device object that this query was performed on.

wait( timeout: float | None = None, ) → Self: Waits for the query to complete. Propagates the exception if an error was encountered.

class radkit_client.sync.NetconfResultStatus

Bases: Enum

The possible statuses of a Netconf query.

FAILURE = 'FAILURE': The Netconf query failed.

SUCCESS = 'SUCCESS': The Netconf query succeeded.

PROCESSING = 'PROCESSING': The Netconf query is still in progress.

class radkit_client.sync.DeviceToSingleXPathResultDict

Bases: SyncDictWrapper[AsyncDeviceToSingleXPathResultDict, str, AsyncGetSingleXPathResult, GetSingleXPathResult]

Mapping from device names to GetSingleXPathResult.

class radkit_client.sync.DeviceToXPathResultsDict

Bases: SyncDictWrapper[AsyncDeviceToXPathResultsDict, str, AsyncGetXPathsResult, GetXPathsResult]

Mapping from device names to GetXPathsResult.

class radkit_client.sync.GetXPathsResult

Bases: SyncDictWrapper[AsyncGetXPathsResult, str, AsyncGetSingleXPathResult, GetSingleXPathResult]

The result of one or more XPath GET queries on a single device. Maps each XPath string to a GetSingleXPathResult object.

class radkit_client.sync.YangDataMapping

Bases: dict[str, object]

A set of key-value items in a processed Netconf response, in the form of a dict. Can be turned into JSON with |to_json or into a standard dict with |to_dict (see Pipe commands).

When processing the raw Netconf response from the device, when a list of dicts is encountered, this code attempts to turn it into a dict of dicts by auto-detecting the appropriate primary key.

Any field that is present in all dicts in the list and has a different value in every dict, is considered a possible primary key. If several candidate fields exist and one of those is called name, it is preferred, else any one of the possible keys is picked. If no candidate field is found at all, then the original list of dicts is included as a YangDataSequence.

For example, in this raw response to /ietf-interfaces:interfaces:

{
    "interfaces": {
        "@xmlns": "urn:ietf:params:xml:ns:yang:ietf-interfaces",
        "interface": [
            {
                "name": "GigabitEthernet1",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:ethernetCsmacd"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.10.10.1", "netmask": "255.255.255.192"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
            {
                "name": "GigabitEthernet2",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:ethernetCsmacd"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.20.20.1", "netmask": "255.255.255.0"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
            {
                "name": "Loopback1",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:softwareLoopback"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.2.1.1", "netmask": "255.255.255.0"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
            {
                "name": "Loopback2",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:softwareLoopback"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.2.2.1", "netmask": "255.255.255.0"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
        ],
    }
}

The name field is present and has a different value in each of the entries in the interface list, therefore it can be used as the key for turning this list of dicts into a dict of dicts, indexed by interface name:

{
    "interfaces": {
        "@xmlns": "urn:ietf:params:xml:ns:yang:ietf-interfaces",
        "interface": {
            "GigabitEthernet1": {
                "name": "GigabitEthernet1",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:ethernetCsmacd"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.10.10.1", "netmask": "255.255.255.192"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
            "GigabitEthernet2": {
                "name": "GigabitEthernet2",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:ethernetCsmacd"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.20.20.1", "netmask": "255.255.255.0"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
            "Loopback1": {
                "name": "Loopback1",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:softwareLoopback"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.2.1.1", "netmask": "255.255.255.0"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
            "Loopback2": {
                "name": "Loopback2",
                "type": {"@xmlns:ianaift": "urn:ietf:params:xml:ns:yang:iana-if-type", "#text": "ianaift:softwareLoopback"},
                "enabled": "true",
                "ipv4": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip", "address": {"ip": "10.2.2.1", "netmask": "255.255.255.0"}},
                "ipv6": {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-ip"},
            },
        },
    }
}

In some cases however, this logic may pick the “wrong” field, for instance in this list of CDP neighbors:

{
    "cdp-neighbor-details": {
        "@xmlns": "http://cisco.com/ns/yang/Cisco-IOS-XE-cdp-oper",
        "cdp-neighbor-detail": [
            {"device-id": "2041", "port-id": "Port 1"},
            {"device-id": "6124", "port-id": "GigabitEthernet1/0/1"},
            {"device-id": "8165", "port-id": "GigabitEthernet2/0/40"},
            {"device-id": "8605", "port-id": "GigabitEthernet0"},
        ],
    }
}

The code may pick port-id as the key, while the YANG model actually defines device-id as the key:

{
    "cdp-neighbor-details": {
        "@xmlns": "http://cisco.com/ns/yang/Cisco-IOS-XE-cdp-oper",
        "cdp-neighbor-detail": {
            "GigabitEthernet0": {"device-id": "8605", "port-id": "GigabitEthernet0"},
            "GigabitEthernet1/0/1": {"device-id": "6124", "port-id": "GigabitEthernet1/0/1"},
            "GigabitEthernet2/0/40": {"device-id": "8165", "port-id": "GigabitEthernet2/0/40"},
            "Port 1": {"device-id": "2041", "port-id": "Port 1"},
        },
    }
}

This is a known limitation that can only be properly solved by parsing the entire YANG model, which is something that will be made possible in a later version.

class radkit_client.sync.YangDataSequence

Bases: list[YangDataMapping]

A list of items in a processed Netconf response. This will only occur if a list was found in the raw data and it could not be turned into a YangDataMapping.

Can be turned into JSON with |to_json or into a standard dict with |to_dict (see Pipe commands).

class radkit_client.sync.NetconfAPIError

Bases: ClientError

Exception raised by the Netconf API.

SNMP 

class radkit_client.sync.SingleDeviceSNMP_API

Bases: SyncWrapper[AsyncSingleDeviceSNMP_API]

The SNMP API for a single device.

get( oids: p.RequestOID | Sequence[p.RequestOID], *, timeout: float | None = None, limit: int | None = None, retries: int | None = None, background_fetch: bool = True, ) → SingleDeviceSNMPResult

Get the values of one or more snmp OIDs.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1.1.0”.
- tuple of integers.
timeout – The timeout (in seconds) for the SNMP request.
limit – The maximum number of OIDs to look up in one request.
retries – How many times to retry the SNMP request if it times out.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.get("1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.2.0", "1.3.6.1.4.1.9.1.2349").wait().result

walk( oids: p.RequestOID | Sequence[p.RequestOID], *, timeout: float | None = None, limit: int | None = None, retries: int | None = None, concurrency: int = 100, background_fetch: bool = True, ) → SingleDeviceSNMPResult

Get the values of all keys rooted at the given OIDs. Implemented using:

GETNEXT (many requests) for SNMPv1 devices.
GETBULK (few requests) for SNMPv2 or above devices.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1”.
- tuple of integers.
timeout – The individual timeout (in seconds) for each SNMP request.
limit – The maximum number of SNMP entries to fetch in one request.
retries – How many times to retry each SNMP request that times out.
concurrency – The maximum number of queries to fetch at once.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.walk("1.3.6.1.2.1.1").wait().result

Get the values after one or more snmp OIDs.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1.1.0”.
- tuple of integers.
timeout – The timeout (in seconds) for the SNMP request.
limit – The maximum number of OIDs to look up in one request.
retries – How many times to retry the SNMP request if it times out.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.get_next("1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.2.0", "1.3.6.1.4.1.9.1.2349").wait().result

Get the next several values after each of the given OIDs. Raises an error if performed on a SNMP v1 device.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1”.
- tuple of integers.
timeout – The timeout (in seconds) for the SNMP request.
limit – The number of SNMP entries to get after each OID.
retries – How many times to retry each SNMP request that times out.
concurrency – The maximum number of queries to fetch at once.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.get_bulk("1.3.6.1.2.1.1").wait().result

device_name

Property type: str

The canonical name of the device that this query was performed on.

device

Property type: Device

The Device object that this query was performed on.

class radkit_client.sync.SNMP_API

Bases: SyncWrapper[AsyncSNMP_API]

The SNMP API for a set of devices.

get( oids: p.RequestOID | Sequence[p.RequestOID], *, timeout: float | None = None, limit: int | None = None, retries: int | None = None, background_fetch: bool = True, ) → SNMPResult

Get the values of one or more snmp OIDs.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1.1.0”
- tuple of integers
timeout – The timeout (in seconds) for the SNMP request.
limit – The maximum number of OIDs to look up in one request.
retries – How many times to retry the SNMP request if it times out.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.get("1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.2.0", "1.3.6.1.4.1.9.1.2349").wait().result

walk( oids: p.RequestOID | Sequence[p.RequestOID], *, timeout: float | None = None, limit: int | None = None, retries: int | None = None, concurrency: int = 100, background_fetch: bool = True, ) → SNMPResult

Get the values of all keys rooted at the given OIDs. Implemented using:

GETNEXT (many requests) for SNMPv1 devices.
GETBULK (few requests) for SNMPv2 or above devices.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1”.
- tuple of integers.
timeout – The individual timeout (in seconds) for each SNMP request.
limit – The maximum number of SNMP entries to fetch in one request.
retries – How many times to retry each SNMP request that times out.
concurrency – The maximum number of queries to fetch at once.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.walk("1.3.6.1.2.1.1").wait().result

Get the values after one or more snmp OIDs.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1.1.0”
- tuple of integers
timeout – The timeout (in seconds) for the SNMP request.
limit – The maximum number of OIDs to look up in one request.
retries – How many times to retry the SNMP request if it times out.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.get_next("1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.2.0", "1.3.6.1.4.1.9.1.2349").wait().result

Get the next several values after each of the given OIDs. Raises an error if performed on a SNMP v1 device.

Parameters:

oids –
one or more of:
- dot-separated strings of numbers: “1.3.6.1.2.1.1”.
- tuple of integers.
timeout – The timeout (in seconds) for the SNMP request.
limit – The maximum number of SNMP entries to get after each OID.
retries – How many times to retry each SNMP request that times out.
concurrency – The maximum number of queries to fetch at once.
background_fetch – If True (default), immediately start a background fetch, by calling SNMPResult.resume_fetch(). See there for more advanced usage.

Returns:

a response containing SNMPTable of the SNMP values.

Example:

>>> device.snmp.get_bulk("1.3.6.1.2.1.1").wait().result

class radkit_client.sync.SNMPTable

Bases: Mapping[int, _T_SNMPRow]

A general container for SNMPRow. Can be assembled manually from any collection of SNMPRow objects.

Can be indexed only by row number.

classmethod join( tables: Collection[SNMPTable[_T_SNMPRow]], ) → SNMPTable[_T_SNMPRow]: Combine multiple SNMPTable into one.

errors() → SNMPTable[SNMPErrorRow]: Filter out the regular rows, returning only errors

without_errors() → SNMPTable[SNMPValueRow]: Filter out the error rows, returning only real rows

class radkit_client.sync.SingleDeviceSNMPTable

Bases: Mapping[int, _T_SNMPRow]

Wrapper to SNMPTable that does not show the device name when printed

errors() → SingleDeviceSNMPTable[SNMPErrorRow]: Filter out the regular rows, returning only errors

without_errors() → SingleDeviceSNMPTable[SNMPValueRow]: Filter out the error rows, returning only real rows

class radkit_client.sync.SNMPRow

Bases: object

A single SNMP entry on a single device.

device_name: str: The name of the device this result came from.

oid: tuple[int, ...] | None = None: My object id, as a tuple of integers: (1, 3, 6, 1, 2, 1, 1, 3, 0).

label: tuple[str, ...] = (): My MIB-resolved object id, as a tuple of strings: ("iso", "org", "dod", "internet", "mgmt", "mib-2", "system", "sysUpTime").

mib_module: str = '': My MIB module: "SNMPv2-MIB".

mib_variable: str = '': My MIB variable: "sysUpTime".

mib_indices: Sequence[Sequence[int]] = (): My MIB indices: ((0,),)

type: str = '': A string description of my ASN.1 type: "TimeTicks".

value: Any = None: The value of this entry: 26721423.

error_code: SNMPNetworkErrorCode | SNMPResponseErrorCode | SNMPVarBindErrorCode = 0: The error code of this entry. SNMPResponseErrorCode.NO_ERROR if no error.

property oid_str: str: My object id, as a dot-separated string: "1.3.6.1.2.1.1.3.0".

property value_str: str: My value as a string.

property label_str: str: My MIB-resolved object id, as a dot-separated string: "iso.org.dod.internet.mgmt.mib-2.system.sysUpTime".

property mib_str: str: My MIB module::variable.indices: "SNMPv2-MIB::sysUpTime.0".

property is_error: bool: Whether or not this row contains an error.

property error_str: str: My error code as a short string.

property value_or_error_str: str: My value as a string, or my error code as a short string.

property value_or_error_msg: str: My value as a string, or my error code with message.

class radkit_client.sync.SNMPQuerySet

Bases: SyncDictWrapper[AsyncSNMPQuerySet, int, AsyncSNMPQuery, SNMPQuery]

A list of SNMP queries on one or more devices being run in parallel, along with their status

stats

Property type: SNMPQueryStatistics

Return a summary of the current state of the running query set

class radkit_client.sync.SNMPQuery

Bases: SyncWrapper[AsyncSNMPQuery]

Contains the state of an individual query in a SNMPQuerySet

device_name

Property type: str

The canonical name of the device that this query was performed on.

device

Property type: Device

The Device object that this query was performed on.

description

Property type: str

Query description

result_count

Property type: int

How many rows of data have been received from the device, so far in this run.

done

Property type: bool

True if the query has completed successfully, False otherwise. Queries that are done will be ignored during re-runs

request_count

Property type: int

How many requests have been sent to the device, so far in this run.

response_count

Property type: int

How many responses have been received from the device, so far in this run.

response_time

Property type: float

The total time spend waiting for the device to respond to all service requests, so far in this run.

dropped_packets

Property type: int

How many requests never received a response from the device, so far in this run.

ping_time

Property type: float

Average time to receive a response from the device, so far in this run.

failed_count

Property type: int

How many runs of this query have ended in failure

status

Property type: SNMPResultStatus

Returns the status of this SNMP query. See SNMPResultStatus for values.

total_row_count

Property type: int

How many rows have been received from the device, so far, in this and previous runs. Counts everything: data rows, error rows, and rows filtered out by the query

error_messages

Property type: list[str]

The error messages from the latest run of this query. Blank if in-progress, or successful

short_error_message

Property type: str

The first error message, but truncated to one line

raw_response

Property type: bytes

The latest SNMP response received, as a unparsed, BER-encoded protocol data unit. Blank if no responses received yet.

class radkit_client.sync.SNMPQueryStatistics

Bases: object

Various read-only statistics about the state of an in-progress SNMPQuerySet

failed_sessions: int: The number of sessions that have completed without receiving all results

all_results: int: The total number of results received from all my queries, including data, errors, filtered results, and previous runs

success_results: int: The number of results successfully received from all my queries during the latest run

error_results: int: The number of result errors received from all my queries during the latest run

packets_received: int: The number of packets received from all my queries during the latest run

packets_dropped: int: The number of packets dropped from all my queries during the latest run

ping_time: float: The average ping time of all my queries during the latest run

total_queries: int: The number of queries total

failed_queries: int: The number of queries that have completed without receiving all results

finished_queries: int: The number of queries that have completed and receiving all results

running_queries: int: The number of queries that are currently running

unstarted_queries: int: The number of queries that have not started

status: SNMPResultStatus: The status of these SNMP queries. See SNMPResultStatus for values.

first_error_message: str: The first error message from my queries. Blank if none

classmethod from_queries( queries: Sequence[AsyncSNMPQuery], ) → Self

progress_message() → str: A message summarizing the progress of the queryset

class radkit_client.sync.SingleDeviceSNMPResult

Bases: SyncWrapper[AsyncSingleDeviceSNMPResult]

The result of a running set of queries on a single device

result

Property type: SingleDeviceSNMPTable[SNMPValueOrErrorRow]

Returns the data retrieved from the SNMP devices so far. Will be incomplete until status() returns SNMPResultStatus.SUCCESS

queries

Property type: SNMPQuerySet

Returns detailed status info about the individual queries being composed together into my result

resume_fetch( concurrency: int = 100, timeout: float | None = None, limit: int | None = None, retries: int | None = None, ) → None

Start a background fetch, storing the results in my result. This starts any unfinished queries and retries any failed queries

This method is called automatically by all SNMP_API methods unless you pass background_fetch=False. Use False in scripts where you want to consume the results directly using SNMPQuerySet.start_session(), without storing them in memory (async-only)

Parameters:

concurrency – The maximum number of queries to fetch at once.
timeout – The individual timeout (in seconds) for each SNMP request.
limit – The maximum number of SNMP entries to fetch in one request.
retries – How many times to retry each SNMP request that times out.

pause_fetch() → None: Pause the current background fetch

wait( timeout: float | None = None, ) → Self: Waits for the current background fetch to complete

device_name

Property type: str

The canonical name of the device that this query was performed on.

device

Property type: Device

The Device object that this query was performed on.

class radkit_client.sync.SNMPResult

Bases: SyncWrapper[AsyncSNMPResult]

The result of a running set of queries on a set of devices

aggregate

Property type: SNMPTable[SNMPValueOrErrorRow]

Returns the data retrieved from the SNMP devices so far, not grouped by device. Will be incomplete until status() returns SNMPResultStatus.SUCCESS

result

Property type: SNMPTableDeviceView

Returns the data retrieved from the SNMP devices so far, grouped by device. Will be incomplete until status() returns SNMPResultStatus.SUCCESS

queries

Property type: SNMPQuerySet

Returns detailed status info about the individual queries being composed together into my result

resume_fetch( concurrency: int = 100, timeout: float | None = None, limit: int | None = None, retries: int | None = None, ) → None

Start a background fetch, storing the results in my result. This starts any unfinished queries and retries any failed queries

This method is called automatically by all SNMP_API methods unless you pass background_fetch=False. Use False in scripts where you want to consume the results directly using SNMPQuerySet.start_session(), without storing them in memory (async-only)

Parameters:

concurrency – The maximum number of queries to fetch at once.
timeout – The individual timeout (in seconds) for each SNMP request.
limit – The maximum number of SNMP entries to fetch in one request.
retries – How many times to retry each SNMP request that times out.

pause_fetch() → None: Pause the current background fetch

wait( timeout: float | None = None, ) → Self: Waits for the current background fetch to complete

class radkit_client.sync.SNMPTableDeviceView

Bases: Mapping[str, SingleDeviceSNMPTable[SNMPValueRow | SNMPErrorRow]]

A view on a SNMPTable that groups the rows by device

keys() → a set-like object providing a view on D's keys

values() → an object providing a view on D's values

items() → a set-like object providing a view on D's items

class radkit_client.sync.SNMP_APIError

Bases: ClientError

Exception raised by the SNMP API.

class radkit_client.sync.SNMPResultStatus

Bases: Enum

The possible statuses of an SNMP query.

FAILURE = 'FAILURE': The SNMP query failed.

SUCCESS = 'SUCCESS': The SNMP query succeeded.

PARTIAL_SUCCESS = 'PARTIAL_SUCCESS': The SNMP query returned some results but was unable to finish due to timeout or query limit.

PROCESSING = 'PROCESSING': The SNMP query is still in progress.

PAUSED = 'PAUSED': The SNMP query was paused and needs to be resumed.

NOT_YET_SUBMITTED = 'NOT_YET_SUBMITTED': The SNMP query wasn’t sent, or was made to zero devices.

classmethod join( statuses: Iterable[SNMPResultStatus], ) → SNMPResultStatus

Swagger and HTTP 

Data types 

There are a number of complex data types that are used for function arguments in the Swagger and HTTP APIs in RADKit, which are modeled on the httpx library’s own API. The purpose of these data types is to simplify the function signatures and clearly indicate what is/isn’t allowed for each argument.

Note

Only HttpVerb is actually needed for type-checking (see example); it can be imported from radkit_client.sync if you need it (typically only if you are using mypy). The other data types mentioned in this section are simply aliases that can readily be replaced with simpler types in API calls, such as ContentType with bytes in the example.

# Does not pass the type checker
d.swagger.call_path(path="/somepath", method="POST", content=b"something")
# error: Argument "method" has incompatible type "str"; expected "HttpVerb"  [arg-type]

# Passes the type checker
from radkit_client.sync import HttpVerb
d.swagger.call_path(path="/somepath", method=HttpVerb("POST"), content=b"something")

radkit_client.sync.HttpVerb: Used for HTTP method names in the definition of path operations. Expressed as an uppercase string: GET, POST, PATCH, PUT or DELETE.

ContentType

Used for the content argument to Swagger requests. Represents the raw data to be sent in the body of the HTTP request. Can be a str, bytes, or Iterable[bytes]. Example: content=b"mydata"

DataType

Key-value pairs to be form-encoded, expressed as a dict with str keys, and scalar values of types str, bytes, int, float or None (value types can be mixed). Example: data={"one":1, "two":"2"}

FilesType

Used to send the contents of a file over HTTP, similar to multipart file uploads in httpx.

FilesType = Union[Mapping[str, FileType], Sequence[tuple[str, FileType]]]

FileType = Union[
    # file (or bytes)
    FileContent,
    # (filename, file (or bytes))
    Tuple[Optional[str], FileContent],
    # (filename, file (or bytes), content_type)
    Tuple[Optional[str], FileContent, Optional[str]],
    # (filename, file (or bytes), content_type, headers)
    Tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]],
]

FileContent = Union[bytes, io.BufferedIOBase]

PathType

Used for the path portion of the HTTP request. Can be str or bytes.

ParamsType

Used to pass URL query parameters.

ParamsType = Union[
    Mapping[str, Union[PrimitiveData, Sequence[PrimitiveData]]],
    list[tuple[str, PrimitiveData]],
    Tuple[tuple[str, PrimitiveData], ...],
    str,
    bytes,
]

PrimitiveData = Optional[Union[str, int, float, bool]]

HeadersType

Used to pass headers for the HTTP request. Can be either a Dict[bytes, bytes] or a Dict[str, str] (i.e. a dict where keys and values must be the same type, either str or bytes).

CookiesType

Used to pass cookies for the HTTP request. Can be either a dict, or a CookieJar (from http.cookiejar).

CertType

Currently unused. Used to pass a certificate for TLS. Can be either str or Tuple[str, str].

Swagger 

Swagger support makes it possible to browse the device’s HTTP API from the Client side after parsing the device’s Swagger/OpenAPI model definition on the Service. It also provides authentication using Service-side credentials for supported device types.

class radkit_client.sync.SwaggerAPI

Bases: SyncWrapper[AsyncSwaggerAPI]

Represents the Swagger API of a device. This API is only usable if the Swagger model has been successfully retrieved from the device, parsed and returned by the Service as a result of Device.update_swagger().

Currently this API is only supported on a single device at a time, i.e. it is not possible to query Swagger paths on a DeviceDict, although it is possible to do update_swagger() on a device dict.

status

Property type: SwaggerAPIStatus

The status of this Swagger API. See SwaggerAPIStatus for possible values.

metadata

Property type: dict[str, Any]

Metadata returned by the Service alongside the Swagger model. This is used to provide additional information for specific device types.

paths

Property type: SwaggerPathsDict

A dict where the keys are the paths present in the device’s Swagger model, and the values are SwaggerPath objects that can be queried.

call_path( path: str, method: HttpVerb, parameters: dict[str, str] | None = None, content: ContentType | None = None, data: DataType | None = None, files: FilesType | None = None, json: Any = None, timeout: float | None = None, ) → FillerSwaggerRequest

Calls a Swagger path with any HTTP method. This function does check the path parameters against the Swagger model. You should favor the get(), post(), patch(), put() and delete() helpers when possible.

Parameters:

path – the Swagger path to request.
method – the HTTP method to use for the request (see Data types).
parameters – optional – a dict of parameters for this path (can be empty {}).
content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

get( path: str, parameters: dict[str, str] | None = None, timeout: float | None = None, ) → FillerSwaggerRequest

Wrapper around call_path() for GET requests.

Parameters:

path – the Swagger path to request.
parameters – optional – a dict of parameters for this path (can be empty {}).
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

Wrapper around call_path() for POST requests.

Parameters:

path – the Swagger path to request.
parameters – a dict of parameters for this path (can be empty {}).
content – request body or None (see Data types).
data – data to be form-encoded (see Data types).
files – file(s) to upload (see Data types).
json – an object to be serialized into JSON and sent in the request body.
timeout – the timeout for the request on the Service side, in seconds. Defaults to None i.e. the value configured on the Service will be used.

Wrapper around call_path() for PATCH requests.

Parameters:

path – the Swagger path to request.
parameters – optional – a dict of parameters for this path (can be empty {}).
content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

Wrapper around call_path() for PUT requests.

Parameters:

path – the Swagger path to request.
parameters – optional – a dict of parameters for this path (can be empty {}).
content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

delete( path: str, parameters: dict[str, str] | None = None, timeout: float | None = None, ) → FillerSwaggerRequest

Wrapper around call_path() for DELETE requests.

Parameters:

path – the Swagger path to request.
parameters – optional – a dict of parameters for this path (can be empty {}).
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

class radkit_client.sync.SwaggerAPIStatus

Bases: Enum

Possible statuses for AsyncSwaggerAPI.

UNKNOWN = 'UNKNOWN': Swagger capabilities not known.

NO_CONFIG = 'NO_CONFIG': Swagger credentials/config ARE NOT available on Service.

CONFIGURED = 'CONFIGURED': Swagger credentials/config ARE available on Service.

AVAILABLE = 'AVAILABLE': Swagger capabilities downloaded successfully.

UNAVAILABLE = 'UNAVAILABLE': Error downloading Swagger capabilities.

class radkit_client.sync.SwaggerPath

Bases: SyncWrapper[AsyncSwaggerPath]

Represents a single path in a device’s Swagger model. Each instance of this class will dynamically expose methods for the HTTP operations supported by this specific path. Those methods are similar to SwaggerAPI.get(), SwaggerAPI.post(), SwaggerAPI.patch(), SwaggerAPI.put() and SwaggerAPI.delete().

path

Property type: str

Returns this Swagger path as a string.

operations

Property type: SwaggerPathOperationsDict

Maps HTTP methods (GET, POST, PATCH, PUT, DELETE) to SwaggerPathOperation objects.

Performs an HTTP request on this Swagger path.

Parameters:

method – the HTTP method to use for the request (see Data types).
parameters – optional – a dict of parameters for this path (can be empty {}).
content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

get(

timeout: float | None = None,

**kwargs: str,

) → FillerSwaggerRequest

Performs a GET operation on this path. Path parameters must be passed as additional keyword arguments to this function.

Parameters:: timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

post(

content: ContentType | None = None,

data: DataType | None = None,

files: FilesType | None = None,

json: Any = None,

timeout: float | None = None,

**kwargs: str,

) → FillerSwaggerRequest

Performs a POST operation on this path. Path parameters must be passed as additional keyword arguments to this function.

Parameters:

content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

patch(

content: ContentType | None = None,

data: DataType | None = None,

files: FilesType | None = None,

json: Any = None,

timeout: float | None = None,

**kwargs: str,

) → FillerSwaggerRequest

Performs a PATCH operation on this path. Path parameters must be passed as additional keyword arguments to this function.

Parameters:

content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

put(

content: ContentType | None = None,

data: DataType | None = None,

files: FilesType | None = None,

json: Any = None,

timeout: float | None = None,

**kwargs: str,

) → FillerSwaggerRequest

Performs a PUT operation on this path. Path parameters must be passed as additional keyword arguments to this function.

Parameters:

content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be serialized into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

delete(

timeout: float | None = None,

**kwargs: str,

) → FillerSwaggerRequest

Performs a DELETE operation on this path. Path parameters must be passed as additional keyword arguments to this function.

Parameters:: timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

class radkit_client.sync.SwaggerPathOperation

Bases: SyncWrapper[AsyncSwaggerPathOperation]

Represents a specific operation on a specific Swagger path, along with its description and all its parameters, as defined by the device’s Swagger model.

description

Property type: str

Returns the description of this operation, as a string.

parameters

Property type: Sequence[SwaggerPathOperationParameterModel]

Returns the SwaggerPathOperationParameterModel for this operation.

call( parameters: dict[str, str] | None = None, content: ContentType | None = None, data: DataType | None = None, files: FilesType | None = None, json: Any = None, timeout: float | None = None, ) → FillerRequest[DeviceAction[p.CallSwaggerPath], DeviceActionPartialResponse[p.CallSwaggerPathResponse], SwaggerResponse]

Calls this Swagger path. Performs parameter validation.

Parameters:

parameters – optional – a dict of parameters for this path (can be empty {}).
content – optional – request body or None (see Data types).
data – optional – data to be form-encoded (see Data types).
files – optional – file(s) to upload (see Data types).
json – optional – an object to be turned into JSON and sent in the request body.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).

class radkit_client.sync.SwaggerResponse

Bases: SyncWrapper[AsyncSwaggerResponse]

The response to a Swagger request on a single device.

If some content or body data was provided in the request, it is only stored there, not here, first because it might be large, and second, because it might too easily be confused with a JSON response.

endpoint

Property type: str

The HTTP endpoint (path) that was queried in the request.

method

Property type: str

The HTTP method that was used to query the endpoint (path).

device

Property type: Device

The Device object that the request was performed on.

device_name

Property type: str

The canonical name of the device that the request was performed on.

client_id

Property type: ClientID | None

The identity that is connected to the remote Service for this request.

service_id

Property type: ServiceID | None

The ID of the Service that handled this request.

status_code

Property type: int

The HTTP status code, as an integer, i.e. 200, or 404.

Raises:: RequestPendingError – if the request is still pending.

status_phrase

Property type: str

The HTTP status code phrase, i.e. “OK”, or “Not Found”.

Raises:: RequestPendingError – if the request is still pending.

status_text

Property type: str

The HTTP status code and phrase, i.e. “200 OK”, or “404 Not Found”.

Raises:: RequestPendingError – if the request is still pending.

response_code

Property type: int

Alias for status_code(), for backwards compatibility.

url: Property type: str

content_type

Property type: str

The HTTP response Content-Type.

Raises:: RequestPendingError – if the request is still pending.

content

Property type: bytes

The data returned in the response, as bytes.

Raises:: RequestPendingError – if the request is still pending.

text

Property type: str

The data returned in the response, decoded from UTF-8.

Raises:: RequestPendingError – if the request is still pending.

json

Property type: dict[str, Any]

The data returned in the response, parsed from JSON.

Raises:

RequestPendingError – if the request is still pending.
JSONDecodeError – if the JSON decoding failed.

wait( timeout: float | None = None, ) → Self

Waits until the request has finished and the response is available.

Parameters:: timeout – how long to wait (in seconds) before raising TimeoutError.

property serial: ServiceID | None: Deprecated.

HTTP 

The raw HTTP API can be used when a Swagger/OpenAPI model is not provided by the device, or when Swagger support is not available on RADKit Service for that device type. It also provides authentication using Service-side credentials for supported device types.

class radkit_client.sync.HttpApi

Bases: SyncWrapper[AsyncHttpApi]

The HTTP API for the device.

Note

Changed in 1.6.2: this class was renamed from HTTP_API to HttpApi. Its methods such as get() or post() now return an HttpResponse, instead of a FillerRequest with a result attribute of type HTTPResponseResult. See HttpResponse for more information. The old name HTTP_API still exists for backwards compatibility, but it will be removed in a future version. Also, allow_redirects was removed from all methods.

Changed in 1.6.3: verify, cert and stream were removed from all methods.

Changed in 1.6.5: erase_auth_tokens was added.

Currently this API is only supported on a single device at a time, i.e. it is not possible to perform HTTP requests on a DeviceDict for multiple devices at a time.

The methods for get(), post() etc. follow the general model and signature of their equivalents in httpx, with the exception of certain parameters:

verify and cert are no longer accepted since 1.6.3. The TLS verification is driven by RADKit Service and is governed by the Service settings. RADKit Client delegates the entire TLS certificate validation and other verification steps to RADKit Service, which is the actual client in the TLS client-server model.
stream is not accepted (currently). The underlying code supports streaming as of RADKit 1.6.2, but streaming is not yet available to Client scripts as part of the HTTP API. This will be available in a future version.
allow_redirects is no longer accepted since 1.6.2. An implementation that follows redirects conflicts with the requirement to have end-to-end data streaming (in order to follow a redirect, all uploaded data must first be stored in memory, so that it can be re-uploaded to the new target). It is also incompatible with the upload_to parameter. Of course, when the HTTP API is exposed through the HTTP/SOCKS proxy, then a client that connects through this proxy (e.g., a browser) can still decide to follow redirects. HTTP clients like httpx and curl also do not follow redirects by default.

get( path: PathType, *, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP GET request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

options( path: PathType, *, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP OPTIONS request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

head( path: PathType, *, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP HEAD request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

post( path: PathType, *, content: ContentType | None = None, data: DataType | None = None, files: FilesType | None = None, json: Any = None, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP POST request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
content – optional – request body.
data – optional – data to be form-encoded.
files – optional – file(s) to upload.
json – optional – an object to be serialized into JSON and sent in the request body.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

patch( path: PathType, *, content: ContentType | None = None, data: DataType | None = None, files: FilesType | None = None, json: Any = None, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP PATCH request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
content – optional – request body.
data – optional – data to be form-encoded.
files – optional – file(s) to upload.
json – optional – an object to be serialized into JSON and sent in the request body.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

put( path: PathType, *, content: ContentType | None = None, data: DataType | None = None, files: FilesType | None = None, json: Any = None, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP PUT request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
content – optional – request body.
data – optional – data to be form-encoded.
files – optional – file(s) to upload.
json – optional – an object to be serialized into JSON and sent in the request body.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

delete( path: PathType, *, content: ContentType | None = None, data: DataType | None = None, files: FilesType | None = None, json: Any = None, params: ParamsType | None = None, headers: HeadersType | None = None, cookies: CookiesType | None = None, timeout: float | None = None, upload_to: UploadParameters | None = None, postprocessors: Sequence[str] = (), ) → HttpResponse

Sends an HTTP DELETE request to the device. Most of the parameters are passed indirectly to httpx.request. See Data types for details about the allowed types for each parameter.

Parameters:

path – path to HTTP endpoint, starting with /.
content – optional – request body.
data – optional – data to be form-encoded.
files – optional – file(s) to upload.
json – optional – an object to be serialized into JSON and sent in the request body.
params – optional – URL query parameters.
headers – optional – HTTP headers.
cookies – optional – HTTP cookies.
timeout – optional – the timeout for the request on the Service side, in seconds (default: None = the value configured on the Service will be used).
upload_to – optional – upload the response to a destination from within the Service.
postprocessors – optional – a list of postprocessor names (should exist on the Service). Postprocessors are functions that can transform the data before transferring.

erase_auth_tokens() → EraseAuthenticationTokensResult

overlay_headers

Property type: HttpOverlayHeadersDict

Used to add headers for every request.

Usage:

api = HttpApi(service_state, device_uuid)
api.overlay_headers["accept"] = "application/json"
api.get("/api/devices")

Is equal to api.get(“/api/devices”, headers={“accept”: “application/json”}) but no need in providing them every time a request is performed.

class radkit_client.sync.HttpApiError

Bases: ClientError

Exception raised by the raw HTTP API.

Note

Changed in 1.6.2: this class was renamed from HTTP_APIError to HttpApiError. The old name still exists for backwards compatibility, but it will be removed in a future version.

class radkit_client.sync.HttpResponse

Bases: SyncWrapper[AsyncHttpResponse]

The response to an HTTP request on a single device.

Note

Changed in 1.6.2: this class was renamed from HttpResponseResult to HttpResponse. It is now returned directly by the HttpApi methods such as get() or post(), instead of returning a FillerRequest with a result attribute of type HTTPResponseResult. The result attribute still exists in HttpResponse, but it will log a warning and return the same object to the caller.

If some content or body data was provided in the request, it is only stored there, not here, first because it might be large, and second, because it might too easily be confused with a JSON response.

device

Property type: Device

The Device object that the request was performed on.

device_name

Property type: str

The canonical name of the device that the request was performed on.

client_id

Property type: ClientID | None

The identity that is connected to the remote Service for this request.

service_id

Property type: ServiceID | None

The ID of the Service that handled this request.

status

Property type: RequestStatus

Status from the underlying RPC request.

property request: StreamRequest[DeviceAction[HTTPStreamRequest], HTTPStreamResponse, BinaryData] | FillerRequest[DeviceAction[HTTPRequest], DeviceActionPartialResponse[HTTPResponse], None]: The BaseRequest that was used to perform the HTTP RPC call. Depending on whether or not the service supports streaming HTTP, this will be a StreamRequest or a FillerRequest.

url

Property type: str

The HTTP response URL.

Raises:: RequestPendingError – if the request is still pending.

endpoint: Property type: str

method

Property type: str

The HTTP method that was used to query the endpoint (path).

status_code

Property type: int

The HTTP status code, i.e. 200, or 404.

Raises:: RequestPendingError – if the request is still pending.

status_phrase

Property type: str

The HTTP status code phrase, i.e. “OK”, or “Not Found”.

Raises:: RequestPendingError – if the request is still pending.

status_text

Property type: str

The HTTP status code and phrase, i.e. “200 OK”, or “404 Not Found”. If the code is 3XX Redirect, also show Location header

Raises:: RequestPendingError – if the request is still pending.

headers

Property type: CIMultiDict[str]

The HTTP response headers.

Raises:: RequestPendingError – if the request is still pending.

cookies

Property type: dict[str, str]

The HTTP response cookies.

Raises:: RequestPendingError – if the request is still pending.

content

Property type: bytes

The data returned in the response, as bytes.

Raises:: RequestPendingError – if the request is still pending.

content_type

Property type: str | None

The HTTP response Content-Type.

Raises:: RequestPendingError – if the request is still pending.

text

Property type: str

The data returned in the response, decoded from UTF-8.

Raises:: RequestPendingError – if the request is still pending.

json

Property type: Any

The data returned in the response, parsed from JSON.

Raises:

RequestPendingError – if the request is still pending.
JSONDecodeError – if the JSON decoding failed.

wait( timeout: float | None = None, ) → Self

Waits until the request has finished and the response is available.

Parameters:: timeout – how long to wait (in seconds) before raising TimeoutError.

property result: Self

property serial: ServiceID | None: Deprecated.

Proxying 

class radkit_client.sync.TCPPortForwarder

Bases: SyncWrapper[AsyncTCPPortForwarder]

A TCP port forwarder.

status

Property type: PortForwarderStatus

The status of the port forwarder. See PortForwarderStatus.

device

Property type: Device

Returns the Device object that this forwarder is bound to.

device_name

Property type: str

Returns the canonical name of the device that this forwarder is bound to.

local_port

Property type: int

Returns the local TCP port that this forwarder is forwarding traffic from.

destination_port

Property type: int

Returns the destination TCP port that this forwarder is forwarding traffic to.

start() → None

Starts the port forwarder and waits for it to be actually listening for incoming connections before returning.

Raises:: PortForwarderInvalidStateError – if the port forwarder is already started.

stop() → None: Stops the port forwarder and waits for it to fully terminate before returning.

bytes_uploaded

Property type: int

Returns the total number of bytes sent through this port forwarder so far.

bytes_downloaded

Property type: int

Returns the total number of bytes received through this port forwarder so far.

connections

Property type: Connections

All the connections made through this port forwarder so far, including closed ones, returned as a dict with a numerical index as the key.

get_dynamic_local_ports() → dict[str, int] | None

class radkit_client.sync.TCPPortForwardsDict

Bases: SyncDictWrapper[AsyncTCPPortForwardsDict, int, AsyncTCPPortForwarder, TCPPortForwarder]

Mapping (non-mutable dict) for displaying TCP port forwarders.

client

Property type: Client

The Client object that the port forwarders in this mapping are bound to.

class radkit_client.sync.PortRanges

Bases: str

A string that expresses one or more ranges of ports (TCP or UDP, depending on the context). Each range is either a single port number, or a low and a high port separated by a dash (-). Multiple ranges can be separated by semicolons (;).

A few examples: "8443", "1024-32767", "1024-2047;3456;8000-9999"

is_port_valid( port: int, ) → bool: Returns True if the port in argument is part of this range, False otherwise.

class radkit_client.sync.ProxyPortForwarder

Bases: SyncWrapper[AsyncProxyPortForwarder]

An HTTP or SOCKS5 proxy (a type of port forwarder).

Note

Changed in 1.6.0: this class was renamed from Socks5PortForwarder to ProxyPortForwarder since it now covers both HTTP and SOCKS5.

status

Property type: PortForwarderStatus

The status of the port forwarder. See PortForwarderStatus.

start() → None: Starts the HTTP or SOCKS5 proxy.

stop() → None: Stops the port forwarder and waits for it to fully terminate before returning.

bytes_uploaded

Property type: int

Returns the total number of bytes sent through this port forwarder so far.

bytes_downloaded

Property type: int

Returns the total number of bytes received through this port forwarder so far.

exception

Property type: BaseException | None

Returns the most recent exception raised within the port forwarder.

connections

Property type: Connections

All the connections made through this port forwarder so far, including closed ones, returned as a dict with a numerical index as the key.

endpoint_connections

Property type: EndpointConnectionsDict

Returns all the endpoint connections going through this proxy.

add_procedure( name: str, handler: Callable[..., Any], ) → None

Add custom “stored procedure” as FastAPI POST endpoint. (Allow end-users to extend the radkit-client API.)

When using this, ensure that the given handler has the correct type annotations according to the FastAPI documentation. These are used for deserialization/serialization of the inputs/outputs.

class radkit_client.sync.PortForwarderStatus

Bases: Enum

Enum of the possible statuses for a TCP or HTTP/SOCKS5 port forwarder.

RUNNING = 'RUNNING': The port forwarder is running.

FAILED = 'FAILED': The port forwarder failed with an exception.

STOPPED = 'STOPPED': The port forwarder is not running.

class radkit_client.sync.PortForwarderInvalidStateError

Bases: ClientError

Raised when trying to stop a port forwarder that is already stopped, or when trying to start a port forwarder that is already started.

class radkit_client.sync.Connection

Bases: SyncWrapper[AsyncConnection]

WORK IN PROGRESS

uuid: Property type: UUID

endpoint: Property type: str

status: Property type: ConnectionStatus

opened: Property type: datetime

closed: Property type: datetime | None

exception: Property type: BaseException | None

class radkit_client.sync.Connections

Bases: SyncDictWrapper[AsyncConnections, int, AsyncConnection, Connection]

WORK IN PROGRESS

class radkit_client.sync.EndpointConnections

Bases: SyncWrapper[AsyncEndpointConnections]

WORK IN PROGRESS

connections: Property type: Connections

class radkit_client.sync.EndpointConnectionsDict

Bases: SyncDictWrapper[AsyncEndpointConnectionsDict, str, AsyncEndpointConnections, EndpointConnections]

WORK IN PROGRESS

class radkit_client.sync.ConnectionStatus

Bases: Enum

Enum of the possible statuses for a port forwarding or proxy connection.

OPEN = 'OPEN': The connection is open.

CLOSED = 'CLOSED': The connection has been closed.

FAILED = 'FAILED': An error occurred on the connection.

Device flow 

class radkit_client.sync.DeviceFlow

Bases: SyncWrapper[AsyncDeviceFlow]

DeviceFlow provides a simple workflow for querying a set of devices and filtering out errors (error management). It is created by calling create_device_flow() on a DeviceDict object. DeviceFlow only supports exec() requests at this time.

Initially all the devices are part of the active_devices set. After each request, DeviceFlow looks for two types of errors on each device in the active set:

RADKit errors: timeout, bad credentials, unauthorized, …
functional errors: command not recognized, invalid argument, file not found, …

Depending on the type of error encountered, the failed devices move from active_devices to command_failed or failed_devices. Devices that did not meet any error remain on the active_devices list, ready to be queried again.

Example:

flow = service.inventory.create_device_flow()

# List all the devices ready to take commands (currently, a copy of service.inventory)
flow.active_devices

# This may take a long time and be riddled with errors of various types
versions = await flow.exec_wait("show version")

# List the devices that succeeded the "show version" above without any type of problem
flow.active_devices

# List the devices that failed to be contacted by the Service
# (authentication error, unauthorized, timeout, ...)
flow.failed_devices

# List the devices that failed to return a satisfactory response despite being reachable
# (e.g. "command not found", "No such file or directory", ...)
flow.command_failed

# This second command should execute faster and without error,
# as problematic devices were weeded out after the previous request
routes = await flow.exec_wait("show ip route")

DeviceFlow detects functional (non-RADKit) errors in exec() requests by looking for known error patterns in the command output. A number of patterns are built into RADKit Client; those include:

r".*Invalid input detected.*"
r".*% Bad IP address or host name% Unknown command or computer name, or unable to find computer address.*"
r".*No such file or directory.*"
r".*missing operand.*"
r".*can't open file.*"
r".*file not found.*"
r".*command not found.*"
r".*Invalid argument.*"
r".*Unknown Command.*"

All those regular expressions are compiled with re.compile(..., flags=re.DOTALL) so that the match works on multi-line output. If necessary, a different set of patterns may be provided while creating the DeviceFlow object, for example:

import re
my_patterns = [
    re.compile(r".*processor is too hot.*", flags=re.DOTALL),
    re.compile(r".*coffee is too cold.*", flags=re.DOTALL),
]
flow = my_devices.create_device_flow(exec_error_patterns=my_patterns)

Note

The exec_error_patterns argument must be passed to create_device_flow() as a sequence (usually a list) of compiled string patterns, as returned by re.compile().

active_devices

Property type: DeviceDict

Returns a DeviceDict containing all devices that are currently active as part of this DeviceFlow (i.e. those devices that will receive the next request).

failed_devices

Property type: DeviceDict

Returns a DeviceDict containing all devices that encountered a RADKit error (e.g. timeout, bad credentials, …) during a previous request. Those devices will no longer receive future requests performed using this DeviceFlow.

command_failed

Property type: DeviceDict

Returns a DeviceDict containing all devices that encountered a functional error (e.g. command not found, unknown path, …) during a previous request. Those devices will no longer receive future requests performed using this DeviceFlow.

flow_mode

Property type: DeviceFlowMode

Returns the mode of operation for this DeviceFlow. See DeviceFlowMode for details.

exec_wait( commands: str, synced: bool = False, timeout: int = 0, exec_error_patterns: Sequence[Pattern[str]] | None = None, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecResponse_ByDevice_ToSingle[str]
exec_wait( commands: list[str], synced: bool = False, timeout: int = 0, exec_error_patterns: Sequence[Pattern[str]] | None = None, prompt_detection_strategy: PromptDetectionStrategy | None = None, ) → ExecResponse_ByDevice_ByCommand[str]: Executes one or more commands, waits for those to complete, updates active_devices based on detected errors, and return the pruned output (i.e. without result entries for devices that failed).

class radkit_client.sync.DeviceFlowMode

Bases: Enum

Enum of the possible DeviceFlow modes of operation.

BEST_EFFORT = 'BEST_EFFORT': “Don’t care” mode (default). No exception is raised even if all devices/commands end up failing.

AT_LEAST_1 = 'AT_LEAST_1': At least 1 device/command must succeed. If all devices somehow fail, an exception is raised.

FULL_SUCCESS = 'FULL_SUCCESS': “Strict” mode. All commands must succeed on all devices. Any error causes an exception to be raised.

class radkit_client.sync.DeviceFlowFailedError

Bases: ClientError

Can be raised when DeviceFlow encounters an error, depending on the mode (see DeviceFlowMode):

in BEST_EFFORT mode, this exception is never raised;
in AT_LEAST_1 mode, this is raised when there are no more devices to work on;
in FULL_SUCCESS mode, this is raised when at least one device failed.

Requests 

Note

This section will be expanded in the future with a simplified reference covering the various request classes in RADKit Client, covering only the portions of the API which are actually relevant to the end user.

class radkit_client.sync.RequestStatus

Bases: Enum

The possible statuses for a request object.

NOT_YET_SUBMITTED = 'NOT_YET_SUBMITTED': The request was not yet sent.

SUBMITTED = 'SUBMITTED': The request is being sent to the Forwarder.

PUBLISHED = 'PUBLISHED': The request has been delivered to the Forwarder.

DELIVERED = 'DELIVERED': The request has been delivered to the Service.

SUCCESS = 'SUCCESS': The request has been processed by the Service.

PROCESSING = 'PROCESSING': The request is being processed by the Service.

SEND_TIMEOUT = 'SEND_TIMEOUT': The request could not be delivered on time.

FAILURE = 'FAILURE': The request could not be processed.

TIMEOUT = 'TIMEOUT': The request timed out during processing.

CANCELLED = 'CANCELLED': The request has been successfully cancelled.

REQUEST_HANDLER_FAILED = 'REQUEST_HANDLER_FAILED': The response to the request could not be processed.

PARTIAL_SUCCESS = 'PARTIAL_SUCCESS': A complex request encountered some failures.

class radkit_client.sync.RequestPendingError

Bases: ClientError

Raised when accessing the result of a request that is still pending.

Joining 

radkit_client.sync.join( *awaitables: Task, ) → Generator[JoinedTasks, None, None]

Join multiple awaitables into one.

This is a context manager, because this will spawn an async task underneath to monitor the progress and update the status message and counters. The context manager ensures proper cancellation of this underlying task. Use the join method from the radkit_client.sync.context as an alternative that does not require a context manager syntax.

Usage:

with join(awaitable1, awaitable2) as joined_awaitables:
    joined_awaitables.show_progress()
    # - or -
    joined_awaitables.wait()

class radkit_client.sync.JoinedTasks

Bases: SyncDictWrapper[AsyncJoinedTasks, int, AsyncTask, Task]

Utility for waiting for multiple radkit_client tasks (like requests) to complete.

show_progress( label: str = Tasks completed, ) → None: Show a progress bar displaying the progress of the wrapped awaitables.

wait( timeout: float | None = None, ) → Self

Wait for all the underlying awaitables to complete.

Parameters:: timeout – optional – timeout in seconds (default: wait forever).

status: Property type: JoinedTasksStatus

done_count

Property type: int

Counter number of items that are completed.

success_count

Property type: int

Counter number of items that are completed successfully.

failure_count

Property type: int

Counter number of items that are completed successfully.

as_completed() → Generator[Task, None]

Iterate over requests when they are completed.

Example usage:

async for request in joined_request.as_completed():
    print(request)

class radkit_client.sync.JoinedTasksStatus

Bases: Enum

IN_PROGRESS = 'IN_PROGRESS': At least one task is still in progress.

SUCCESS = 'SUCCESS': All joined tasks completed successfully.

PARTIAL_SUCCESS = 'PARTIAL_SUCCESS': All joined tasks completed. Some successfully, some with failure.

FAILED = 'FAILED': All joined tasks completed with failure.

radkit_client.sync.joining.join( *awaitables: Task, ) → Generator[JoinedTasks, None, None]

Join multiple awaitables into one.

This is a context manager, because this will spawn an async task underneath to monitor the progress and update the status message and counters. The context manager ensures proper cancellation of this underlying task. Use the join method from the radkit_client.sync.context as an alternative that does not require a context manager syntax.

Usage:

with join(awaitable1, awaitable2) as joined_awaitables:
    joined_awaitables.show_progress()
    # - or -
    joined_awaitables.wait()

Pipes 

The RADKit Client REPL (text-based Python interface) allows adding UNIX-style pipes after commands like:

some_expression | grep("Keyword")
some_expression | regrep(r"regular.[Ee]xpression")
some_expression | sort | tail(40)

These pipes are implemented as Python objects that are exposed in the REPL scope and implement the __ror__ operator. They can also be imported from radkit_client.sync.pipes into stand-alone scripts, for example:

from radkit_client.sync.pipes import grep, regrep

Note

Changed in 1.4.4: the pipe functions can no longer be imported from radkit_client.sync. You must import those explicitly from radkit_client.sync.pipes.

Note that if you do from radkit_client.sync.pipes import * (not recommended), you will also replace your built-in print function with the one from RADKit Client.

radkit_client.sync.pipes.grep

Performs a string search in the input.

Syntax: <input> | grep(...)

Parameters:

keyword – the string to look for anywhere in the input.
ignore_case – whether to perform a case-insensitive search, False by default.
inverse – whether to exclude lines where a match is found, False by default.

radkit_client.sync.pipes.regrep

Performs a regular expression search in the input.

The search is performed using re.search. Flags such as re.MULTILINE or re.IGNORECASE can be added through the flags argument.

Syntax: <input> | regrep(...)

Parameters:

regex – the regular expression to look for in the input.
flags – one or more re.RegexFlag values OR’ed together.
inverse – whether to exclude lines where a match is found, False by default.

radkit_client.sync.pipes.head = Head(count=10)

Returns the first count lines of the input.

Syntax: <input> | head or <input> | head(...)

Parameters:: count – the number of lines to return, 10 by default.

radkit_client.sync.pipes.tail = Tail(count=10)

Returns the last count lines of the input.

Syntax: <input> | tail or <input> | tail(...)

Parameters:: count – the number of lines to return, 10 by default.

radkit_client.sync.pipes.sort = Sort(reverse=False)

Sorts the lines in the input. The sort is performed using sorted() which returns items in lexicographical order.

Syntax: <input> | sort or <input> | sort(...)

Parameters:: reverse – whether to reverse the sort order (last items first), False by default.

radkit_client.sync.pipes.print = Print()

Prints the input on the terminal and paginates if necessary. Can be used as a pipe, as well as a print function.

Syntax: <input> | print

radkit_client.sync.pipes.to_json = ToJSON(dumps_args={'indent': 4})

Serializes a Python object into its JSON string representation. Keyword arguments are passed directly to json.dumps(). The indent=4 argument is passed by default for pretty-printing, but can be overridden by passing another value explicitly.

Syntax: <input> | to_json or <input> | to_json(...)

radkit_client.sync.pipes.from_json = FromJSON(loads_args={})

Deserializes a Python object from its JSON string representation. Keyword arguments are passed directly to json.loads().

Syntax: <input> | from_json or <input> | from_json(...)

radkit_client.sync.pipes.to_dict = ToDict(recursive=True)

Converts a RADKit Client object into a standard Python dictionary. The effect of this operation depends on the type of object provided as input. If recursive is set to False, the resulting object will be a pure Python dictionary, but its values may still be RADKit Client objects.

Syntax: <input> | to_dict or <input> | to_dict(...)

Parameters:: recursive – whether to perform a recursive conversion, True by default.

radkit_client.sync.pipes.read_from_file( file_path: Path | str, *, text: Literal[False] = False, ) → bytes | FileOperationFailure

radkit_client.sync.pipes.read_from_file( file_path: Path | str, *, text: Literal[True], encoding: str = DEFAULT_FILE_ENCODING, ) → str | FileOperationFailure

Reads the content of a file and returns it as a string or bytes.

Parameters:

file_path – the path to the file to read.
text – whether to read the file as text (True) or as bytes (False), False by default.

radkit_client.sync.pipes.write_to_file

Writes the input (which must be a str or bytes) to a local file.

Syntax: <input> | write_to_file(...)

Parameters:: file_path – the path to a local file. If the file exists, it will be overwritten.

radkit_client.sync.pipes.append_to_file

Appends the input (which must be a str) to a local file.

Syntax: <input> | append_to_file(...)

Parameters:: file_path – the path to a local file. If the file does not exist, it will be created.

radkit_client.sync.pipes.encrypt

Encrypts a string or byte string using AES256-GCM and returns a modern PKCS#7 enveloped called CMS (RFC 5652).

By default, the key is salted and derived using PBKDF2 with SHA-512.

If salt_key is False, the key must be exactly 32 bytes long.

encrypt and decrypt can be used as a pipe or as a function:: “abc” | encrypt(“key”) | decrypt(“key”)
or: e = encrypt(“key”) cypher = e(“abc”) d = decrypt(“key”) d(cypher)

Parameters:

key – the key to use for encryption.
salt_key – whether to salt the key and derive it using PBKDF2 with SHA-512, True by default.

radkit_client.sync.pipes.decrypt

radkit_client.sync.pipes.encrypt_openssl

Encrypts a string or byte string using AES-256-BCS. The encryption key is derived from the key using PBKDF2 with SHA-512.

The output is compatible with OpenSSL.

Syntax: <input> | encrypt_openssl(...)

Example 1:

Encrypt data with the following Python code: > “Hello World!” | encrypt_openssl(“MyPassword”) | write_to_file(“cipher.bin”)

The output can be decrypted with (shell): cat cipher.bin | openssl enc -aes-256-cbc -d -salt -saltlen 16 -pbkdf2 -iter 100000 -md sha512 -k “MyPassword”

Example 2:

Encrypt data with the following Python code: > “Hello World!” | encrypt_openssl(“MyPassword”) | to_base64 | write_to_file(“cipher.b64”)

The output can be decrypted with (shell): cat cipher.b64 | openssl enc -aes-256-cbc -d -salt -saltlen 16 -pbkdf2 -iter 100000 -md sha512 -k “MyPassword” -a -A

Parameters:

key – the key to use for encryption.
salt_size – the size of the salt to use for key derivation, 16 by default.
iterations – the number of iterations to use for key derivation, 100000 by default.

radkit_client.sync.pipes.decrypt_openssl

Decrypts a byte string using AES-256-BCS. The decryption key is derived from the key using PBKDF2 with SHA-512.

The input is compatible with OpenSSL.

Syntax: <input> | decrypt_openssl(...)

Example 1:

Encrypt data with the following openssl command (shell): echo -n ‘Hello World!’ | openssl enc -aes-256-cbc -e -salt -saltlen 16 -pbkdf2 -iter 100000 -md sha512 -k “MyPassword” > cipher.bin

The output can be decrypted with: > read_from_file(“cipher.bin”) | decrypt_openssl(“MyPassword”)

Example 2:

Encrypt data with the following openssl command: echo -n ‘Hello World!’ | openssl enc -aes-256-cbc -e -salt -saltlen 16 -pbkdf2 -iter 100000 -md sha512 -k “MyPassword” -a -A > cipher.b64

The output can be decrypted with (shell): > read_from_file(“cipher.b64”) | from_base64 | decrypt_openssl(“MyPassword”)

Parameters:

key – the key to use for decryption.
salt_size – the size of the salt to use for key derivation, 16 by default.
iterations – the number of iterations to use for key derivation, 100000 by default.

radkit_client.sync.pipes.to_base64 = ToBase64()

Encodes the input to base64. The output is a string

Syntax: <input> | to_base64

radkit_client.sync.pipes.from_base64 = FromBase64()

Decodes the input from a base64 string or bytestring.

Syntax: <input> | from_base64