Skip to main content

br_rapp_sdk.oam_services.terminal

OAMTerminalService Objects

class OAMTerminalService()

This class provides methods to interact with the OAM Services Terminal-related API in BubbleRAN environment.

It allows you to list, get, apply, and delete terminals in the Kubernetes cluster.

Attributes:

  • kubeconfig_path str - Path to the kubeconfig file for Kubernetes API access.
  • namespace str - Kubernetes namespace where the Terminal CRs are located.

Examples:

from br_rapp_sdk.oam_services.terminal import OAMTerminalService

terminal_service = OAMTerminalService()
terminal_spec = TerminalSpec(
# Fill in the required fields for the terminal specification
)
result = terminal_service.apply_terminal("my-new-terminal", terminal_spec)
if result.status == 'success':
print("Terminal applied successfully: ", result.data.get('terminal_id'))
else:
print("Failed to apply terminal: ", result.error)

__init__

def __init__(kubeconfig_path: str = None, namespace: str = "trirematics")

Initialize the Terminal client by loading the Kubernetes configuration and setting up defaults.

Arguments:

  • kubeconfig_path Optional[str] - Path to the kubeconfig file (default: None - use the default kubeconfig).
  • namespace str - Kubernetes namespace for the Terminal CRs (default: "trirematics").

Raises:

  • RuntimeError - If the kubeconfig cannot be loaded.

list_terminals

def list_terminals() -> KubectlOperationResult

Get the list of terminals.

Returns:

  • KubectlOperationResult - An object representing the result of the operation, containing a list of TermId and TerminalSpec tuples if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.terminal import OAMTerminalService

terminal_service = OAMTerminalService()
result = terminal_service.list_terminals()
# Check if the operation was successful
if result.status == 'success':
for term_id, spec in result.data.get('items'):
# Use term_id and spec as needed
print(f"Terminal ID: {term_id}, Spec: {spec}")
else:
print("Failed to retrieve terminals: ", result.error)

get_terminal

def get_terminal(terminal_id: TermId) -> KubectlOperationResult

Get a terminal by its ID.

Arguments:

  • terminal_id TermId - The ID of the terminal to retrieve.

Returns:

  • KubectlOperationResult - An object representing the result of the operation, containing the TerminalSpec if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.terminal import OAMTerminalService

terminal_service = OAMTerminalService()
term_id = TermId("sample-terminal")
result = terminal_service.get_terminal(term_id)
# Check if the operation was successful
if result.status == 'success':
# Use the terminal_spec as needed
terminal_spec = result.data.get('item')
else:
print("Failed to retrieve terminal: ", result.error)

apply_terminal

def apply_terminal(terminal_name: str,
terminal_spec: TerminalSpec) -> KubectlOperationResult

Apply the terminal to the OAM Services API.

Arguments:

  • terminal_name str - The name of the terminal.
  • terminal_spec TerminalSpec - The terminal specification to apply.

Returns:

  • KubectlOperationResult - The result of the operation, containing the terminal ID if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.terminal import OAMTerminalService

terminal_service = OAMTerminalService()
terminal_spec = TerminalSpec(
# Fill in the required fields for the terminal spec
)
result = terminal_service.apply_terminal("my-new-terminal", terminal_spec)
if result.status == 'success':
# Use the terminal_id as needed
terminal_id = result.data.get('terminal_id')
print("Terminal applied successfully:", terminal_id)
else:
print("Failed to apply terminal:", result.error)

delete_terminal

def delete_terminal(terminal_id: TermId) -> KubectlOperationResult

Delete the terminal from the OAM Services API.

Arguments:

  • terminal_id TermId - The ID of the terminal to delete.

Returns:

  • KubectlOperationResult - The result of the delete operation, containing an empty data dictionary if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.terminal import OAMTerminalService

terminal_service = OAMTerminalService()
terminal_id = TermId("sample-terminal")
result = terminal_service.delete_terminal(terminal_id)
if result.status == 'success':
print("Terminal deleted successfully.")
else:
print("Failed to delete terminal:", result.error)

br_rapp_sdk.oam_services.network

OAMNetworkService Objects

class OAMNetworkService()

This class provides methods to interact with the OAM Services Network-related API in BubbleRAN environment.

It allows you to manage networks, including creating, retrieving, updating, and deleting network configurations.

Attributes:

  • kubeconfig_path Optional[str] - Path to the kubeconfig file for Kubernetes API access.
  • namespace str - The Kubernetes namespace where the network CRs are located (default: "trirematics").

Examples:

