cobra_db.mongo_dao

Module Contents

Classes

Connector	Connect and auth to a mongo database.
EntityDao	Base class for all entity DAOs.
PatientDao	Get or insert patient from / to the database
StudyDao	Get or insert RadiologicalStudy from / to the database
SeriesDao	Get or insert RadiologicalSeries from / to the database
ImageMetadataDao	Get or insert an instance of ImageMetadata from / to the database

Functions

get_obj_id(→ bson.ObjectId)

Get ObjectId regardless of the initial type

cobra_db.mongo_dao.get_obj_id(_id: cobra_db.types.Id) → bson.ObjectId

Get ObjectId regardless of the initial type

class cobra_db.mongo_dao.Connector(host: str, port: int, db_name: str, username: str = None, password: str = None)

Connect and auth to a mongo database.

connect()

Connect to the mongo database

static _get_uri(host: str, port: str, username=None, password=None, db_name='')

close()

Close the connection to the database

classmethod get_pass(host: str, port: int, db_name: str, username: str)

Create a Connector instance by prompting the user for a password. Most useful in jupyter notebooks.

classmethod get_env_pass(host: str, port: int, db_name: str, username: str, env_var: str = 'MONGOPASS')

Create a Connector instance by getting the password from the environment variable defined in env_var. Useful when creating scripts.

__str__()

Return str(self).

class cobra_db.mongo_dao.EntityDao(connector: Connector, entity_type: Type[cobra_db.model.Entity])

Base class for all entity DAOs. Children need to implement a concrete to_dict() method to be able to use get_by_id(x, obj=True)

_get_collection(entity_type: Type[cobra_db.model.Entity]) → pymongo.collection.Collection

Get a collection from the database according to the entity type

get_by_id(_id: cobra_db.types.Id, obj=True) → Union[cobra_db.model.Entity, dict]

Find a document by _id.

find(filter: dict = {}, **kwargs) → email.generator.Generator

Generate instances for the documents that match the filter. This is not a very fast approach, but it is easy to use.

filter: dict, valid mongodb query to find the desired entities. kwargs to pass to pymongo.Collection.find(filter, **kwargs)

delete_by_id(_id: cobra_db.types.Id)

Delete a document from the database by its id.

Parameters:

_id – ObjectId or str version of ObjectId of the document that will be

deleted.

insert_one(entity_instance: cobra_db.model.Entity) → bson.ObjectId

Insert an instance into the database.

class cobra_db.mongo_dao.PatientDao(connector)

Bases: EntityDao

Get or insert patient from / to the database (Manages access to the Patient collection in MongoDB)

_ensure_index_exists()

Check if the index for anon_id exists

anon_id_to_mongo_id(anon_id: str) → bson.ObjectId

Get the _id of a patient from its anon_id

get_patient_by_id(_id: cobra_db.types.Id = None, obj=True) → Union[cobra_db.model.Patient, dict]

Get a patient from its _id

get_patient_by_anon_id(anon_id: str, obj=True) → Union[cobra_db.model.Patient, dict]

Get a patient from its anon_id

get_patient(patient: cobra_db.model.Patient, obj=True) → Union[cobra_db.model.Patient, dict]

Get a patient if it exists in the database. Expect an IndexError if it does not exist.

exists(patient: cobra_db.model.Patient) → Tuple[bool, Union[bson.ObjectId, None]]

insert_one(patient: cobra_db.model.Patient, skip_duplicates=True)

Insert a patient into the database. raises DuplicateKeyError if the patient already exists and check_unique is False

get_by_id(_id: cobra_db.types.Id, obj=True) → Union[cobra_db.model.Patient, dict]

redifining only for type hinting

class cobra_db.mongo_dao.StudyDao(connector)

Bases: EntityDao

Get or insert RadiologicalStudy from / to the database (Manages access to the RadiologicalStudy collection in MongoDB)

_ensure_index_exists()

Check if the index for study_uid exists

insert_one(study: cobra_db.model.RadiologicalStudy, skip_duplicates=True)

Insert a study into the database. raises DuplicateKeyError if the study already exists and skip_duplicates is False

update_patient_id(study_id: cobra_db.types.Id, patient_id: cobra_db.types.Id)

Once the patient doc is created the series_id of the metadata should be updated

get_all_ids(patient: Union[cobra_db.model.Patient, cobra_db.types.Id], *modality: str, other_filters: dict = {}) → List[bson.ObjectId]

Given a patient or patient_id, retrieve all studies that match any of the specified modality. other_filters allows you to narrow down the query.

get_by_id(_id: cobra_db.types.Id, obj=True) → Union[cobra_db.model.RadiologicalStudy, dict]

Get a RadiologicalStudy by _id

class cobra_db.mongo_dao.SeriesDao(connector)

Bases: EntityDao

Get or insert RadiologicalSeries from / to the database (Manages access to the RadiologicalSeries collection in MongoDB)

_ensure_index_exists()

Check if the index for anon_series_id exists

insert_one(series: cobra_db.model.RadiologicalSeries, study: cobra_db.model.RadiologicalStudy = None, skip_duplicates=True)

Insert a series into the database first retrieving the study id. raises DuplicateKeyError if the series already exists and skip_duplicates is False

update_study_id(series_id: cobra_db.types.Id, study_id: cobra_db.types.Id)

Once the series doc is created the series_id of the metadata should be updated

get_all_ids(study: Union[cobra_db.model.RadiologicalStudy, cobra_db.types.Id], other_filters: dict = {})

Given a study or study_id, retrieve all series that belong to this study. other_filter can be used to narrow down the query.

get_by_id(_id: cobra_db.types.Id, obj=True) → Union[cobra_db.model.RadiologicalSeries, dict]

Get RadiologicalSeries by _id

class cobra_db.mongo_dao.ImageMetadataDao(connector)

Bases: EntityDao

Get or insert an instance of ImageMetadata from / to the database (Manages access to the ImageMetadata collection in MongoDB)

_ensure_index_exists()

Check if the index for anon_image_id exists

insert_one(im_metadata: cobra_db.model.ImageMetadata)

Insert an instance into the database.

update_series_id(im_meta_id: cobra_db.types.Id, series_id: cobra_db.types.Id)

Once the series doc is created the series_id of the metadata should be updated

update_study_id(im_meta_id: cobra_db.types.Id, study_id: cobra_db.types.Id)

Once the series doc is created the series_id of the metadata should be updated

get_all_ids(series: Union[cobra_db.model.RadiologicalSeries, cobra_db.types.Id], other_filters: dict = {})

Given a series or series_id, retrieve all images that belong to this series. other_filter can be used to narrow down the query.

get_by_id(_id: cobra_db.types.Id, obj=True) → Union[cobra_db.model.ImageMetadata, dict]

Get an ImageMetadata by _id

add_aka(im_meta_id: cobra_db.types.Id, file_source: cobra_db.model.FileSource)

Adds an AKA file source. This is useful when a pseudonymized version of the file is saved somewhere else. :param im_meta_id: The id of the document that will be updated :param file_source: The AKA filesource