Chrononaut’s API

Core library classes

class chrononaut.Versioned

A mixin for use with Flask-SQLAlchemy declarative models. To get started, simply add the Versioned mixin to one of your models:

class User(db.Model, Versioned):
    __tablename__ = 'appuser'
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255))
    ...

The above will then automatically track updates to the User model and create an appuser_history table for tracking prior versions of each record. By default, all columns are tracked. By default, change information includes a user_id and remote_addr, which are set to automatically populate from Flask-Login’s current_user in the _capture_change_info() method. Subclass Versioned and override a combination of _capture_change_info(), _fetch_current_user_id(), and _get_custom_change_info(). This change_info is stored in a JSON column in your application’s database and has the following rough layout:

{
    "user_id": "A unique user ID (string) or None",
    "remote_addr": "The user IP (string) or None",
    "extra": {
        ...  # Optional extra fields
    },
    "hidden_cols_changed": [
        ...  # A list of any hidden fields changed in the version
    ]
}

Note that the latter two keys will not exist if they would otherwise be empty. You may provide a list of column names that you do not want to track using the optional __chrononaut_untracked__ field or you may provide a list of columns you’d like to “hide” (i.e., track updates to the columns but not their values) using the __chrononaut_hidden__ field. This can be useful for sensitive values, e.g., passwords, which you do not want to retain indefinitely.

_capture_change_info()

Capture the change info for the new version. By default calls:

  1. _fetch_current_user_id() which should return a string or None; and
  2. _fetch_remote_addr() which should return an IP address string or None;
  3. _get_custom_change_info() which should return a 1-depth dict of additional keys.

These 3 methods generate a change_info and with 2+ top-level keys (user_id, remote_addr, and any keys from _get_custom_change_info())

_fetch_current_user_id()

Return the current user ID.

Returns:A unique user ID string or None if not available.
_fetch_remote_addr()

Return the IP address for the current user.

Returns:An IP address string or None if not available.
_get_custom_change_info()

Optionally return additional change_info fields to be inserted into the history record.

Returns:A dictionary of additional change_info keys and values
diff(from_model, to=None, include_hidden=False)

Enumerate the changes from a prior history model to a later history model or the current model’s state (if to is None).

Parameters:
  • from_model – A history model to diff from.
  • to – A history model or None.
Returns:

A dict of column names and (from, to) value tuples

has_changed_since(since)

Check if there are any changes since a given time.

Parameters:since – The DateTime from which to find any history records
Returns:True if there have been any changes. False if not.
version_at(at)

Fetch the history model at a specific time (or None)

Parameters:at – The DateTime at which to find the history record.
Returns:A history model at the given point in time or the model itself if that is current.
versions(before=None, after=None, return_query=False)

Fetch the history of the given object from its history table.

Parameters:
  • before – Return changes only _before_ the provided DateTime.
  • before – Return changes only _after_ the provided DateTime.
  • return_query – Return a SQLAlchemy query instead of a list of models.
Returns:

List of history models for the given object (or a query object).

class chrononaut.VersionedSQLAlchemy(app=None, use_native_unicode=True, session_options=None, metadata=None, query_class=<class 'flask_sqlalchemy.BaseQuery'>, model_class=<class 'flask_sqlalchemy.model.Model'>)

A subclass of the SQLAlchemy used to control a SQLAlchemy integration to a Flask application.

Two usage modes are supported (as in Flask-SQLAlchemy). One is directly binding to a Flask application:

app = Flask(__name__)
db = VersionedSQLAlchemy(app)

The other is by creating the db object and then later initializing it for the application:

db = VersionedSQLAlchemy()

# Later/elsewhere
def configure_app():
    app = Flask(__name__)
    db.init_app(app)
    return app

At its core, the VersionedSQLAlchemy class simply ensures that database session objects properly listen to events and create version records for models with the Versioned mixin.

Helper functions

chrononaut.extra_change_info(*args, **kwds)

A context manager for appending extra change_info into Chrononaut history records for Versioned models.

Usage:

with extra_change_info(change_rationale='User request'):
    user.email = 'new-email@example.com'
    db.session.commit()

Note that the db.session.commit() change needs to occur within the context manager block for additional fields to get injected into the history table change_info JSON within an extra info field. Any number of keyword arguments with string values are supported.

The above example yields a change_info like the following:

{
    "user_id": "admin@example.com",
    "remote_addr": "127.0.0.1",
    "extra": {
        "change_rationale": "User request"
    }
}