acme.runtime.Storage.Storage

class documentation

class Storage:

This class implements the entry points to the CSE's underlying database functions.

Method	`addBatchNotification`	Add a batch notification to the DB.
Method	`addOriginator`	Add an originator to the database.
Method	`addRequest`	Add a request to the requests database.
Method	`backupDB`	Creating a backup from the DB to a sub directory.
Method	`countBatchNotifications`	Count the number of batch notifications for a target resource and a notification URI.
Method	`countDirectChildResources`	Count the number of direct child resources.
Method	`countResources`	Count the overall number of CSE resources.
Method	`createResource`	Create a new resource in the database.
Method	`deleteRequests`	Delete all requests from the database.
Method	`deleteResource`	Delete a resource from the database.
Method	`directChildResources`	Return a list of direct child resources, or an empty list
Method	`directChildResourcesRI`	Return a list of direct child resource IDs, or an empty list
Method	`getActionRepr`	Retrieve an action representation from the DB.
Method	`getAllActionReprs`	Retrieve all action representations from the DB.
Method	`getBatchNotifications`	Retrieve the batch notifications for a target resource and a notification URI.
Method	`getOriginator`	Get an originator and its information from the database.
Method	`getRequests`	Get requests for a resource ID, or all requests.
Method	`getSchedules`	Retrieve the schedules data from the DB.
Method	`getStatistics`	Retrieve the statistics data from the DB.
Method	`getSubscription`	Retrieve a subscription representation (not a oneM2M `Resource` object) from the DB.
Method	`getSubscriptionsForParent`	Retrieve all subscriptions representations (not oneM2M `Resource` objects) for a parent resource.
Method	`hasResource`	Check whether a resource with either the ri or the srn already exists.
Method	`identifier`	Search for the resource identifer mapping with the given unstructured resource ID.
Method	`initialize`	Initialization of the storage manager.
Method	`purge`	Reset and clear the databases.
Method	`purgeStatistics`	Purge the statistics DB.
Method	`removeActionRepr`	Remove an action representation from the DB.
Method	`removeBatchNotifications`	Remove the batch notifications for a target resource and a notification URI.
Method	`removeOriginator`	Remove an originator from the database.
Method	`removeSchedule`	Remove a schedule from the DB.
Method	`removeSubscription`	Remove a subscription from the DB.
Method	`retrieveResource`	Return a resource via different addressing methods.
Method	`retrieveResourceRaw`	Retrieve a resource as a raw dictionary.
Method	`retrieveResourcesByType`	Return all resources of a certain type.
Method	`searchActionReprsForSubject`	Search for action representation for a subject resource.
Method	`searchByFilter`	Return a list of resources that match the given filter, or an empty list.
Method	`searchByFragment`	Search and return all resources that match the given fragment dictionary/document.
Method	`searchScheduleForTarget`	Search for schedules for a target resource.
Method	`shutdown`	Shutdown the storage manager.
Method	`structuredIdentifier`	Search for the resource identifer mapping with the given structured resource ID.
Method	`updateActionRepr`	Update an action representation (not an actual action) in the DB.
Method	`updateResource`	Update a resource in the database.
Method	`updateStatistics`	Update the statistics DB with new data.
Method	`upsertAction`	Update or add an action as an action representation in the DB.
Method	`upsertSchedule`	Add or update a schedule in the DB.
Method	`upsertSubscription`	Add or update a subscription to the DB.
Class Variable	`__slots__`	Define slots for instance variables.
Class Variable	`factory`	Injected Factory instance.
Class Variable	`postgreSQLBinding`	Injected PostgreSQLBinding instance.
Class Variable	`tinyDBBinding`	Injected TinyDBBinding instance.
Instance Variable	`db`	The database object.
Method	`_validateDB`	Trying to validate the database files.

def addBatchNotification(self, ri: str, nu: str, request: JSON) -> bool: ¶

Add a batch notification to the DB.

Parameters
ri:`str`	The resource ID of the target resource.
nu:`str`	The notification URI.
request:`JSON`	The request to store.

Returns
`bool`	Boolean value to indicate success or failure.

def addOriginator(self, originator: str, type: OriginatorType) -> bool: ¶

Add an originator to the database.

Parameters
originator:`str`	The originator to add.
type:`OriginatorType`	The type of the originator.

