Context package

Submodules

Context.ContextCli module

class Context.ContextCli.ContextCli(host: str, port: int, base_path: str = '', notification_uri: str | None = None, check_subscription_conflicts: bool = False)[source]

Bases: object

A client for managing common operations on a context broker.

Initialize the ContextCli.

Parameters:

  • host (str) – Address of the context broker.

  • port (int) – Port of the context broker.

  • base_path (str, optional) – URL path to the context broker. Defaults to “”.

  • notification_uri (str, optional) – The URI used for the subscription notifications. Defaults to None.

  • check_subscription_conflicts (bool, optional) – If True, the subscription will be checked for conflicts before being created. Defaults to False.

property broker_url: str
delete_entity(entity_id: str) → bool[source]

Delete an entity from the context broker.

Parameters:

entity_id (str) – The ID of the entity to delete.

Raises:

requests.exceptions.HTTPError – If there was an error deleting the entity.

Returns:

True if successful.

Return type:

bool

get_all_entities(entity_type: List[str] | str | None = None, attrs: List[str] | str | None = None, entity_id: List[str] | str | None = None, id_pattern: str | None = None, query: str | None = None, order_by: str | None = None, as_dict: bool = False) → List[Type[BaseModel]] | List[dict][source]

Get all entities from the context broker.

Parameters:

  • entity_type (Optional[Union[List[str], str]], optional) – A single or a list of entity types. Defaults to None.

  • attrs (Optional[Union[List[str], str]], optional) – A single or a list of attributes to return. Defaults to None.

  • entity_id (Optional[Union[List[str], str]], optional) – A single or a list of entity IDs. Defaults to None.

  • id_pattern (Optional[str], optional) – A pattern to match the entity IDs. Defaults to None.

  • query (Optional[str], optional) – A query to filter entities. Defaults to None.

  • order_by (Optional[str], optional) – Order entities by an attribute. Comma-separated list of attributes can be used to order by the first attribute and on tie ones ordered by the subsequent attributes. A “!” before the attribute name means that the order is reversed. e.g. !dateCreated. Default to None.

  • as_dict (bool, optional) – If True, the entities will be returned as dictionaries. Otherwise, they will be converted to a toolbox data model. Defaults to False.

Raises:

  • requests.exceptions.HTTPError – If there was an error getting the entities.

  • KeyError – If the entity type is not recognized and as_dict is False.

Returns:

A list of data model

objects or dictionaries.

Return type:

Union[List[Type[BaseModel]], List[dict]]

get_all_subscriptions() → List[Subscription][source]

Get all the current subscriptions in the context broker.

Raises:

requests.exceptions.HTTPError – If the subscriptions could not be retrieved successfully.

Returns:

List of the subscriptions in the context broker.

Return type:

List[Subscription]

get_conflicting_subscriptions(subscription: Subscription) → List[Subscription][source]

Get a list of subscriptions in the context broker that are virtually the same as the given subscription.

Parameters:

subscription (Subscription) – A subscription object.

Raises:

requests.exceptions.HTTPError – If the subscriptions could not be retrieved successfully.

Returns:

A list with conflicting subscriptions.

Return type:

List[Subscription]

get_entities_page(entity_type: List[str] | str | None = None, attrs: List[str] | str | None = None, entity_id: List[str] | str | None = None, id_pattern: str | None = None, query: str | None = None, limit: int = 100, offset: int = 0, order_by: str | None = None, as_dict: bool = False) → List[Type[BaseModel]] | List[dict][source]

Get a list of entities from the context broker.

Parameters:

  • entity_type (Optional[Union[List[str], str]], optional) – A single or a list of entity types. Defaults to None.

  • attrs (Optional[Union[List[str], str]], optional) – A single or a list of attributes to return. Defaults to None.

  • entity_id (Optional[Union[List[str], str]], optional) – A single or a list of entity IDs. Defaults to None.

  • id_pattern (Optional[str], optional) – A pattern to match the entity IDs. Defaults to None.

  • query (Optional[str], optional) – A query to filter entities. Defaults to None.

  • limit (int, optional) – Maximum number of entities to return. The maximum value is 1000. Defaults to 100.

  • offset (int, optional) – Pagination offset. Defaults to 0.

  • order_by (Optional[str], optional) – Order entities by an attribute. Comma-separated list of attributes can be used to order by the first attribute and on tie ones ordered by the subsequent attributes. A “!” before the attribute name means that the order is reversed. e.g. !dateCreated. Default to None.

  • as_dict (bool, optional) – If True, the entities will be returned as dictionaries. Otherwise, they will be converted to a toolbox data model. Defaults to False.

