acmecse.runtime.Management.ManagementSupport

class documentation

class ManagementSupport: (source)

Management support class. Provides functions for accessing and managing CSE internal functions. Also provides functions for retrieving and formatting CSE status, statistics, and runtime information.

Method	`addBasicCredentials`	Add new HTTP or WebSocket Basic credentials to the CSE.
Method	`addTokenCredentials`	Add new HTTP or WebSocket Token credentials to the CSE.
Method	`deleteBasicCredentials`	Delete HTTP or WebSocket Basic credentials from the CSE.
Method	`deleteTokenCredentials`	Delete HTTP or WebSocket Token credentials from the CSE.
Method	`doExportInstances`	Export instances of a container resource to a CSV file in the tmp directory, or return as a string.
Method	`doExportResource`	Export a resource and its children to the tmp directory as a shell script with curl commands.
Method	`getAttributePolicies`	Get the current attribute type policies of the CSE.
Method	`getBasicCredentials`	Get the current HTTP or WS credentials of the CSE.
Method	`getConfig`	Get the current configuration of the CSE as a JSON string.
Method	`getCSEStatus`	Get the current status of the CSE.
Method	`getCSEStatusAsDict`	Get the status, statistics, and runtime information of the CSE.
Method	`getFlexContainerPolicies`	Get the current flexcontainer type policies of the CSE.
Method	`getInterceptors`	Get the current registered interceptors of the CSE.
Method	`getLogGenerator`	Get a generator that yields log entries as strings.
Method	`getLoglevel`	Get the current log level of the CSE.
Method	`getPlugins`	Get the current loaded plugins of the CSE.
Method	`getRegistrations`	Get the current registration status of the CSE as a JSON string.
Method	`getRegistrationsRich`	Create and return an overview about the registrar, registrees, and descendant CSE's.
Method	`getRegistrationStatus`	Return the registration status of the CSE, and registrations of CSEs and AEs.
Method	`getRequests`	Get the current requests of the CSE as a generator of JSON strings.
Method	`getRequestsRich`	Generate a Rich table with all requests and a PlantUML sequence diagram.
Method	`getResourceTreeRich`	This function will generate a Rich tree structure of a CSE's resource structure.
Method	`getResourceTreeText`	This function will generate a Text tree of a CSE's resource structure.
Method	`getResourceTypePolicies`	Get the current resource type policies of the CSE.
Method	`getServices`	Get the current registered services and their endpoints of the CSE.
Method	`getStatusRich`	Generate an overview about various resources, event counts, and more.
Method	`getStructurePuml`	The CSE, Type, http, port
Method	`getTokenCredentials`	Get the current HTTP or WebSocket Bearer credentials of the CSE.
Method	`refreshRegistrations`	Force the CSE to immediately check the registrations with the remote registrar(s).
Method	`reloadCredentials`	Reload the credentials from the credential storage.
Method	`resetCSE`	Reset the CSE to its initial state.
Method	`restartCSE`	Restart the CSE.
Method	`setLogLevel`	Set the log level of the CSE.
Method	`setRequestRecording`	Enable or disable request recording.
Method	`shutdownCSE`	Shutdown the CSE.
Method	`updateBasicCredentials`	Update existing HTTP or WebSocket Basic credentials in the CSE.
Method	`updateTokenCredentials`	Update existing HTTP or WebSocket Token credentials in the CSE.
Class Variable	`credentialsManager`	Injected CredentialsManager instance.
Class Variable	`cseForceShutdown`	Injected function to shutdown the CSE.
Class Variable	`cseReset`	Injected function to reset the CSE.
Class Variable	`dispatcher`	Injected Dispatcher instance.
Class Variable	`requestManager`	Injected RequestManager instance.
Class Variable	`securityManager`	Injected SecurityManager instance.
Class Variable	`storage`	Injected Storage instance.
Class Variable	`validator`	Injected Validator instance.
Method	`_checkBindingPlugins`	Check if the required binding plugins are enabled.
Method	`_handleAttributePolicy`	Handle an attribute policy and return a JSON representation of it.

