sqlmodel_data_gateway

oxytcmri.infrastructure.gateways.sqlmodel_data_gateway

A module using SQLModel ORM, with SQLite as the database engine.

Classes:

Name	Description
BaseDTO	Base class for all Data Transfer Objects (DTO) with entity conversion methods.
SubjectDTO	DTO for Subject entity with database mapping.
AtlasDTO	DTO for Atlas entity with database mapping.
MRIDataType	Enum for mapping MRIData subtypes.
MRIDataDTO	DTO for MRIData entity with database mapping.
MRIExamDTO	DTO for MRIExam entity with database mapping.
CenterDTO	DTO for Center entity with database mapping.
NormativeValuesDTO	DTO for NormativeValues entity with database mapping.
SQLModelSQLiteDataGateway	SQLModel implementation of DataBaseGateway for SQLite.

Attributes:

Name	Type	Description
logger
EntityType

logger = logging.getLogger(__name__) module-attribute

EntityType = TypeVar('EntityType') module-attribute

BaseDTO

Bases: SQLModel, Generic[EntityType], ABC

Base class for all Data Transfer Objects (DTO) with entity conversion methods.

Methods:

Name	Description
from_entity	Create a DTO from a domain entity.
to_entity	Convert the DTO to a domain entity.

from_entity(entity) abstractmethod classmethod

Create a DTO from a domain entity.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
@abstractmethod
def from_entity(cls, entity: EntityType) -> "BaseDTO[EntityType]":
    """Create a DTO from a domain entity."""

to_entity() abstractmethod

Convert the DTO to a domain entity.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@abstractmethod
def to_entity(self) -> EntityType:
    """Convert the DTO to a domain entity."""

SubjectDTO

Bases: BaseDTO[Subject]

DTO for Subject entity with database mapping.

Methods:

Name	Description
from_entity
to_entity

Attributes:

Name	Type	Description
id	str
subject_type	str
center_id	int

id = Field(primary_key=True) class-attribute instance-attribute

subject_type instance-attribute

center_id instance-attribute

from_entity(entity) classmethod

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
def from_entity(cls, entity: Subject) -> "SubjectDTO":
    return cls(id=str(entity.id), subject_type=entity.subject_type, center_id=entity.center_id)

to_entity()

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def to_entity(self) -> Subject:
    return Subject.from_string_id(self.id)

AtlasDTO

Bases: BaseDTO[Atlas]

DTO for Atlas entity with database mapping.

Methods:

Name	Description
from_entity
to_entity

Attributes:

Name	Type	Description
id	int
name	str
labels	List[int]

id = Field(primary_key=True) class-attribute instance-attribute

name instance-attribute

labels = Field(sa_type=JSON) class-attribute instance-attribute

from_entity(entity) classmethod

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
def from_entity(cls, entity: Atlas) -> "AtlasDTO":
    return cls(id=entity.id, name=entity.name, labels=entity.labels)

to_entity()

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def to_entity(self) -> Atlas:
    return Atlas(id=self.id, name=self.name, labels=self.labels)

MRIDataType

Bases: str, Enum

Enum for mapping MRIData subtypes.

Attributes:

Name	Type	Description
GENERIC
ATLAS_SEGMENTATION
DTI_MAP

GENERIC = 'generic' class-attribute instance-attribute

ATLAS_SEGMENTATION = 'atlas_segmentation' class-attribute instance-attribute

DTI_MAP = 'dti_map' class-attribute instance-attribute

MRIDataDTO

Bases: BaseDTO[MRIData]

DTO for MRIData entity with database mapping.

Warnings

This class assumes that the voxel data is of type NiftiVoxelData.

Attributes:

Name	Type	Description
id	str	Unique identifier (primary key) of the MRI data
mri_exam_id	str	Unique identifier of the MRI exam (foreign key)
name	str	Name of the MRI data
nifti_data_path	str	Path to the Nifti file

Methods:

Name	Description
from_entity
to_entity

id = Field(primary_key=True) class-attribute instance-attribute