Raises:

  • requests.exceptions.HTTPError – If there was an error getting the entities.

  • KeyError – If the entity type is not recognized and as_dict is False.

Returns:

A list of data model

objects or dictionaries.

Return type:

Union[List[Type[BaseModel]], List[dict]]

get_entity(entity_id: str, as_dict: bool = False) → Type[BaseModel] | dict | None[source]

Retrieve an entity from the context broker by its ID.

Parameters:

  • entity_id (str) – The ID of an entity.

  • as_dict (bool, optional) – If True, the entity will be returned as a dictionary. Otherwise, it will be converted to a toolbox data model. Defaults to False.

Raises:

  • requests.exceptions.HTTPError – If there was an error getting the entity.

  • KeyError – If the entity type is not recognized and as_dict is False.

Returns:

A data model object, a

dictionary or None if the entity does not exist.

Return type:

Union[Type[BaseModel], dict, None]

get_subscription(subscription_id: str) → Subscription | None[source]

Get a subscription from the context broker by its id.

Parameters:

subscription_id (str) – The subscription id.

Raises:

requests.exceptions.HTTPError – If there was an error getting the subscription.

Returns:

The retrieved Subscription object or

None if the subscription was not found.

Return type:

Union[Subscription, None]

get_subscriptions_page(limit: int = 100, offset: int = 0) → List[Subscription][source]

Get a list of subscriptions in the context broker.

Parameters:

  • limit (int, optional) – Maximum number of subscriptions to return. Maximum value is 1000. Defaults to 100.

  • offset (int, optional) – Pagination offset. Defaults to 0.

Raises:

requests.exceptions.HTTPError – If the subscriptions could not be retrieved successfully.

Returns:

List of Subscription objects.

Return type:

List[Subscription]

get_types() → List[str][source]

Get a list of the current entity types in the context broker.

Returns:

A list of entity types.

Return type:

List[str]

headers: dict

The headers used in requests.

iterate_entities(entity_type: List[str] | str | None = None, attrs: List[str] | str | None = None, entity_id: List[str] | str | None = None, id_pattern: str | None = None, query: str | None = None, limit: int = 100, order_by: str | None = None, as_dict: bool = False) → Iterator[List[Type[BaseModel]] | List[dict]][source]

Iterate through a list of entities from the context broker.

Parameters:

  • entity_type (Optional[Union[List[str], str]], optional) – A single or a list of entity types. Defaults to None.

  • attrs (Optional[Union[List[str], str]], optional) – A single or a list of attributes to return. Defaults to None.

  • entity_id (Optional[Union[List[str], str]], optional) – A single or a list of entity IDs. Defaults to None.

  • id_pattern (Optional[str], optional) – A pattern to match the entity IDs. Defaults to None.

  • query (Optional[str], optional) – A query to filter entities. Defaults to None.

  • limit (int, optional) – Maximum number of entities to return. The maximum value is 1000. Defaults to 100.

  • order_by (Optional[str], optional) – Order entities by an attribute. Comma-separated list of attributes can be used to order by the first attribute and on tie ones ordered by the subsequent attributes. A “!” before the attribute name means that the order is reversed. e.g. !dateCreated. Default to None.

  • as_dict (bool, optional) – If True, the entities will be returned as dictionaries. Otherwise, they will be converted to a toolbox data model. Defaults to False.

Raises:

  • requests.exceptions.HTTPError – If there was an error getting the entities.

  • KeyError – If the entity type is not recognized and as_dict is False.

Returns:

An iterator of

data model objects or dictionaries.

Return type:

Iterator[Union[List[Type[BaseModel]], List[dict]]]

iterate_subscriptions(limit: int = 100) → Iterator[List[Subscription]][source]

Iterate over the current subscriptions in the context broker.

Parameters:

limit (int, optional) – Maximum number of subscriptions to return in each iteration. Defaults to 100.

Raises:

requests.exceptions.HTTPError – If the subscriptions could not be retrieved successfully.

Yields:

Iterator[List[Subscription]] – Iterator over the subscriptions.