Returns
`bool`	Boolean value to indicate success or failure.

def addRequest(self, op: Operation, ri: str, srn: str, originator: str, outgoing: bool, ot: str, request: JSON, response: JSON, additionalCB: Callable | None = None) -> bool: ¶

Add a request to the requests database.

Parameters
op:`Operation`	Operation.
ri:`str`	Resource ID of a request's target resource.
srn:`str`	Structured resource ID of a request's target resource.
originator:`str`	Request originator.
outgoing:`bool`	If true, then this is a request sent by the CSE.
ot:`str`	Request creation time.
request:`JSON`	The request to store.
response:`JSON`	The response to store.
additionalCB:`Callable \| None`	Undocumented

Returns
`bool`	Boolean value to indicate success or failure.

def backupDB(self) -> bool: ¶

Creating a backup from the DB to a sub directory.

Returns
`bool`	Boolean indicating the success of the backup operation.

def countBatchNotifications(self, ri: str, nu: str) -> int: ¶

Count the number of batch notifications for a target resource and a notification URI.

Parameters
ri:`str`	The resource ID of the target resource.
nu:`str`	The notification URI.

Returns
`int`	The number of matching batch notifications.

def countDirectChildResources(self, pi: str, ty: ResourceTypes | None = None) -> int: ¶

Count the number of direct child resources.

Parameters
pi:`str`	The parent resource's Resource ID.
ty:`ResourceTypes \| None`	Optional resource type to filter the result.

Returns
`int`	The number of child resources.

def countResources(self) -> int: ¶

Count the overall number of CSE resources.

Returns
`int`	The number of CSE resources.

def createResource(self, resource: Resource, overwrite: bool | None = True): ¶

Create a new resource in the database.

Parameters
resource:`Resource`	The resource to store in the database.
overwrite:`bool \| None`	Indicator whether an existing resource shall be overwritten.

Raises
`CONFLICT`	In case the resource already exists and overwrite is "False".

def deleteRequests(self, ri: str | None = None): ¶

Delete all requests from the database.

Parameters
ri:`str \| None`	Optional resouce ID. Only requests for this resource ID will be deleted.

def deleteResource(self, resource: Resource): ¶

Delete a resource from the database.

Parameters
resource:`Resource`	Resource to delete.

Raises
`NOT_FOUND`	In case the resource does not exist.

def directChildResources(self, pi: str, ty: ResourceTypes | list[ResourceTypes] | None = None, raw: bool | None = False) -> list[JSON] | list[Resource]: ¶

Return a list of direct child resources, or an empty list

Parameters
pi:`str`	The parent resource's Resource ID.
ty:`ResourceTypes \| list[ResourceTypes] \| None`	Optional resource type or list of resource types to filter the result.
raw:`bool \| None`	When "True" then return the child resources as resource dictionary instead of resources.

Returns
`list[JSON] \| list[Resource]`	Return a list of resources, or a list of raw resource dictionaries.

def directChildResourcesRI(self, pi: str, ty: ResourceTypes | list[ResourceTypes] | None = None) -> list[str]: ¶

Return a list of direct child resource IDs, or an empty list

Parameters
pi:`str`	The parent resource's Resource ID.
ty:`ResourceTypes \| list[ResourceTypes] \| None`	Optional resource type or list of resource types to filter the result.

Returns
`list[str]`	Return a list of resource IDs.

def getActionRepr(self, ri: str) -> JSON | None: ¶

Retrieve an action representation from the DB.

Parameters
ri:`str`	The action's resource ID.

Returns
`JSON \| None`	The action's data as a Document, or None.

def getAllActionReprs(self) -> list[JSON]: ¶

Retrieve all action representations from the DB.

Returns
`list[JSON]`	List of Documents. May be empty.

def getBatchNotifications(self, ri: str, nu: str) -> list[JSON]: ¶

Retrieve the batch notifications for a target resource and a notification URI.

Parameters
ri:`str`	The resource ID of the target resource.
nu:`str`	The notification URI.

Returns
`list[JSON]`	List of batch notifications.

def getOriginator(self, originator: str) -> tuple[str, OriginatorType] | None: ¶

Get an originator and its information from the database.

Parameters
originator:`str`	The originator to search for.

