Skip to content

climate_ref.models.mixins #

Model mixins for shared functionality

CreatedUpdatedMixin #

Mixin for models that have a created_at and updated_at fields

Source code in packages/climate-ref/src/climate_ref/models/mixins.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class CreatedUpdatedMixin:
    """
    Mixin for models that have a created_at and updated_at fields
    """

    created_at: Mapped[datetime.datetime] = mapped_column(server_default=func.now())
    """
    When the dataset was added to the database
    """

    updated_at: Mapped[datetime.datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now(), index=True
    )
    """
    When the dataset was updated.
    """

created_at = mapped_column(server_default=(func.now())) class-attribute instance-attribute #

When the dataset was added to the database

updated_at = mapped_column(server_default=(func.now()), onupdate=(func.now()), index=True) class-attribute instance-attribute #

When the dataset was updated.

DimensionMixin #

Mixin that adds controlled vocabulary dimension support to a model

This mixin provides methods and properties for managing CV dimensions on database models. Dimensions are stored as individual indexed columns that are registered at runtime based on the controlled vocabulary.

Classes using this mixin must: - Define _cv_dimensions as a ClassVar[list[str]] = [] - Have a tablename attribute (SQLAlchemy requirement)

Source code in packages/climate-ref/src/climate_ref/models/mixins.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class DimensionMixin:
    """
    Mixin that adds controlled vocabulary dimension support to a model

    This mixin provides methods and properties for managing CV dimensions
    on database models. Dimensions are stored as individual indexed columns
    that are registered at runtime based on the controlled vocabulary.

    Classes using this mixin must:
    - Define _cv_dimensions as a ClassVar[list[str]] = []
    - Have a __tablename__ attribute (SQLAlchemy requirement)
    """

    _cv_dimensions: ClassVar[list[str]]

    @property
    def dimensions(self) -> dict[str, str]:
        """
        Get the non-null dimensions and their values

        Any changes to the resulting dictionary are not reflected in the object

        Returns
        -------
            Collection of dimension names and their values
        """
        dims = {}
        for key in self._cv_dimensions:
            value = getattr(self, key)
            if value is not None:
                dims[key] = value
        return dims

    @staticmethod
    def build_dimension_column(dimension: "Dimension") -> Column[str]:
        """
        Create a column representing a CV dimension

        These columns are not automatically generated with alembic revisions.
        Any changes to this functionality likely require a manual database migration
        of the existing columns.

        Parameters
        ----------
        dimension
            Dimension definition to create the column for.
            Currently only the "name" field is being used.

        Returns
        -------
            An instance of a sqlalchemy Column

            This doesn't create the column in the database,
            but enables the ORM to access it.
        """
        return Column(
            dimension.name,
            Text,
            index=True,
            nullable=True,
            info={"skip_autogenerate": True},
        )

    @classmethod
    def register_cv_dimensions(cls, cv: "CV") -> None:
        """
        Register the dimensions supplied in the controlled vocabulary

        This has to be done at run-time to support custom CVs.
        Any extra columns already in the database, but not in the CV are ignored.

        Parameters
        ----------
        cv
            Controlled vocabulary being used by the application.
            This controlled vocabulary contains the definitions of the dimensions that can be used.
        """
        model_name = cls.__name__

        for dimension in cv.dimensions:
            target_attribute = dimension.name
            if target_attribute in cls._cv_dimensions:
                continue

            cls._cv_dimensions.append(target_attribute)
            logger.debug(f"Registered {model_name} dimension: {target_attribute}")

            if hasattr(cls, target_attribute):
                # This should only occur in test suite as we don't support removing dimensions at runtime
                logger.warning(f"Column attribute already exists on {model_name}. Ignoring")
            else:
                setattr(cls, target_attribute, cls.build_dimension_column(dimension))

    @classmethod
    def _reset_cv_dimensions(cls) -> None:
        """
        Remove any previously registered dimensions

        Used by the test suite and should not be called at runtime.

        This doesn't remove any previous column definitions due to a limitation that columns in
        declarative classes cannot be removed.
        This means that `hasattr(cls, "old_attribute")`
        will still return True after resetting, but the values will not be included in any executions.
        """
        model_name = cls.__name__
        logger.warning(f"Removing {model_name} dimensions: {cls._cv_dimensions}")

        keys = list(cls._cv_dimensions)
        for key in keys:
            cls._cv_dimensions.remove(key)

        assert not len(cls._cv_dimensions)

dimensions property #

Get the non-null dimensions and their values

Any changes to the resulting dictionary are not reflected in the object

Returns:

Type Description
Collection of dimension names and their values

build_dimension_column(dimension) staticmethod #

Create a column representing a CV dimension

These columns are not automatically generated with alembic revisions. Any changes to this functionality likely require a manual database migration of the existing columns.

Parameters:

Name Type Description Default
dimension Dimension

Dimension definition to create the column for. Currently only the "name" field is being used.

 required

Returns:

Type Description
An instance of a sqlalchemy Column

This doesn't create the column in the database, but enables the ORM to access it.
Source code in packages/climate-ref/src/climate_ref/models/mixins.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@staticmethod
def build_dimension_column(dimension: "Dimension") -> Column[str]:
    """
    Create a column representing a CV dimension

    These columns are not automatically generated with alembic revisions.
    Any changes to this functionality likely require a manual database migration
    of the existing columns.

    Parameters
    ----------
    dimension
        Dimension definition to create the column for.
        Currently only the "name" field is being used.

    Returns
    -------
        An instance of a sqlalchemy Column

        This doesn't create the column in the database,
        but enables the ORM to access it.
    """
    return Column(
        dimension.name,
        Text,
        index=True,
        nullable=True,
        info={"skip_autogenerate": True},
    )

register_cv_dimensions(cv) classmethod #

Register the dimensions supplied in the controlled vocabulary

This has to be done at run-time to support custom CVs. Any extra columns already in the database, but not in the CV are ignored.

Parameters:

Name Type Description Default
cv CV

Controlled vocabulary being used by the application. This controlled vocabulary contains the definitions of the dimensions that can be used.

 required
Source code in packages/climate-ref/src/climate_ref/models/mixins.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@classmethod
def register_cv_dimensions(cls, cv: "CV") -> None:
    """
    Register the dimensions supplied in the controlled vocabulary

    This has to be done at run-time to support custom CVs.
    Any extra columns already in the database, but not in the CV are ignored.

    Parameters
    ----------
    cv
        Controlled vocabulary being used by the application.
        This controlled vocabulary contains the definitions of the dimensions that can be used.
    """
    model_name = cls.__name__

    for dimension in cv.dimensions:
        target_attribute = dimension.name
        if target_attribute in cls._cv_dimensions:
            continue

        cls._cv_dimensions.append(target_attribute)
        logger.debug(f"Registered {model_name} dimension: {target_attribute}")

        if hasattr(cls, target_attribute):
            # This should only occur in test suite as we don't support removing dimensions at runtime
            logger.warning(f"Column attribute already exists on {model_name}. Ignoring")
        else:
            setattr(cls, target_attribute, cls.build_dimension_column(dimension))