-
Notifications
You must be signed in to change notification settings - Fork 5
SQLAlchemy attributes and units
Note: This isn't fully implemented yet (although it is for the speed example given here).
All attributes on SQLAlchemy objects (State
, Contact
etc) that have a dimension will be stored in the database in SI units, and properties will be used so that the value has to be set as a value with units (a pint Quantity
), and is always returned as a Quantity.
To do this, properties are defined on the SQLAlchemy objects through a mixin class, which allows the sharing of these properties between the SQLite classes (defined in sqlite_db.py
) and the Postgres classes (defined in postgres_db.py
). These mixins are located in common_db.py
and are have names like StateMixin
or ContactMixin
.
An example will be given with the State.speed attribute, which should be stored in the database in metres per second. To restrict access to the actual speed
attribute on the State
class, it is renamed _speed
and the definition is changed from:
speed = Column(REAL)
to
_speed = Column("speed", REAL)
with the extra first argument passed to Column giving an explicit name for the column, so the actual column names don't have _
's prepended to them.
Then, in the StateMixin
, three parts of the property are defined:
@hybrid_property
def speed(self):
# Return all speeds as metres per second
if self._speed is None:
return None
else:
return self._speed * (unit_registry.metre / unit_registry.second)
This is the 'getter' for the property, and returns Quantity
objects with units of metres per second. The check for None
has to be done to ensure that this field can be left blank if necessary (as speed
is an optional field - defined with the default SQLAlchemy option of nullable=True
).
The setter is more complicated:
@speed.setter
def speed(self, speed):
if speed is None:
self._speed = None
return
# Check the given speed is a Quantity with a dimension of 'length / time'
try:
if not speed.check('[length]/[time]'):
raise ValueError('Speed must be a Quantity with a dimensionality of [length]/[time]')
except AttributeError:
raise TypeError('Speed must be a Quantity')
# Set the actual speed attribute to the given value converted to metres per second
self._speed = speed.to(unit_registry.metre / unit_registry.second).magnitude
This again checks for None
and returns it if necessary (as it is impossible to set units on a value of None
), and then checks various aspects of the given Quantity: firstly that it is a Quantity (checked implicitly by the speed.check check) and then that it has the required dimensionality. It is then converted to metres per second, and the underlying _speed
attribute it set to the magnitude of its value.
One more part of the property is required to be defined. This is the code that will be used to extract the value for a SQLAlchemy expression, when the property value is used on a class rather than an instance. For example, when running .filter(State.speed = 5)
. In this case, SQLAlchemy can't cope with a quantity here, as the underlying SQL doesn't know anything about quantities - so we must return a scalar value, knowing that it is in metres per second:
@speed.expression
def speed(self):
# We need a separate @speed.expression function to return a float rather than a
# Quantity object, as otherwise this won't work in the SQLAlchemy filtering functions
return self._speed
The same applies to all other attributes that have dimensions