def addBasicCredentials(self, jsn: JSON, ty: BindingType) -> str: (source) ¶

Add new HTTP or WebSocket Basic credentials to the CSE.

Parameters
jsn:`JSON`	A JSON object containing the new credentials in the format { "username": "password" }.
ty:`BindingType`	The type of credentials to add.

Returns
`str`	A JSON object with a message indicating success or failure.

def addTokenCredentials(self, jsn: JSON, ty: BindingType) -> str: (source) ¶

Add new HTTP or WebSocket Token credentials to the CSE.

Parameters
jsn:`JSON`	A JSON object containing the new credentials in the format { "username": "password" }.
ty:`BindingType`	The type of credentials to add.

Returns
`str`	A JSON object with a message indicating success or failure.

def deleteBasicCredentials(self, username: str, ty: BindingType) -> str: (source) ¶

Delete HTTP or WebSocket Basic credentials from the CSE.

Parameters
username:`str`	The username of the credentials to delete.
ty:`BindingType`	The type of credentials to delete.

Returns
`str`	A JSON object with a message indicating success or failure.

def deleteTokenCredentials(self, token: str, ty: BindingType) -> str: (source) ¶

Delete HTTP or WebSocket Token credentials from the CSE.

Parameters
token:`str`	The token of the credentials to delete.
ty:`BindingType`	The type of credentials to delete.

Returns
`str`	A JSON object with a message indicating success or failure.

def doExportInstances(self, ri: str, asString: bool | None = False) -> tuple[int, str]: (source) ¶

Export instances of a container resource to a CSV file in the tmp directory, or return as a string.

Parameters
ri:`str`	Resource ID of the container resource.
asString:`bool \| None`	Return the CSV string instead of writing to a file.

Returns
`tuple[int, str]`	Tuple with the number of instances exported, and the filename of the exported file or the CSV string.

def doExportResource(self, ri: str, withChildResources: bool | None = False) -> tuple[int, str]: (source) ¶

Export a resource and its children to the tmp directory as a shell script with curl commands.

Parameters
ri:`str`	Resource ID of the resource to export.
withChildResources:`bool \| None`	If True, also export child resources.

Returns
`tuple[int, str]`	Tuple with the number of resources exported, and the filename of the exported file.

def getAttributePolicies(self, filter: str | None = None) -> str: (source) ¶

Get the current attribute type policies of the CSE.

Parameters
filter:`str \| None`	An optional filter for the management command.

Returns
`str`	The attribute type policies of the CSE in JSON format.

def getBasicCredentials(self, ty: BindingType) -> str: (source) ¶

Get the current HTTP or WS credentials of the CSE.

Parameters
ty:`BindingType`	The type of credentials to retrieve.

Returns
`str`	The HTTP or WScredentials of the CSE in JSON format, or an error message if the HTTP server plugin is not enabled.

def getConfig(self) -> str: (source) ¶

Get the current configuration of the CSE as a JSON string.

Returns
`str`	The configuration of the CSE in JSON format.

def getCSEStatus(self) -> str: (source) ¶

Get the current status of the CSE.

Returns
`str`	The status of the CSE in JSON format.

def getCSEStatusAsDict(self) -> JSON: (source) ¶

Get the status, statistics, and runtime information of the CSE.

Returns
`JSON`	Status as JSON object.

def getFlexContainerPolicies(self, filter: str | None = None) -> str: (source) ¶

Get the current flexcontainer type policies of the CSE.

Parameters
filter:`str \| None`	An optional filter for the management command.

Returns
`str`	The flexcontainer type policies of the CSE in JSON format.

def getInterceptors(self) -> str: (source) ¶

Get the current registered interceptors of the CSE.