mri_exam_id = Field(foreign_key='mri_exams.id') class-attribute instance-attribute

mri_exam = Relationship(back_populates='mri_data') class-attribute instance-attribute

name instance-attribute

nifti_data_path instance-attribute

data_type = Field(default=MRIDataType.GENERIC) class-attribute instance-attribute

atlas_id = Field(default=None, foreign_key='atlases.id') class-attribute instance-attribute

atlas = Relationship() class-attribute instance-attribute

dti_metric = Field(default=None) class-attribute instance-attribute

from_entity(entity) classmethod

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
def from_entity(cls, entity: MRIData) -> "MRIDataDTO":
    if type(entity.get_voxel_data()) != NiftiVoxelData:
        raise ValueError(f"Unsupported voxel data type: {type(entity.voxel_data)}")

    # ensure voxel_data is of type NiftiVoxelData
    voxel_data = cast(NiftiVoxelData, entity.voxel_data)

    # default values
    data_type = MRIDataType.GENERIC
    atlas_id = None
    dti_metric = None

    if isinstance(entity, AtlasSegmentation):
        data_type = MRIDataType.ATLAS_SEGMENTATION
        atlas_id = entity.atlas.id
    elif isinstance(entity, DTIMap):
        data_type = MRIDataType.DTI_MAP
        dti_metric = str(entity.dti_metric)

    return cls(id=entity.id,
               mri_exam_id=str(entity.mri_exam_id),
               name=entity.name,
               nifti_data_path=voxel_data.get_nifti_path_string(),
               data_type=data_type,
               atlas_id=atlas_id,
               dti_metric=dti_metric)

to_entity()

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def to_entity(self) -> MRIData:
    if self.data_type == MRIDataType.ATLAS_SEGMENTATION:
        return AtlasSegmentation(id=self.id,
                                 name=self.name,
                                 voxel_data=NiftiVoxelData(Path(self.nifti_data_path)),
                                 atlas=self.atlas.to_entity())
    elif self.data_type == MRIDataType.DTI_MAP:
        return DTIMap(id=self.id,
                      name=self.name,
                      voxel_data=NiftiVoxelData(Path(self.nifti_data_path)),
                      dti_metric=self.dti_metric)
    return MRIData(id=self.id,
                   name=self.name,
                   voxel_data=NiftiVoxelData(Path(self.nifti_data_path)))

MRIExamDTO

Bases: BaseDTO[MRIExam]

DTO for MRIExam entity with database mapping.

Methods:

Name	Description
from_entity
to_entity

Attributes:

Name	Type	Description
id	str
subject_id	str
mri_data	list[MRIDataDTO]

id = Field(primary_key=True) class-attribute instance-attribute

subject_id = Field(foreign_key='subjects.id') class-attribute instance-attribute

mri_data = Relationship(back_populates='mri_exam') class-attribute instance-attribute

from_entity(entity) classmethod

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
def from_entity(cls, entity: MRIExam) -> "MRIExamDTO":
    return cls(id=str(entity.id), subject_id=str(entity.subject_id))

to_entity()

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def to_entity(self) -> MRIExam:
    return MRIExam.from_string_exam_id(
        exam_id=self.id,
        data=[MRIDataDTO.to_entity(mri_data) for mri_data in self.mri_data]
    )

CenterDTO

Bases: BaseDTO[Center]

DTO for Center entity with database mapping.

Methods:

Name	Description
from_entity
to_entity

Attributes:

Name	Type	Description
id	int
name	str

id = Field(primary_key=True) class-attribute instance-attribute

name instance-attribute

from_entity(entity) classmethod

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
def from_entity(cls, entity: Center) -> "CenterDTO":
    return cls(id=entity.id, name=entity.name)

to_entity()

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def to_entity(self) -> Center:
    return Center(id=self.id, name=self.name)

NormativeValuesDTO

Bases: BaseDTO[NormativeValue]

DTO for NormativeValues entity with database mapping.

Attributes:

Name	Type	Description
id	int	Unique identifier (primary key) of the normative value
value	float	Value of the normative value
center_id	int	Unique identifier of the center (foreign key)
center	CenterDTO	Center of the normative value
dti_metric	DTIMetric	DTI metric of the normative value
atlas_id	int	Unique identifier of the atlas (foreign key)
atlas_label	int	Label of the atlas
atlas	AtlasDTO	Atlas of the normative value
statistic_strategy	StatisticsStrategies	Statistic strategy of the normative value

Methods:

Name	Description
from_entity
to_entity

id = Field(default=None, primary_key=True) class-attribute instance-attribute

value instance-attribute

center_id = Field(foreign_key='centers.id') class-attribute instance-attribute

center = Relationship() class-attribute instance-attribute

dti_metric instance-attribute

atlas_id = Field(foreign_key='atlases.id') class-attribute instance-attribute

atlas_label instance-attribute

atlas = Relationship() class-attribute instance-attribute

statistic_strategy instance-attribute

from_entity(entity) classmethod

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

@classmethod
def from_entity(cls, entity: EntityType) -> "NormativeValuesDTO":
    return cls(
        center_id=entity.center.id,
        dti_metric=str(entity.dti_metric),
        atlas_id=entity.atlas.id,
        atlas_label=entity.atlas_label,
        statistic_strategy=entity.statistic_strategy.name,
        value=entity.value
    )

to_entity()

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def to_entity(self) -> NormativeValue:
    return NormativeValue(center=self.center.to_entity(),
                          dti_metric=DTIMetric.from_acronym(self.dti_metric),
                          atlas=self.atlas.to_entity(),
                          atlas_label=self.atlas_label,
                          statistic_strategy=StatisticsStrategies.get_by_name(self.statistic_strategy),
                          value=self.value)

SQLModelSQLiteDataGateway(database_path)

Bases: DataBaseGateway[EntityType]

SQLModel implementation of DataBaseGateway for SQLite.

Initialize the gateway with the path to the database.

Parameters:

Name	Type	Description	Default
database_path	str	Path to the SQLite file	required

Methods:

Name	Description
update
find_by_id	Retrieve an entity by its ID.
find_by_filters	Retrieve an entity by filters.
find_all	Retrieve all entities of the given type.
save	Save an entity (create or update).
save_list	Save a list of entities to the database in a single transaction.
delete	Delete an entity.

Attributes:

Name	Type	Description
database_path
engine
entity_to_model_map	Dict[Type, Type[SQLModel]]
type_converters

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def __init__(self, database_path: str):
    """
    Initialize the gateway with the path to the database.

    Parameters
    ----------
    database_path : str
        Path to the SQLite file
    """
    self.database_path = Path(database_path)
    # Ensure the parent directory exists
    self.database_path.parent.mkdir(parents=True, exist_ok=True)

    # Create the engine
    self.engine = create_engine(f"sqlite:///{database_path}", echo=False)

    # Create tables if they don't exist
    SQLModel.metadata.create_all(self.engine)

    # Mapping from entity types to SQLModel model classes
    self.entity_to_model_map: Dict[Type, Type[SQLModel]] = {
        Center: CenterDTO,
        Subject: SubjectDTO,
        Atlas: AtlasDTO,
        MRIExam: MRIExamDTO,
        MRIData: MRIDataDTO,
        AtlasSegmentation: MRIDataDTO,
        DTIMap: MRIDataDTO,
        NormativeValue: NormativeValuesDTO,
    }

    # Define type converters for different entity types
    self.type_converters = {
        SubjectId: lambda x: str(x),
        MRIExamId: lambda x: str(x),
        DTIMetric: lambda x: str(x),
        StatisticStrategy: lambda x: x.name,
    }

database_path = Path(database_path) instance-attribute

engine = create_engine(f'sqlite:///{database_path}', echo=False) instance-attribute

entity_to_model_map = {Center: CenterDTO, Subject: SubjectDTO, Atlas: AtlasDTO, MRIExam: MRIExamDTO, MRIData: MRIDataDTO, AtlasSegmentation: MRIDataDTO, DTIMap: MRIDataDTO, NormativeValue: NormativeValuesDTO} instance-attribute

