ochra.manager.connections

db_connection

class ochra.manager.connections.db_connection.DbConnection(*args, **kwds)

Bases: object

DbConnection is a singleton class that provides an interface for interacting with any database. This class currently acts as a wrapper around the MongoAdapter, offering CRUD operations and query methods for managing documents within specified collections. However, any database adapter that adheres to the expected interface can be integrated in the future.

db_adapter

Adapter for MongoDB operations. This can be replaced with any other database adapter that follows the same interface.

Type:: MongoAdapter

Initialize a DbConnection instance.

Parameters:

hostname (str, optional) – Address of the database host. Defaults to “127.0.0.1:27017”.
db_name (str, optional) – Name of the database. Defaults to “ochra_test_db”.
args (Any)
kwds (Any)

Return type:

Any

create(db_data, doc)

Create a new document in the specified collection.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection.
doc (Dict[str, Any]) – The document to be created.

Returns:

The result of the create operation, typically the created document or its identifier.

Return type:

Any

delete(db_data)

Delete documents from the specified collection that match the query.

Parameters:: db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection and query parameters.
Returns:: The result of the delete operation, typically a status indicator or the count of deleted documents.
Return type:: Any

find(db_data, search_params)

Find documents from the specified collection that match the search parameters.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection.
search_params (Dict[str, Any]) – The search parameters to filter the documents.

Returns:

The result of the find operation, typically a matching document or None if not found.

Return type:

Any

find_all(db_data, search_params)

Find all documents from the specified collection that match the query.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection.
search_params (Dict[str, Any]) – The search parameters to filter the documents.

Returns:

The result of the find_all operation, typically a list of matching documents or an empty list if none found.

Return type:

Any

read(db_data, property=None, file=False)

Read documents from the specified collection that match the query.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection and query parameters.
property (str, optional) – Specific property to retrieve from the documents. Defaults to None.
file (bool, optional) – Flag indicating if the read operation involves file data. Defaults to False.

Returns:

The result of the read operation, which could be a document, a specific property, or file data.

Return type:

Any

update(db_data, update, file=False)

Update documents in the specified collection that match the query.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection and query parameters.
update (Dict[str, Any]) – The update operations to be applied to the matching documents.
file (bool, optional) – Flag indicating if the update operation involves file data. Defaults to False.

Returns:

The result of the update operation, typically the updated document or a status indicator.

Return type:

Any

mongo_adapter

class ochra.manager.connections.mongo_adapter.MongoAdapter(hostname, db_name, logger=None)

Bases: object

Adapter class for interacting with a MongoDB database. This class provides methods to connect to a MongoDB instance, perform CRUD operations, manage collections, handle GridFS file storage, and perform advanced updates on documents.

Initialize the MongoAdapter.

Parameters:

hostname (str) – MongoDB server address.
db_name (str) – Name of the database to connect to.
logger (logging.Logger, optional) – Logger instance for logging. Defaults to None.

clear_collection(collection)

clears the given collection inside the db

Parameters:: collection (str) – name of collection
Returns:: None
Return type:: None

create(db_data, document)

Create a new document in the specified collection.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection.
document (Document) – The document to be created.

Returns:

The result of the create operation, typically the created document or its identifier.

delete(db_data)

Delete documents from the specified collection that match the query.

Parameters:: db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection and query parameters.
Returns:: The result of the delete operation, typically a status indicator or the count of deleted documents.
Return type:: Any

delete_database()

Deletes database from host

Returns:: None
Return type:: None

find(db_data, search_params)

Find documents from the specified collection that match the search parameters.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection.
search_params (Dict[str, Any]) – The search parameters to filter the documents.

Returns:

The result of the find operation, typically a matching document or None if not found.

Return type:

Any

find_all(db_data, search_params)

Find all documents from the specified collection that match the query.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection.
search_params (Dict[str, Any]) – The search parameters to filter the documents.

Returns:

The result of the find_all operation, typically a list of matching documents or an empty list if none found.

Return type:

Any

is_collection_populated(collection)

Check if collection exists on the db

Parameters:: collection (str) – name of the collection
Returns:: True if collection exists. Otherwise False
Return type:: bool

is_database_existing()

Check if database exists on the host

Returns:: True if database exists. Otherwise False
Return type:: bool

read(db_data, property, file=False)

Read documents from the specified collection that match the query.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection and query parameters.
property (str, optional) – Specific property to retrieve from the documents. Defaults to None.
file (bool, optional) – Flag indicating if the read operation involves file data. Defaults to False.

Returns:

The result of the read operation, which could be a document, a specific property, or file data.

Return type:

Any

update(db_data, update, file=False)

Update documents in the specified collection that match the query.

Parameters:

db_data (Dict[str, Any]) – Dictionary containing database information, including the target collection and query parameters.
update (Dict[str, Any]) – The update operations to be applied to the matching documents.
file (bool, optional) – Flag indicating if the update operation involves file data. Defaults to False.

Returns:

The result of the update operation, typically the updated document or a status indicator.

Return type:

Any

station_connection

class ochra.manager.connections.station_connection.StationConnection(hostname='10.24.57.154:8000', api_key='', ssl_verify=False)

Bases: object

Class that provides a high-level interface for interacting with a remote station, utilizing RestAdapter for communication.

Parameters:

hostname (str)
api_key (str)
ssl_verify (bool)

execute_op(op, endpoint)

Execute an operation on the remote station.

Parameters:

op (Operation) – The operation to be executed.
endpoint (str) – The API endpoint for executing the operation.

Returns:

The response from the remote station after executing the operation.

Return type:

Result