post_data_model(data_model: Type[BaseModel]) → dict[source]

Post a toolbox data model object to the context broker. If the data model ID is None, a new one will be assigned.

Parameters:

data_model (Type[DataModels.BaseModel]) – The data model object to upload to the context broker.

Raises:

requests.exceptions.HTTPError – If there was an error posting the data model.

Returns:

The uploaded JSON.

Return type:

dict

post_entity_json(entity: dict)[source]

Post a JSON entity to the context broker.

Parameters:

  • entity (dict) – The entity to upload to the context broker as a dictionary.

  • ngsi_ld (bool, optional) – If True, the entity data is in NGSI-LD format. Defaults to True.

Raises:

requests.exceptions.HTTPError – If there was an error posting the entity.

subscribe(subscription: Subscription | None = None, **kwargs) → str[source]

Create a subscription in the context broker from a Subscription object or from the given kwargs.

Parameters:

  • subscription (Optional[Subscription]) – A Subscription object. If None, a Subscription will be built from the kwargs. Defaults to None.

  • kwargs – The subscription data as keyword arguments (see Subscription for the list of valid arguments). If the subscription is not None, the kwargs will be ignored. Defaults to None.

Raises:

  • requests.exceptions.HTTPError – If the subscription could not be created.

  • Exception – If there was an error creating the subscription.

Returns:

The subscription id.

Return type:

str

property subscription_ids: List[str]

Get the list of subscription IDs created within the ContextCli.

subscription_name: str

The name used in subscriptions

unsubscribe(subscription_id: str) → bool[source]

Delete a subscription from the context broker.

Parameters:

subscription_id (str) – The id of the subscription to delete.

Raises:

requests.exceptions.HTTPError – If there was an error deleting the subscription.

Returns:

True if successful.

Return type:

bool

unsubscribe_all() → bool[source]

Delete all the subscriptions created within the ContextCli.

Raises:

requests.exceptions.HTTPError – If the subscriptions could not be retrieved successfully.

Returns:

True if all the subscriptions were deleted successfully.

Return type:

bool

update_data_model(data_model: Type[BaseModel], create: bool = True) → dict[source]

Update an existing entity in the context broker.

Parameters:

  • data_model (Type[DataModels.BaseModel]) – The data model to update.

  • create (bool) – If the entity should be created if it does not exists. Defaults to False.

Raises:

  • requests.exceptions.HTTPError – If there was an error updating the entity.

  • ValueError – If the entity does not exist and create is False.

Returns:

The updated JSON.

Return type:

dict

update_entity_json(entity: dict, create: bool = True) → dict[source]

Update a JSON entity in the context broker.

Parameters:

  • entity (dict) – The entity to update in the context broker as a dictionary.

  • create (bool, optional) – If True, the entity will be created if it does not exist in the context broker. Defaults to True.

Raises:

  • requests.exceptions.HTTPError – If there was an error updating the entity.

  • ValueError – If the entity does not exist and create is False.

Returns:

The updated entity.

Return type:

dict

Context.Subscription module

class Context.Subscription.Subscription(notification_uri: str, subscription_id: str | None = None, name: str | None = None, description: str | None = None, entity_type: List[str] | str | None = None, entity_id: List[str] | str | None = None, entity_id_pattern: List[str] | str | None = None, watched_attributes: List[str] | None = None, query: str | None = None, notification_attributes: List[str] | None = None, notification_format: str = 'normalized', notification_accept: str = 'application/json', expires: str | datetime | None = None, throttling: int | None = None)[source]

Bases: object

Class that represents a context broker subscription to one or more entities.

notification_uri
subscription_id
name
description
entity_type
entity_id
entity_id_pattern
watched_attributes
query
notification_attributes
notification_format
notification_accept
expires
throttling
Overloaded operators:

  • __eq__

  • __str__

Create a subscription object.

