API¶
Configuration Database¶
SDP State¶
High-level SDP state API.
-
class
sip_config_db.states.sdp_state.
SDPState
[source]¶ SDP state data object.
-
allowed_state_transitions
¶ Get dictionary of allowed state transitions.
-
allowed_states
¶ Get list of allowed object states.
-
allowed_target_states
¶ Get dictionary of allowed target states / commands.
-
current_state
¶ Get the current state.
-
current_timestamp
¶ Get the current state timestamp.
-
get_event_queue
(subscriber: str)¶ Get an event queue for the specified subscriber.
-
static
get_subscribers
() → List[str]¶ Get the list of subscribers to state events.
- Returns:
List[str], list of subscriber names.
-
id
¶ Return the object id.
-
is_target_state_allowed
(value)¶ Test if a transition is allowed.
- Args:
value (str): New value for target state
- Returns:
bool, transition is allowed
-
publish
(event_type: str, event_data: dict = None)¶ Publish an state event.
- Args:
event_type (str): Type of event. event_data (dict, optional): Event data.
-
static
subscribe
(subscriber: str) → sip_config_db._events.event_queue.EventQueue¶ Subscribe to state events.
- Args:
subscriber (str): Subscriber name.
- Returns:
events.EventQueue, Event queue object for querying events.
-
target_state
¶ Get the target state.
-
target_timestamp
¶ Get the target state timestamp.
-
update_current_state
(value: str, force: bool = False) → datetime.datetime¶ Update the current state.
- Args:
value (str): New value for sdp state force (bool): If true, ignore allowed transitions
- Returns:
datetime, update timestamp
- Raises:
ValueError: If the specified current state is not allowed.
-
update_target_state
(value: str, force: bool = True) → datetime.datetime¶ Set the target state.
- Args:
value (str): New value for target state force (bool): If true, ignore allowed transitions
- Returns:
datetime, update timestamp
- Raises:
RuntimeError, if it is not possible to currently set the target state. ValueError, if the specified target stat is not allowed.
-
Service States¶
High-level service state API.
-
class
sip_config_db.states.service_state.
ServiceState
(subsystem: str, name: str, version: str)[source]¶ SDP state data object.
-
static
get_service_state_object_id
(subsystem: str, name: str, version: str) → str[source]¶ Return service state data object key.
- Args:
subsystem (str): Subsystem the service belongs to name (str): Name of the Service version (str): Version of the Service
- Returns:
str, Key used to store the service state data object
-
name
¶ Get the name of the service.
-
subsystem
¶ Get the subsystem the service belongs to.
-
version
¶ Get the version of the service.
-
static
Services¶
High-level service state API.
-
class
sip_config_db.states.service_state.
ServiceState
(subsystem: str, name: str, version: str)[source] SDP state data object.
-
static
get_service_state_object_id
(subsystem: str, name: str, version: str) → str[source] Return service state data object key.
- Args:
subsystem (str): Subsystem the service belongs to name (str): Name of the Service version (str): Version of the Service
- Returns:
str, Key used to store the service state data object
-
name
Get the name of the service.
-
subsystem
Get the subsystem the service belongs to.
-
version
Get the version of the service.
-
static
Utility modules¶
Misc functions for interacting with datetime.
-
sip_config_db.utils.datetime_utils.
datetime_from_isoformat
(value: str)[source]¶ Return a datetime object from an isoformat string.
- Args:
value (str): Datetime string in isoformat.
Module to generate test data for sbi and pb client.
FIXME(BMo) Move this to a method on the SchedulingBlockInstance object?
-
sip_config_db.utils.generate_sbi_config.
add_workflow_definitions
(sbi_config: dict)[source]¶ Add any missing SBI workflow definitions as placeholders.
This is a utility function used in testing and adds mock / test workflow definitions to the database for workflows defined in the specified SBI config.
- Args:
sbi_config (dict): SBI configuration dictionary.
-
sip_config_db.utils.generate_sbi_config.
generate_pb_config
(pb_id: str, pb_config: dict = None, workflow_config: dict = None) → dict[source]¶ Generate a PB configuration dictionary.
- Args:
pb_id (str): Processing Block Id pb_config (dict, optional) PB configuration. workflow_config (dict, optional): Workflow configuration
- Returns:
dict, PB configuration dictionary.
-
sip_config_db.utils.generate_sbi_config.
generate_sb
(date: datetime.datetime, project: str, programme_block: str) → dict[source]¶ Generate a Scheduling Block data object.
- Args:
date (datetime.datetime): UTC date of the SBI project (str): Project Name programme_block (str): Programme
- Returns:
str, Scheduling Block Instance (SBI) ID.
-
sip_config_db.utils.generate_sbi_config.
generate_sbi_config
(num_pbs: int = 3, project: str = 'sip', programme_block: str = 'sip_demos', pb_config: Union[dict, List[dict]] = None, workflow_config: Union[dict, List[dict]] = None, register_workflows=False) → dict[source]¶ Generate a SBI configuration dictionary.
- Args:
num_pbs (int, optional): Number of Processing Blocks (default = 3) project (str, optional): Project to associate the SBI with. programme_block (str, optional): SBI programme block pb_config (dict, List[dict], optional): PB configuration workflow_config (dict, List[dict], optional): Workflow configuration register_workflows (bool, optional): If true also register workflows.
- Returns:
dict, SBI configuration dictionary
-
sip_config_db.utils.generate_sbi_config.
generate_sbi_json
(num_pbs: int = 3, project: str = 'sip', programme_block: str = 'sip_demos', pb_config: Union[dict, List[dict]] = None, workflow_config: Union[dict, List[dict]] = None, register_workflows=True) → str[source]¶ Return a JSON string used to configure an SBI.
-
sip_config_db.utils.generate_sbi_config.
generate_version
(max_major: int = 1, max_minor: int = 7, max_patch: int = 15) → str[source]¶ Select a random version.
- Args:
max_major (int, optional) maximum major version max_minor (int, optional) maximum minor version max_patch (int, optional) maximum patch version
- Returns:
str, Version String