Add version support and migrations
This is much simpler than using pydantic-versions, which seems not to be well-maintained and is buggy.
- Add docs / tutorial
- add metdata change migration step, for e.g. changing fits_keyword.
- make
Productversioned
This provides a very simple internally implemented (no external package) method for schema change tracking. So far it supports only simple operations like rename and drp, but is easy to extend if we need it.
It works as follows: only top-level classes (the ones that are
serialized) need inherit from VersionedModel, and then they can
contain functions decorated with @migration that define the changes
from one version to another. The history of changes is tracked and can
be read programmatically (with VersionedModel.migration_history()),
and the migrations can be applied to pre-validated dicts using
VersionedModel.model_validate_versioned(data)
The design is quite simple, and easy to extend: there is a base Migration class with implementation so far for history recording and for actually doing the migration on a dict. Others could be implemented.
Migrations implemented in this version
- rename field (affects JSON/YAML representation)
- change field metadata (e.g.
fits_keyword) (affects FITS representation) - drop field (affects all)
- change default value (affects all)
- apply function transform (json)
Example:
class SubModel(ModelBase):
a: str
b_renamed: str
class Target(VersionedModel):
"""A simple example"""
name: str = AstroField("target name", fits_keyword="OBJECT")
ra: Quantity["deg"] = AstroField(
"Right ascension", fits_column_dtype="float64", ucd="pos.eq.ra"
)
dec: Quantity["deg"] = AstroField(
"Declination", fits_column_dtype="float64", ucd="pos.eq.dec"
)
sub: SubModel
@migration("1.0.0", "1.1.0")
def _(m):
m.rename("object_name", "name")
m.rename("sub.b", "sub.b_renamed")
m.drop("sub.c")
return m
print(Target.migration_history())[{'from': '1.0.0',
'to': '1.1.0',
'operations': ["Rename 'object_name' → 'name'",
"Rename 'sub.b' → 'sub.b_renamed'",
"Drop 'sub.c'"]}]And then using it for validation:
targ_v1 = dict(
object_name="Crab",
ra=83.624 * u.deg,
dec=22.0174 * u.deg,
sub=dict(a="x", b="y", c="z"),
)
targ_v2 = dict(
name="Crab", ra=83.624 * u.deg, dec=22.0174 * u.deg, sub=dict(a="x", b_renamed="y")
)
v1 = Target.model_validate_versioned(targ_v1)
v2 = Target.model_validate_versioned(targ_v2)
print(v1)
print(v2)Target(name='Crab', ra=<Quantity 83.624 deg>, dec=<Quantity 22.0174 deg>, sub=SubModel(a='x', b_renamed='y'))
Target(name='Crab', ra=<Quantity 83.624 deg>, dec=<Quantity 22.0174 deg>, sub=SubModel(a='x', b_renamed='y'))FITS keyword migrations are also supported
class Target(VersionedModel):
"""A simple example"""
name: str = AstroField("target name", fits_keyword="OBJECT")
ra: Quantity[u.deg] = AstroField(
"Right ascension", ucd="pos.eq.ra", fits_keyword="RA"
)
dec: Quantity[u.deg] = AstroField(
"Declination", ucd="pos.eq.dec", fits_keyword="DEC"
)
@migration("1.0.0", "1.1.0")
def _(m: Migration):
m.update_metadata("ra", "fits_keyword", "RA2000", "RA")
m.update_metadata("ra", "fits_keyword", "DEC2000", "DEC")
header_v1 = fits.Header(dict(
OBJECT="Crab",
RA2000=83.624,
DEC2000=22.0174 ,
))
print("*** ORIGINAL, OLD FORMAT")
print(header_v1)
print("*** MIGRATED")
print(dm.instance_to_fits_header(dm.fits_header_to_instance(header_v1, Target)))*** ORIGINAL, OLD FORMAT
OBJECT = 'Crab '
RA2000 = 83.624
DEC2000 = 22.0174
*** MIGRATED
OBJECT = 'Crab ' / target name
RA = 83.624 / [deg] Right ascension
DEC = 22.0174 / [deg] Declination