utils

scheduler

class ochra.manager.lab.utils.scheduler.Scheduler

Bases: object

A class to manage and schedule operations in the lab.

op_queue

A list to hold the queued operations.

Type:

list

add_operation(operation)

Adds an operation to the scheduling queue.

Parameters:

operation (Operation) – The operation to be added to the queue.

Return type:

None

run()

Starts the scheduling thread.

Return type:

None

stop()

Stops the scheduling thread.

Return type:

None

lab_service

class ochra.manager.lab.utils.lab_service.LabService(folderpath=None)

Bases: object

LabService provides common functionality for managing laboratory objects and operations, abstracting database interactions and file management to avoid code duplication across routers.

db_conn

Database connection instance for CRUD operations.

Type:

DbConnection

folderpath

Path to the folder for storing files, if provided.

Type:

Optional[Path]

Initialize the LabService with an optional folder path for file storage.

Parameters:

folderpath (Optional[str]) – Path to the folder for storing files. If None, file operations are disabled.

call_on_object(object_id, object_type, call_req)

Invoke a method on the specified object and record the operation.

Parameters:

  • object_id (str) – ID of the target object.

  • object_type (str) – Type of the target object.

  • call_req (ObjectCallRequest) – Request containing method name, arguments, and caller information.

Returns:

The created Operation instance representing the method call.

Return type:

Operation

Raises:

HTTPException – If the operation cannot be created or stored.

construct_object(construct_req, collection)

Create or update an object in the specified database collection.

Parameters:

  • construct_req (ObjectConstructionRequest) – Request containing the object’s JSON definition.

  • collection (str) – Name of the database collection.

Returns:

The ID of the constructed or updated object.

Return type:

str

Raises:

HTTPException – If object creation or update fails.

delete_object(object_id, collection)

Delete an object from the specified database collection.

Parameters:

  • object_id (str) – Unique identifier of the object to delete.

  • collection (str) – Name of the database collection containing the object.

Raises:

HTTPException – If the object is not found or deletion fails.

Return type:

None

get_all_objects(collection, query_dict=None)

Retrieve all objects from the specified collection, optionally filtered by a query.

Parameters:

  • collection (str) – Name of the database collection containing the objects.

  • query_dict (Dict[str, Any], optional) – Dictionary specifying query filters. Defaults to None.

Returns:

List of objects represented as JSON dictionaries.

Return type:

List[Dict[str, Any]]

Raises:

HTTPException – If no objects are found or retrieval fails.

get_file(object_id, collection)

Retrieve the file associated with an object from the database and local storage.

Parameters:

  • object_id (str) – Unique identifier of the object whose file is to be retrieved.

  • collection (str) – Name of the database collection containing the object.

Returns:

The file path and a flag indicating if the file should be deleted after use.

Return type:

Tuple[Path, bool]

Raises:

HTTPException – If the folder path is not set or the file cannot be found.

get_object_by_id(object_id, collection)

Retrieve an object by its unique ID from the specified collection.

Parameters:

  • object_id (str) – Unique identifier of the object to retrieve.

  • collection (str) – The database collection containing the object.

Returns:

The object’s JSON representation.

Return type:

Dict[str, Any]

Raises:

HTTPException – If the object is not found.

get_object_by_name(name, collection)

Retrieve an object by its name from the specified collection.

Parameters:

  • name (str) – The name of the object to retrieve.

  • collection (str) – The database collection containing the object.

Returns:

The object’s JSON representation.

Return type:

Dict[str, Any]

Raises:

HTTPException – If the object is not found.

get_object_property(object_id, collection, request)

Retrieve a specific property value from an object in the given collection.

Parameters:

  • object_id (str) – Unique identifier of the object.

  • collection (str) – Name of the database collection containing the object.

  • request (ObjectPropertyGetRequest) – Request specifying the property to retrieve.

Returns:

The value of the requested property.

Return type:

Any

Raises:

HTTPException – If the object or property is not found.

patch_file(object_id, collection, result_data)

Update the file associated with an object in the database and manage its storage.

Parameters:

  • object_id (str) – Unique identifier of the object to update.

  • collection (str) – Name of the database collection containing the object.

  • result_data (bytes) – Binary data to store as the file.

Return type:

None

patch_object(object_id, collection, set_req, file=False)

Update properties of an object in the specified collection.

Parameters:

  • object_id (str) – Unique identifier of the object to update.

  • collection (str) – Name of the database collection containing the object.

  • set_req (ObjectPropertyPatchRequest) – Request containing the property name and new value.

  • file (bool, optional) – Indicates if the property being updated is a file. Defaults to False.

Returns:

True if the update was successful.

Return type:

bool

Raises:

HTTPException – If the object does not exist or the update fails.

lab_logging

ochra.manager.lab.utils.lab_logging.configure_lab_logging(log_root_path, console_log_level=20)

Configures logging for the lab management system.

Parameters:

  • log_root_path (str) – The root directory where log files will be stored.

  • console_log_level (int) – The logging level for the console output (e.g., logging.INFO, logging.DEBUG).

Return type:

None