Parameters:

  • entity_type (Optional[Union[str, List[str]]]) – Entity type to subscribe to. If None, watched_attributes must be provided. A single entity type or a list of entity types.

  • notification_uri (str) – URI which conveys the endpoint which will receive the notification.

  • subscription_id (Optional[str], optional) – Subscription ID. If None, a new one will be generated. Defaults to None.

  • name (Optional[str], optional) – A (short) name given to this Subscription. Defaults to None.

  • description (Optional[str], optional) – Subscription description. Defaults to None.

  • entity_id (Optional[Union[str, List[str]]], optional) – ID of the entity to subscribe or a list of entity IDs to subscribe to. If not provided, all entities of the given entity type will be subscribed. If used, entity_id_pattern must be None. Defaults to None.

  • entity_id_pattern (Optional[Union[str, List[str]]], optional) – A regular expression that denotes a pattern that shall be matched by the provided or subscribed Entities. A single or a list of expressions. If used, entity_id must be None. Defaults to None.

  • watched_attributes (Optional[List[str]], optional) – Watched Attributes (Properties or Relationships). If None, entity_type must be provided and all attributes will be watched. Defaults to None.

  • query (Optional[str], optional) – Query that shall be met by subscribed entities in order to trigger the notification. Defaults to None.

  • notification_attributes (Optional[List[str]], optional) – Entity Attribute Names (Properties or Relationships) to be included in the notification payload body. If None it will mean all Attributes. Defaults to None.

  • notification_format (str, optional) – Conveys the representation format of the entities delivered at notification time. “keyValues” or “normalized”. Defaults to “normalized”.

  • notification_accept (str, optional) – MIME type of the notification payload body (“application/json” or “application/ld+json”). Defaults to “application/json”.

  • expires (Optional[Union[str, datetime]], optional) – Expiration date for the subscription. datetime object or ISO 8601 String. Defaults to None.

  • throttling (Optional[int], optional) – Minimal period of time in seconds which shall elapse between two consecutive notifications. Defaults to None.

Raises:

ValueError – If the provided arguments are not valid.

static from_json(json: dict) → Subscription[source]

Builds a subscription from a JSON.

Parameters:

json (dict) – Subscription JSON dict.

Returns:

A Subscription object.

Return type:

Subscription

property json: dict

Returns the subscription as a JSON object.

Context.entity_parser module

Context.entity_parser.create_random_id(entity_type: str = '', prefix: str = 'urn:ngsi-ld:', uuid4: bool = False, shortener: bool = True) → str[source]

Create a random URN ID with the entity type and a UUID.

Parameters:

  • entity_type (str, optional) – The type of the entity. Defaults to “”.

  • prefix (str, optional) – Optional prefix to include at the start of the id. Defaults to “urn:ngsi-ld”.

  • uuid4 (bool, optional) – If UUID4 should be used instead of UUID1. Defaults to False.

  • shortener (bool, optional) – Use a shorter version of the UUID. Defaults to True.

Returns:

A random id.

Return type:

str

Context.entity_parser.data_model_to_json(data_model: Type[BaseModel]) → dict[source]

Parse a toolbox data model to an NGSI-LD entity JSON. If the data model id is None, a new one will be set.

Parameters:

data_model (Type[BaseModel]) – The toolbox data model object to parse.

Returns:

The parsed data model as a NGSI-LD entity JSON.

Return type:

dict

Context.entity_parser.get_entity_field(field: any, field_type: any) → any[source]

Convert a field from an entity to its data type.

Parameters:

  • field (any) – The field value.

  • field_type (any) – The data type to which cast the field.

Returns:

The field converted to the data type.

Return type:

any

Context.entity_parser.json_to_data_model(entity: dict) → Type[BaseModel][source]

Parse an entity from a dict to a toolbox data model object.

Parameters:

entity (dict) – A ngsi-ld entity as a dict.

Raises:

KeyError – If the entity type is not recognized.

Returns:

A data model object.

Return type:

Type[BaseModel]

Context.entity_parser.parse_entity(entity: dict, data_model_type: Type[BaseModel]) → Type[BaseModel][source]

Parse an NGSI-LD JSON entity and convert it to a toolbox data model.

Parameters:

  • entity (dict) – The entity dictionary to be parsed.

  • data_model_type (Type[BaseModel]) – The data model class to which convert the entity. One from toolbox.DataModels.

Raises:

  • TypeError – If the entity type does not match the data model type.

  • ValueError – If the entity does not have a field that is required by the data model.

Returns:

The parsed entity.

Return type:

Type[BaseModel]

Context.entity_parser.set_entity_field(entity: Entity, field: any, name: str)[source]

Parse and add a new field to an NGSI-LD entity.

Parameters:

  • entity (Entity) – The Entity object to add a field.

  • field (any) – The value to be added.

  • name (str) – The field name.

Module contents