Returns
`str`	The registered interceptors of the CSE in JSON format.

def getLogGenerator(self) -> Generator[str, None, None]: (source) ¶

Get a generator that yields log entries as strings.

Returns
`Generator[str, None, None]`	A generator that yields log entries as strings.

def getLoglevel(self) -> str: (source) ¶

Get the current log level of the CSE.

Returns
`str`	The current log level of the CSE.

def getPlugins(self) -> str: (source) ¶

Get the current loaded plugins of the CSE.

Returns
`str`	The loaded plugins of the CSE in JSON format.

def getRegistrations(self) -> str: (source) ¶

Get the current registration status of the CSE as a JSON string.

Returns
`str`	The registration status of the CSE in JSON format.

def getRegistrationsRich(self, style: Style | None = Style(), textStyle: Style | None = None) -> Table: (source) ¶

Create and return an overview about the registrar, registrees, and descendant CSE's.

Parameters
style:`Style \| None`	Style for the general output.
textStyle:`Style \| None`	Style for the text.

Returns
`Table`	Rich formatted string.

def getRegistrationStatus(self) -> JSON: (source) ¶

Return the registration status of the CSE, and registrations of CSEs and AEs.

Returns
`JSON`	A JSON object with the registration status of the CSE, and registrations of CSEs and AEs.

def getRequests(self) -> Generator[str, None, None]: (source) ¶

Get the current requests of the CSE as a generator of JSON strings.

Returns
`Generator[str, None, None]`	A generator that yields JSON strings of the request.

def getRequestsRich(self, id: str | None = None) -> tuple[Table, str]: (source) ¶

Generate a Rich table with all requests and a PlantUML sequence diagram.

Parameters
id:`str \| None`	If set, then only the requests for this resource ID are returned.

Returns
`tuple[Table, str]`	A tuple with a Rich Table object and a PlantUML sequence diagram string.

def getResourceTreeRich(self, maxLevel: int = 0, parent: str | None = None, style: Style | None = Style(), withProgress: bool | None = True, treeMode: TreeMode | None = TreeMode.NORMAL) -> Tree: (source) ¶

This function will generate a Rich tree structure of a CSE's resource structure.

Parameters
maxLevel:`int`	The maximum level for the result tree.
parent:`str \| None`	The resource ID from where to start the tree. The default is the CSEBase.
style:`Style \| None`	The Rich Style to use.
withProgress:`bool \| None`	Display a progress indicator while gathering the tree.
treeMode:`TreeMode \| None`	Undocumented

Returns
`Tree`	Return a Rich Tree object.

def getResourceTreeText(self, maxLevel: int = 0) -> str: (source) ¶

This function will generate a Text tree of a CSE's resource structure.

Todo

Support the maxLevel parameter.

Parameters
maxLevel:`int`	Maximum tree level to render. Currently not supported.

Returns
`str`	Pure text rendering of the resource tree.

def getResourceTypePolicies(self, filter: str | None = None) -> str: (source) ¶

Get the current resource type policies of the CSE.

Parameters
filter:`str \| None`	An optional filter for the management command.

Returns
`str`	The resource type policies of the CSE in JSON format.

def getServices(self) -> str: (source) ¶

Get the current registered services and their endpoints of the CSE.

Returns
`str`	The registered services and their endpoints of the CSE in JSON format.

def getStatusRich(self, style: Style | None = Style(), withProgress: bool | None = True, textStyle: Style | None = None) -> Table: (source) ¶

Generate an overview about various resources, event counts, and more.

Parameters
style:`Style \| None`	Rich style.
withProgress:`bool \| None`	Display with progress indicator.
textStyle:`Style \| None`	Rich text style. If this is not set then the style is used for the text as well.

Returns
`Table`	Rich Table object.

def getStructurePuml(self, maxLevel: int | None = 0) -> str: (source) ¶

This function will generate a PlanUML graph of a CSE's structure, including:

The CSE, Type, http, port
The CSE's resource tree
The Registrar CSE (if any)
A list of descendant CSE's (if any)

This function calls itself recursively to generate the tree structure.

Parameters
maxLevel:`int \| None`	The maximum level of the tree to print. 0 means all levels.

Returns
`str`	The PlanUML graph as a string.

def getTokenCredentials(self, ty: BindingType) -> str: (source) ¶

Get the current HTTP or WebSocket Bearer credentials of the CSE.

Parameters
ty:`BindingType`	The type of credentials to retrieve.

Returns
`str`	The HTTP or WebSocket Bearer credentials of the CSE in JSON format, or an error message if the corresponding server plugin is not enabled.

def refreshRegistrations(self) -> str: (source) ¶

Force the CSE to immediately check the registrations with the remote registrar(s).

This is useful for testing purposes, e.g. when the CSE is started and the registration is not done automatically.

Returns
`str`	A message indicating the result of the registration.

def reloadCredentials(self) -> str: (source) ¶

Reload the credentials from the credential storage.

def resetCSE(self): (source) ¶

Reset the CSE to its initial state.

def restartCSE(self): (source) ¶

Restart the CSE.

This is done by setting the CSE status to SHUTTINGDOWNRESTART and calling the forceShutdown method.

def setLogLevel(self, level: str) -> str: (source) ¶

Set the log level of the CSE.

Parameters
level:`str`	The log level to set. Should be one of the Logging levels.

Returns
`str`	A response indicating the result

def setRequestRecording(self, param: str) -> str: (source) ¶

Enable or disable request recording.

Parameters
param:`str`	The parameter to set. Should be 'enable' or 'disable'.

Returns
`str`	The new status of the request recording.

def shutdownCSE(self): (source) ¶

Shutdown the CSE.

def updateBasicCredentials(self, jsn: JSON, ty: BindingType) -> str: (source) ¶

Update existing HTTP or WebSocket Basic credentials in the CSE.

Parameters
jsn:`JSON`	A JSON object containing the updated credentials in the format { "username": "password" }.
ty:`BindingType`	The type of credentials to update.

Returns
`str`	A JSON object with a message indicating success or failure.

def updateTokenCredentials(self, jsn: JSON, ty: BindingType) -> str: (source) ¶

Update existing HTTP or WebSocket Token credentials in the CSE.

Parameters
jsn:`JSON`	A JSON object containing the updated credentials in the format { "token": "<token>" }.
ty:`BindingType`	The type of credentials to update.

Returns
`str`	A JSON object with a message indicating success or failure.

credentialsManager: CredentialsManager = (source) ¶

Injected CredentialsManager instance.

cseForceShutdown: Callable[[], None] = (source) ¶

Injected function to shutdown the CSE.

cseReset: Callable[[], None] = (source) ¶

Injected function to reset the CSE.

dispatcher: Dispatcher = (source) ¶

Injected Dispatcher instance.

requestManager: RequestManager = (source) ¶

Injected RequestManager instance.

securityManager: SecurityManager = (source) ¶

Injected SecurityManager instance.

storage: Storage = (source) ¶

Injected Storage instance.

validator: Validator = (source) ¶

Injected Validator instance.

def _checkBindingPlugins(self, ty: BindingType) -> str | None: (source) ¶

Check if the required binding plugins are enabled.

Parameters
ty:`BindingType`	The type of binding to check.

Returns
`str \| None`	None if the required binding plugins are enabled, or an error message if not.

def _handleAttributePolicy(self, rt: ResourceTypes | str | None, policy: AttributePolicy) -> JSON: (source) ¶

Handle an attribute policy and return a JSON representation of it.

Parameters
rt:`ResourceTypes \| str \| None`	Undocumented
policy:`AttributePolicy`	The attribute policy to handle.

Returns
`JSON`	A JSON representation of the attribute policy.