from br_rapp_sdk.oam_services.network import OAMNetworkService

network_service = OAMNetworkService()
network_spec = NetworkSpec(
# Fill in the required fields for the network specification
)
result = network_service.apply_network("my-new-network", network_spec)
if result.status == 'success':
print("Network applied successfully: ", result.data.get('network_id'))
else:
print("Failed to apply network: ", result.error)

__init__

def __init__(kubeconfig_path: Optional[str] = None,
namespace: str = "trirematics")

Initialize the Network client by loading the Kubernetes configuration and setting up defaults.

Arguments:

  • kubeconfig_path Optional[str] - Path to the kubeconfig file (default: None - use the default kubeconfig).
  • namespace str - Kubernetes namespace for the Network CRs (default: "trirematics").

Raises:

  • RuntimeError - If the kubeconfig cannot be loaded.

list_networks

def list_networks(
network_id: Optional[NetworkId] = None,
part: Optional[NetworkPart] = None) -> KubectlOperationResult

Get the list of networks.

Arguments:

  • network_id Optional[NetworkId] - The ID of the network to filter by (default: None - list all networks). Useful in combination with the part parameter to list specific parts of a network.
  • part Optional[NetworkPart] - The part of the network to list ('access', 'core', 'edge', or None for full spec) (default: None).

Returns:

  • KubectlOperationResult - An object representing the result of the operation, containing a list of NetworkId and NetworkSpec tuples if successful, or an error message if not.

Examples:

Listing all networks:

from br_rapp_sdk.oam_services.network import OAMNetworkService

network_service = OAMNetworkService()
result = network_service.list_networks()
# Check if the operation was successful
if result.status == 'success':
for net_id, spec in result.data.get('items'):
# Use net_id and spec as needed
print(f"Network ID: {net_id}, Spec: {spec}")
else:
print("Failed to retrieve networks: ", result.error)

Listing the core part of the mynet network:

from br_rapp_sdk.oam_services.network import OAMNetworkService
from br_rapp_sdk.oam_services.network_types import NetworkId

network_service = OAMNetworkService()
network_id = NetworkId("mynet")
result = network_service.list_networks(network_id=network_id, part='core')
# Check if the operation was successful
if result.status == 'success':
for core_id, core_spec in result.data.get('items'):
# Use core_id and core_spec as needed
print(f"Core Network ID: {core_id}, Spec: {core_spec}")
else:
print("Failed to retrieve network core: ", result.error)

get_network

def get_network(network_id: NetworkId) -> KubectlOperationResult

Get a specific network by its ID.

Arguments:

  • network_id NetworkId - The ID of the network to retrieve.

Returns:

  • KubectlOperationResult - An object representing the result of the operation, containing the NetworkSpec if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.network import OAMNetworkService

network_service = OAMNetworkService()
network_id = NetworkId("sample-network")
result = network_service.get_network(network_id)
# Check if the operation was successful
if result.status == 'success':
# Use the network_spec as needed
network_spec = result.data.get('item')
else:
print("Failed to retrieve network: ", result.error)

apply_network

def apply_network(network_name: str,
network_spec: NetworkSpec) -> KubectlOperationResult

Apply the network to OAM services API.

Arguments:

  • network_name str - The name of the network.
  • network_spec NetworkSpec - The network specification to apply.

Returns:

  • KubectlOperationResult - The result of the apply operation, containing the NetworkId if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.network import OAMNetworkService

network_service = OAMNetworkService()
network_spec = NetworkSpec(
slices=[...], # Fill in the required fields for the network spec
access=[...], # AccessNetworkSpec instances
core=[...], # CoreNetworkSpec instances
edge=[...] # EdgeNetworkSpec instances
)
result = network_service.apply_network("my-new-network", network_spec)
if result.status == 'success':
# Use the network_id as needed
network_id = result.data.get('network_id')
print("Network applied successfully:", network_id)
else:
print("Failed to apply network:", result.error)

delete_network

def delete_network(network_id: NetworkId) -> KubectlOperationResult

Delete the network from the OAM Services API.

Arguments:

  • network_id NetworkId - The ID of the network to delete.

Returns:

  • KubectlOperationResult - The result of the delete operation, containing an empty data dictionary if successful, or an error message if not.

Examples:

from br_rapp_sdk.oam_services.network import OAMNetworkService

network_service = OAMNetworkService()
network_id = NetworkId("sample-network")
result = network_service.delete_network(network_id)
if result.status == 'success':
print("Network deleted successfully.")
else:
print("Failed to delete network:", result.error)