type_converters = {SubjectId: lambda x: str(x), MRIExamId: lambda x: str(x), DTIMetric: lambda x: str(x), StatisticStrategy: lambda x: x.name} instance-attribute

update(entity)

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def update(self, entity: EntityType) -> None:
    raise NotImplementedError("Update method is not implemented yet.")  # pragma: no cover

find_by_id(entity_type, id_value)

Retrieve an entity by its ID.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def find_by_id(self, entity_type: Type[EntityType], id_value: Any) -> Optional[EntityType]:
    """Retrieve an entity by its ID."""
    model_class = self._get_model_class(entity_type)
    converted_id_value = self._convert_id_value(id_value)
    try:
        with Session(self.engine) as session:
            statement = select(model_class).where(model_class.id == converted_id_value)
            model = session.exec(statement).first()

            if model is None:
                return None

            return self._model_to_entity(model, entity_type)
    except ProgrammingError as programming_error:
        raise RuntimeError(f"Error executing query: {statement} "
                           f"when trying to find entity {entity_type} "
                           f"with id {id_value}") \
            from programming_error

find_by_filters(entity_type, filters)

Retrieve an entity by filters.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def find_by_filters(self, entity_type: Type[Entity], filters: dict[str, Any]) -> Optional[Entity]:
    """Retrieve an entity by filters."""
    model_class = self._get_model_class(entity_type)

    with Session(self.engine) as session:
        # Start with a base query
        query = select(model_class)

        # Process filters to make them compatible with SQLite
        processed_filters = self._prepare_filters_for_query(filters)

        # Apply filters to the query
        for key, value in processed_filters.items():
            if not hasattr(model_class, key):
                continue

            attr = getattr(model_class, key)

            # If the attribute is a relationship, we need to join the related model
            if hasattr(attr, 'property') and hasattr(attr.property, 'mapper'):
                related_model = attr.property.mapper.class_
                # Perform a join
                query = query.join(related_model)

                # Obtain the primary key name of the related model
                primary_key_column = related_model.__table__.primary_key.columns.values()[0]
                primary_key_name = primary_key_column.name

                # Filter by the primary key of the related model
                if hasattr(value, primary_key_name):
                    pk_value = getattr(value, primary_key_name)
                    query = query.where(getattr(related_model, primary_key_name) == pk_value)
                else:
                    raise ValueError(f"Value {value} does not have the primary key {primary_key_name} "
                                     f"for related model {related_model.__name__}")
            else:
                # When the attribute is not a relationship, we can filter directly
                query = query.where(attr == value)

        model = session.exec(query).first()

        if model is None:
            return None

        return self._model_to_entity(model, entity_type)

find_all(entity_type)

Retrieve all entities of the given type.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def find_all(self, entity_type: Type[EntityType]) -> List[EntityType]:
    """Retrieve all entities of the given type."""
    model_class = self._get_model_class(entity_type)

    with Session(self.engine) as session:
        statement = select(model_class)
        models = session.exec(statement).all()

        # Convert models to entities
        entities = [self._model_to_entity(model, entity_type) for model in models]
        return entities

save(entity)

Save an entity (create or update).

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def save(self, entity: EntityType) -> None:
    """Save an entity (create or update)."""
    self.save_list([entity])

save_list(entities)

Save a list of entities to the database in a single transaction.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def save_list(self, entities: List[EntityType]) -> None:
    """Save a list of entities to the database in a single transaction."""
    if not entities:
        return

    with Session(self.engine) as session:
        for entity in entities:
            model = self._entity_to_model(entity)
            session.merge(model)
        session.commit()

delete(entity)

Delete an entity.

Source code in oxytcmri/infrastructure/gateways/sqlmodel_data_gateway.py

def delete(self, entity: EntityType) -> None:
    """Delete an entity."""
    model_class = self._get_model_class(type(entity))

    with Session(self.engine) as session:
        statement = select(model_class).where(model_class.id == entity.id)
        model = session.exec(statement).first()

        if model:
            session.delete(model)
            session.commit()