Returns
`tuple[str, OriginatorType] \| None`	Tuple of the originator and its type.

def getRequests(self, ri: str | None = None, sortedByOt: bool = False) -> list[JSON]: ¶

Get requests for a resource ID, or all requests.

Parameters
ri:`str \| None`	The target resource's resource ID. If None or empty, then all requests are returned
sortedByOt:`bool`	If true, then the requests are sorted by their creation time.

Returns
`list[JSON]`	List of Documents. May be empty.

def getSchedules(self) -> list[JSON]: ¶

Retrieve the schedules data from the DB.

Returns
`list[JSON]`	List of Documents. May be empty.

def getStatistics(self) -> JSON: ¶

Retrieve the statistics data from the DB.

Returns
`JSON`	The statistics data as a JSON dictionary.

def getSubscription(self, ri: str) -> JSON | None: ¶

Retrieve a subscription representation (not a oneM2M Resource object) from the DB.

Parameters
ri:`str`	The subscription's resource ID.

Returns
`JSON \| None`	The subscription as a JSON dictionary, or None.

def getSubscriptionsForParent(self, pi: str) -> list[JSON]: ¶

Retrieve all subscriptions representations (not oneM2M Resource objects) for a parent resource.

Parameters
pi:`str`	The parent resource's resource ID.

Returns
`list[JSON]`	List of subscriptions. This is not the oneM2M Subscription resource, but the internal subscription representation.

def hasResource(self, ri: str | None = None, srn: str | None = None) -> bool: ¶

Check whether a resource with either the ri or the srn already exists.

Either one of ri or srn must be provided.

Parameters
ri:`str \| None`	Optional resource ID.
srn:`str \| None`	Optional structured resource name.

Returns
`bool`	True when a resource with the ID or name exists.

def identifier(self, ri: str) -> list[JSON]: ¶

Search for the resource identifer mapping with the given unstructured resource ID.

Parameters
ri:`str`	Unstructured resource ID for the mapping to look for.

Returns
`list[JSON]`	List of found resources identifier mappings, or an empty list.

def initialize(self): ¶

Initialization of the storage manager.

Raises
`RuntimeError`	In case of an error during initialization.

def purge(self): ¶

Reset and clear the databases.

def purgeStatistics(self): ¶

Purge the statistics DB.

Returns
Boolean value to indicate success or failure.

def removeActionRepr(self, ri: str) -> bool: ¶

Remove an action representation from the DB.

Parameters
ri:`str`	The action's resource ID.

Returns
`bool`	Boolean value to indicate success or failure.

def removeBatchNotifications(self, ri: str, nu: str) -> bool: ¶

Remove the batch notifications for a target resource and a notification URI.

Parameters
ri:`str`	The resource ID of the target resource.
nu:`str`	The notification URI.

Returns
`bool`	Boolean value to indicate success or failure.

def removeOriginator(self, originator: str) -> bool: ¶

Remove an originator from the database.

Parameters
originator:`str`	The originator to remove.

Returns
`bool`	Boolean value to indicate success or failure.

def removeSchedule(self, schedule: SCH) -> bool: ¶

Remove a schedule from the DB.

Parameters
schedule:`SCH`	The schedule to remove.

Returns
`bool`	Boolean value to indicate success or failure.

def removeSubscription(self, subscription: Resource) -> bool: ¶

Remove a subscription from the DB.

Parameters
subscription:`Resource`	The subscription `Resource` to remove.

Returns
`bool`	Boolean value to indicate success or failure.

Raises
`NOT_FOUND`	In case the subscription does not exist.

def retrieveResource(self, ri: str | None = None, csi: str | None = None, srn: str | None = None, aei: str | None = None) -> Resource: ¶

Return a resource via different addressing methods.

Either one of ri, srn, csi, or aei must be provided.

Parameters
ri:`str \| None`	The resource is retrieved via its rersource ID.
csi:`str \| None`	The resource is retrieved via its CSE-ID.
srn:`str \| None`	The resource is retrieved via its structured resource name.
aei:`str \| None`	The resource is retrieved via its AE-ID.

Returns
`Resource`	The resource.

Raises
`NOT_FOUND`	In case the resource does not exist.
`INTENRAL_SERVER_ERROR`	In case of a database inconsistency.

def retrieveResourceRaw(self, ri: str) -> JSON: ¶

Retrieve a resource as a raw dictionary.

Parameters
ri:`str`	The resource is retrieved via its rersource ID.

Returns
`JSON`	The resource dictionary.

Raises
`NOT_FOUND`	In case the resource does not exist.
`INTENRAL_SERVER_ERROR`	In case of a database inconsistency.

def retrieveResourcesByType(self, ty: ResourceTypes) -> list[JSON]: ¶

Return all resources of a certain type.

Parameters
ty:`ResourceTypes`	resource type to retrieve.

Returns
`list[JSON]`	List of resource JSON objects, not Resource objects.

def searchActionReprsForSubject(self, subjectRi: str) -> Sequence[JSON]: ¶

Search for action representation for a subject resource.

Parameters
subjectRi:`str`	The subject resource's resource ID.

Returns
`Sequence[JSON]`	List of matching action representations.

def searchByFilter(self, filter: Callable[[JSON], bool]) -> list[Resource]: ¶

Return a list of resources that match the given filter, or an empty list.

Parameters
filter:`Callable[[JSON], bool]`	A callback to provide filter functionality.

Returns
`list[Resource]`	List of `Resource` objects.

def searchByFragment(self, dct: dict, filter: Callable[[JSON], bool] | None = None) -> list[Resource]: ¶

Search and return all resources that match the given fragment dictionary/document.

Parameters
dct:`dict`	A fragment dictionary to use as a filter for the search.
filter:`Callable[[JSON], bool] \| None`	An optional callback to provide additional filter functionality.

Returns
`list[Resource]`	List of `Resource` objects.

def searchScheduleForTarget(self, pi: str) -> list[str]: ¶

Search for schedules for a target resource.

Parameters
pi:`str`	The target resource's resource ID.

Returns
`list[str]`	List of schedule resource IDs.

def shutdown(self) -> bool: ¶

Shutdown the storage manager.

Returns
`bool`	Always True.

def structuredIdentifier(self, srn: str) -> list[JSON]: ¶

Search for the resource identifer mapping with the given structured resource ID.

Parameters
srn:`str`	Structured resource ID for the mapping to look for.

Returns
`list[JSON]`	List of found resources identifier mappings, or an empty list.

def updateActionRepr(self, actionRepr: JSON) -> bool: ¶

Update an action representation (not an actual action) in the DB.

Parameters
actionRepr:`JSON`	The action representation to update.

Returns
`bool`	Boolean value to indicate success or failure.

def updateResource(self, resource: Resource) -> Resource: ¶

Update a resource in the database.

Parameters
resource:`Resource`	Resource to update.

Returns
`Resource`	Updated Resource object.

def updateStatistics(self, stats: JSON) -> bool: ¶

Update the statistics DB with new data.

Parameters
stats:`JSON`	The statistics data to store.

Returns
`bool`	Boolean value to indicate success or failure.

def upsertAction(self, action: ACTR, periodTS: float, count: int) -> bool: ¶

Update or add an action as an action representation in the DB.

Parameters
action:`ACTR`	The action to update or insert.
periodTS:`float`	The period for the action.
count:`int`	The run count for the action.

Returns
`bool`	Boolean value to indicate success or failure.

def upsertSchedule(self, schedule: SCH) -> bool: ¶

Add or update a schedule in the DB.

Parameters
schedule:`SCH`	The schedule to add or update.

Returns
`bool`	Boolean value to indicate success or failure.

def upsertSubscription(self, subscription: Resource) -> bool: ¶

Add or update a subscription to the DB.

Parameters
subscription:`Resource`	The subscription `Resource` to add.

Returns
`bool`	Boolean value to indicate success or failure.

__slots__: tuple[str, ...] = ¶

Define slots for instance variables.

factory: Factory = ¶

Injected Factory instance.

postgreSQLBinding: PostgreSQLBinding = ¶

Injected PostgreSQLBinding instance.

tinyDBBinding: TinyDBBinding = ¶

Injected TinyDBBinding instance.

db: DBBinding = ¶

The database object.

def _validateDB(self) -> bool: ¶

Trying to validate the database files.

This is only a simple test. It performs a couple of read operations on the available database files.

Returns
`bool`	Boolean indicating the validity of the databases.