Module

pghistory

pghistory.AllChange module-attribute

AllChange = pgtrigger.AllChange

If all supplied fields change, trigger the event.

Parameters:

Name	Type	Description	Default
*fields	str	If all supplied fields change, trigger the event. If no fields are supplied, defaults to all tracked fields.	required
exclude	List[str]	Fields to exclude.	required
exclude_auto	bool	Exclude all auto_now and auto_now_add fields automatically.	required

pghistory.AllDontChange module-attribute

AllDontChange = pgtrigger.AllDontChange

If all supplied fields don't change, trigger the event.

Parameters:

Name	Type	Description	Default
*fields	str	If all supplied fields don't change, trigger the event. If no fields are supplied, defaults to all tracked fields.	required
exclude	List[str]	Fields to exclude.	required
exclude_auto	bool	Exclude all auto_now and auto_now_add fields automatically.	required

pghistory.AnyChange module-attribute

AnyChange = pgtrigger.AnyChange

If any supplied fields change, trigger the event.

Parameters:

Name	Type	Description	Default
*fields	str	If any supplied fields change, trigger the event. If no fields are supplied, defaults to all tracked fields.	required
exclude	List[str]	Fields to exclude.	required
exclude_auto	bool	Exclude all auto_now and auto_now_add fields automatically.	required

pghistory.AnyDontChange module-attribute

AnyDontChange = pgtrigger.AnyDontChange

If any supplied fields don't change, trigger the event.

Parameters:

Name	Type	Description	Default
*fields	str	If any supplied fields don't change, trigger the event. If no fields are supplied, defaults to all tracked fields.	required
exclude	List[str]	Fields to exclude.	required
exclude_auto	bool	Exclude all auto_now and auto_now_add fields automatically.	required

pghistory.Condition module-attribute

Condition = pgtrigger.Condition

For specifying free-form SQL in the condition of a trigger.

pghistory.DEFAULT module-attribute

DEFAULT = object()

For setting a configuration value back to its default value

pghistory.Delete module-attribute

Delete = pgtrigger.Delete

For specifying DELETE as the trigger operation.

pghistory.F module-attribute

F = pgtrigger.F

Similar to Django's F object, allows referencing the old and new rows in a trigger condition.

pghistory.Insert module-attribute

Insert = pgtrigger.Insert

For specifying INSERT as the trigger operation.

pghistory.New module-attribute

New = 'NEW'

For storing the trigger's "NEW" row in a pghistory.RowEvent

pghistory.Old module-attribute

Old = 'OLD'

For storing the trigger's "OLD" row in a pghistory.RowEvent

pghistory.Q module-attribute

Q = pgtrigger.Q

Similar to Django's Q object, allows building filter clauses based on the old and new rows in a trigger condition.

pghistory.Update module-attribute

Update = pgtrigger.Update

For specifying UPDATE as the trigger operation.

pghistory.ContextForeignKey

ContextForeignKey(
    *,
    null: bool = True,
    related_query_name: str = constants.DEFAULT,
    **kwargs: Any
)

Bases: ForeignKey

Configuration for the pgh_context field when a foreign key is used.

Overrides null to True.

Attributes:

Name	Type	Description
null	bool, default=True	True if nullable context is allowed
related_query_name	str	The related_query_name to use

Source code in pghistory/config.py

def __init__(
    self, *, null: bool = True, related_query_name: str = constants.DEFAULT, **kwargs: Any
):
    # Note: We will be changing the default context field to have on_delete=PROTECT
    # in version 3.
    super().__init__(null=null, related_query_name=related_query_name, **kwargs)
    self._kwargs.update(_get_kwargs(locals()))

pghistory.ContextJSONField

ContextJSONField(*, null: bool = True, **kwargs: Any)

Bases: Field

Configuration for the pgh_context field when denormalized context is used.

Attributes:

Name	Type	Description
null	bool, default=True	True if nullable context is allowed

Source code in pghistory/config.py

def __init__(self, *, null: bool = True, **kwargs: Any):
    super().__init__(null=null, **kwargs)
    self._kwargs.update(_get_kwargs(locals()))

pghistory.ContextUUIDField

ContextUUIDField(*, null: bool = True, **kwargs: Any)

Bases: Field

Configuration for the pgh_context_id field when denormalized context is used.

Attributes:

Name	Type	Description
null	bool, default=True	True if nullable context is allowed

Source code in pghistory/config.py

def __init__(self, *, null: bool = True, **kwargs: Any):
    super().__init__(null=null, **kwargs)
    self._kwargs.update(_get_kwargs(locals()))

pghistory.DeleteEvent

DeleteEvent(
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None
)

Bases: RowEvent

Creates events based on deletes to a model

The default label used is "delete".

Source code in pghistory/core.py

def __init__(
    self,
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None,
):
    super().__init__(label=label)

    self.condition = condition or self.condition
    self.operation = operation or self.operation
    self.row = row or self.row
    self.trigger_name = trigger_name or self.trigger_name or f"{self.label}_{self.operation}"

    if self.condition is constants.UNSET:
        self.condition = pgtrigger.AnyChange() if self.operation == pgtrigger.Update else None

    if self.row not in ("OLD", "NEW"):  # pragma: no cover
        raise ValueError("Row must be specified as pghistory.Old or pghistory.New")
    elif self.operation == pgtrigger.Insert and self.row == "OLD":  # pragma: no cover
        raise ValueError('There is no "OLD" row on insert events')
    elif self.operation == pgtrigger.Delete and self.row == "NEW":  # pragma: no cover
        raise ValueError('There is no "NEW" row on delete events')

    if not self.operation:  # pragma: no cover
        raise ValueError("Must provide operation to RowEvent")

pghistory.Field

Field(
    *,
    primary_key: bool = constants.UNSET,
    unique: bool = constants.UNSET,
    blank: bool = constants.UNSET,
    null: bool = constants.UNSET,
    db_index: bool = constants.UNSET,
    editable: bool = constants.UNSET,
    unique_for_date: bool = constants.UNSET,
    unique_for_month: bool = constants.UNSET,
    unique_for_year: bool = constants.UNSET
)

Configuration for fields.

The default values for the attributes ensure that event models don't have unnecessary uniqueness constraints carried over from the tracked model.

Attributes:

Name	Type	Description
primary_key	bool, default=False	True if a primary key
unique	bool, default=False	True if unique
blank	bool	True if blank
null	bool	True if null
db_index	bool	True if indexed
editable	bool	True if editable
unique_for_date	bool	True if unique for date
unique_for_month	bool	True if unique for the month
unique_for_year	bool	True if unique for the year

Source code in pghistory/config.py

def __init__(
    self,
    *,
    primary_key: bool = constants.UNSET,
    unique: bool = constants.UNSET,
    blank: bool = constants.UNSET,
    null: bool = constants.UNSET,
    db_index: bool = constants.UNSET,
    editable: bool = constants.UNSET,
    unique_for_date: bool = constants.UNSET,
    unique_for_month: bool = constants.UNSET,
    unique_for_year: bool = constants.UNSET,
):
    self._kwargs = _get_kwargs(locals())
    self._finalized = False

pghistory.ForeignKey

ForeignKey(
    *,
    on_delete: Any = constants.UNSET,
    db_constraint: bool = constants.UNSET,
    **kwargs: Any
)

Bases: RelatedField

Configuration for foreign keys.

Arguments for RelatedField and Field can also be supplied. Note that db_index is overridden to True for all foreign keys

Attributes:

Name	Type	Description
on_delete	default=models.DO_NOTHING	Django's on_delete property
db_constraint	bool, default=False	True to use a datbase constraint for the foreign key

Source code in pghistory/config.py

def __init__(
    self,
    *,
    on_delete: Any = constants.UNSET,
    db_constraint: bool = constants.UNSET,
    **kwargs: Any,
):
    super().__init__(**kwargs)
    self._kwargs.update(_get_kwargs(locals()))

pghistory.InsertEvent

InsertEvent(
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None
)

Bases: RowEvent

Creates events based on inserts to a model

The default label used is "insert".

Source code in pghistory/core.py

def __init__(
    self,
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None,
):
    super().__init__(label=label)

    self.condition = condition or self.condition
    self.operation = operation or self.operation
    self.row = row or self.row
    self.trigger_name = trigger_name or self.trigger_name or f"{self.label}_{self.operation}"

    if self.condition is constants.UNSET:
        self.condition = pgtrigger.AnyChange() if self.operation == pgtrigger.Update else None

    if self.row not in ("OLD", "NEW"):  # pragma: no cover
        raise ValueError("Row must be specified as pghistory.Old or pghistory.New")
    elif self.operation == pgtrigger.Insert and self.row == "OLD":  # pragma: no cover
        raise ValueError('There is no "OLD" row on insert events')
    elif self.operation == pgtrigger.Delete and self.row == "NEW":  # pragma: no cover
        raise ValueError('There is no "NEW" row on delete events')

    if not self.operation:  # pragma: no cover
        raise ValueError("Must provide operation to RowEvent")

pghistory.ManualEvent

ManualEvent(label=None)

Bases: Tracker

For manually tracking an event.

Source code in pghistory/core.py

def __init__(self, label=None):
    self.label = label or self.label

    if not self.label:  # pragma: no cover
        raise ValueError("Must supply label attribute to event")

pghistory.ObjForeignKey

ObjForeignKey(
    *,
    related_name: str = constants.DEFAULT,
    related_query_name: str = constants.DEFAULT,
    **kwargs
)

Bases: ForeignKey

Configuration for the pgh_obj field

Attributes:

Name	Type	Description
related_name	str	The related_name to use
related_query_name	str	The related_query_name to use

Source code in pghistory/config.py

def __init__(
    self,
    *,
    related_name: str = constants.DEFAULT,
    related_query_name: str = constants.DEFAULT,
    **kwargs,
):
    # Note: We will be changing the default object field to nullable with on_delete=SET_NULL
    # in version 3. related_name will also default to "+"
    super().__init__(
        related_name=related_name, related_query_name=related_query_name, **kwargs
    )
    self._kwargs.update(_get_kwargs(locals()))

pghistory.RelatedField

RelatedField(
    *,
    related_name: str = constants.UNSET,
    related_query_name: str = constants.UNSET,
    **kwargs: Any
)

Bases: Field

Configuration for related fields.

By default, related names are stripped to avoid unnecessary clashes.

Note that all arguments from Field can also be supplied.

Attributes:

Name	Type	Description
related_name	str, default="+"	The related_name to use
related_query_name	str, default="+"	The related_query_name to use

Source code in pghistory/config.py

def __init__(
    self,
    *,
    related_name: str = constants.UNSET,
    related_query_name: str = constants.UNSET,
    **kwargs: Any,
):
    super().__init__(**kwargs)
    self._kwargs.update(_get_kwargs(locals()))

pghistory.RowEvent

RowEvent(
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None
)

Bases: Tracker

For tracking an event automatically based on row-level changes.

Source code in pghistory/core.py

def __init__(
    self,
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None,
):
    super().__init__(label=label)

    self.condition = condition or self.condition
    self.operation = operation or self.operation
    self.row = row or self.row
    self.trigger_name = trigger_name or self.trigger_name or f"{self.label}_{self.operation}"

    if self.condition is constants.UNSET:
        self.condition = pgtrigger.AnyChange() if self.operation == pgtrigger.Update else None

    if self.row not in ("OLD", "NEW"):  # pragma: no cover
        raise ValueError("Row must be specified as pghistory.Old or pghistory.New")
    elif self.operation == pgtrigger.Insert and self.row == "OLD":  # pragma: no cover
        raise ValueError('There is no "OLD" row on insert events')
    elif self.operation == pgtrigger.Delete and self.row == "NEW":  # pragma: no cover
        raise ValueError('There is no "NEW" row on delete events')

    if not self.operation:  # pragma: no cover
        raise ValueError("Must provide operation to RowEvent")

pghistory.Tracker

Tracker(label=None)

For tracking an event when a condition happens on a model.

Source code in pghistory/core.py

def __init__(self, label=None):
    self.label = label or self.label

    if not self.label:  # pragma: no cover
        raise ValueError("Must supply label attribute to event")

pghistory_setup

pghistory_setup(event_model)

Registers the tracker for the event model and calls user-defined setup

Source code in pghistory/core.py

def pghistory_setup(self, event_model):
    """Registers the tracker for the event model and calls user-defined setup"""
    tracked_model = event_model.pgh_tracked_model

    if _registered_trackers.get((tracked_model, self.label), event_model) != event_model:
        raise ValueError(
            f'Tracker with label "{self.label}" already exists for a different'
            f' event model of "{tracked_model._meta.label}". Supply a'
            " different label as the first argument to the tracker."
        )

    _registered_trackers[(tracked_model, self.label)] = event_model

    self.setup(event_model)

setup

setup(event_model)

Set up the tracker for the event model

Source code in pghistory/core.py

def setup(self, event_model):
    """Set up the tracker for the event model"""
    pass

pghistory.UpdateEvent

UpdateEvent(
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None
)

Bases: RowEvent

Creates events based on updates to a model.

By default,

• The label used is "update".
• Attributes from the "new" row of the update are stored.
• It only fires when fields of the event model are changed.

All of this behavior can be overridden by supplying a label, a condition, or the row to snapshot.

Source code in pghistory/core.py

def __init__(
    self,
    label: str = None,
    *,
    condition: Union[pgtrigger.Condition, None] = constants.UNSET,
    operation: pgtrigger.Operation = None,
    row: str = None,
    trigger_name: str = None,
):
    super().__init__(label=label)

    self.condition = condition or self.condition
    self.operation = operation or self.operation
    self.row = row or self.row
    self.trigger_name = trigger_name or self.trigger_name or f"{self.label}_{self.operation}"

    if self.condition is constants.UNSET:
        self.condition = pgtrigger.AnyChange() if self.operation == pgtrigger.Update else None

    if self.row not in ("OLD", "NEW"):  # pragma: no cover
        raise ValueError("Row must be specified as pghistory.Old or pghistory.New")
    elif self.operation == pgtrigger.Insert and self.row == "OLD":  # pragma: no cover
        raise ValueError('There is no "OLD" row on insert events')
    elif self.operation == pgtrigger.Delete and self.row == "NEW":  # pragma: no cover
        raise ValueError('There is no "NEW" row on delete events')

    if not self.operation:  # pragma: no cover
        raise ValueError("Must provide operation to RowEvent")

pghistory.context

context(**metadata: Any)

Bases: ContextDecorator

A context manager that groups changes under the same context and adds additional metadata about the event.

Context is added as variables at the beginning of every SQL statement. By default, all variables are localized to the transaction (i.e SET LOCAL), meaning they will only persist for the statement/transaction and not across the session.

Once any code has entered pghistory.context, all subsequent entrances of pghistory.context will be grouped under the same context until the top-most parent exits.

To add context only if a parent has already entered pghistory.context, one can call pghistory.context as a function without entering it. The metadata set in the function call will be part of the context if pghistory.context has previously been entered. Otherwise it will be ignored.

Attributes:

Name	Type	Description
**metadata		Metadata that should be attached to the tracking context

Example

Here we track a "key" with a value of "value":

with pghistory.context(key='value'):
    # Do things..
    # All tracked events will have the same `pgh_context`
    # foreign key, and the context object will include
    # {'key': 'value'} in its metadata.
    # Nesting the tracker adds additional metadata to the current
    # context

# Add metadata if a parent piece of code has already entered
# pghistory.context
pghistory.context(key='value')

Note

Context tracking is compatible for most scenarios, but it currently does not work for named cursors. Django uses named cursors for the .iterator() operator, which has no effect on history tracking. However, there may be other usages of named cursors in Django where history context is ignored.

Source code in pghistory/runtime.py

def __init__(self, **metadata: Any):
    self.metadata = metadata
    self._pre_execute_hook = None

    if hasattr(_tracker, "value"):
        _tracker.value.metadata.update(**self.metadata)

pghistory.ProxyField

ProxyField(proxy: str, field: Type[models.Field])

Proxies a JSON field from a model and adds it as a field in the queryset.

Parameters:

Name	Type	Description	Default
proxy	str	The value to proxy, e.g. "user__email"	required
field	Type[Field]	The field that will be used to cast the resulting value	required

Source code in pghistory/core.py

def ProxyField(proxy: str, field: Type[models.Field]):
    """
    Proxies a JSON field from a model and adds it as a field in the queryset.

    Args:
        proxy: The value to proxy, e.g. "user__email"
        field: The field that will be used to cast the resulting value
    """
    if not isinstance(field, models.Field):  # pragma: no cover
        raise TypeError(f'"{field}" is not a Django model Field instace')

    field.pgh_proxy = proxy
    return field

pghistory.create_event

create_event(
    obj: models.Model, *, label: str, using: str = "default"
) -> models.Model

Manually create a event for an object.

Events are automatically linked with any context being tracked via pghistory.context.

Parameters:

Name	Type	Description	Default
obj	Model	An instance of a model.	required
label	str	The event label.	required
using	str	The database	'default'

Raises:

Type	Description
ValueError	If the event label has not been registered for the model.

Returns:

Type	Description
Model	The created event model object

Source code in pghistory/core.py

def create_event(obj: models.Model, *, label: str, using: str = "default") -> models.Model:
    """Manually create a event for an object.

    Events are automatically linked with any context being tracked
    via [pghistory.context][].

    Args:
        obj: An instance of a model.
        label: The event label.
        using: The database

    Raises:
        ValueError: If the event label has not been registered for the model.

    Returns:
        The created event model object
    """
    # Verify that the provided label is tracked
    if (obj.__class__, label) not in _registered_trackers:
        raise ValueError(
            f'"{label}" is not a registered tracker label for model {obj._meta.object_name}.'
        )

    event_model = _registered_trackers[(obj.__class__, label)]
    event_model_kwargs = {
        "pgh_label": label,
        **{
            field.attname: getattr(obj, field.attname)
            for field in event_model._meta.fields
            if not field.name.startswith("pgh_")
        },
    }
    if hasattr(event_model, "pgh_obj"):
        event_model_kwargs["pgh_obj"] = obj

    event_obj = event_model(**event_model_kwargs)

    # The event model is inserted manually with a custom SQL compiler
    # that attaches the context using the _pgh_attach_context
    # stored procedure. Django does not allow one to use F()
    # objects to reference stored procedures, so we have to
    # inject it with a custom SQL compiler here.
    query = sql.InsertQuery(event_model)
    query.insert_values(
        [field for field in event_model._meta.fields if not isinstance(field, models.AutoField)],
        [event_obj],
    )

    if utils.psycopg_maj_version == 3:
        connections[using].connection.adapters.register_dumper(Literal, LiteralDumper)

    vals = _InsertEventCompiler(query, connections[using], using=using).execute_sql(
        event_model._meta.fields
    )

    # Django <= 2.2 does not support returning fields from a bulk create,
    # which requires us to fetch fields again to populate the context
    if isinstance(vals, int):  # pragma: no cover
        return event_model.objects.get(pgh_id=vals)
    else:
        # Django >= 3.1 returns the values as a list of one element
        if isinstance(vals, list) and len(vals) == 1:  # pragma: no branch
            vals = vals[0]

        for field, val in zip(event_model._meta.fields, vals):
            setattr(event_obj, field.attname, val)

        return event_obj

pghistory.create_event_model

create_event_model(
    tracked_model: Type[models.Model],
    *trackers: Tracker,
    fields: Union[List[str], None] = None,
    exclude: Union[List[str], None] = None,
    obj_field: ObjForeignKey = constants.UNSET,
    context_field: Union[ContextForeignKey, ContextJSONField] = constants.UNSET,
    context_id_field: ContextUUIDField = constants.UNSET,
    append_only: bool = constants.UNSET,
    model_name: Union[str, None] = None,
    app_label: Union[str, None] = None,
    base_model: Type[models.Model] = None,
    attrs: Dict[str, Any] = None,
    meta: Dict[str, Any] = None,
    abstract: bool = True
) -> Type[models.Model]

Create an event model.

Instead of using pghistory.track, which dynamically generates an event model, one can instead construct a event model themselves, which will also set up event tracking for the tracked model.

Parameters:

Name	Type	Description	Default
tracked_model	Type[Model]	The model that is being tracked.	required
*trackers	Tracker	The event trackers. When using any tracker that inherits pghistory.RowEvent, such as pghistory.InsertEvent, a Postgres trigger will be installed that automatically stores the event into the generated event model. Trackers that do not inherit pghistory.RowEvent must be manually created. If no events are supplied, defaults to pghistory.InsertEvent and pghistory.UpdateEvent.	()
fields	Union[List[str], None]	The list of fields to snapshot when the event takes place. When no fields are provided, the entire model is snapshot when the event happens. Note that snapshotting of the OLD or NEW row is configured by the snapshot attribute of the DatabaseTracker object. Manual events must specify these fields during manual creation.	None
exclude	Union[List[str], None]	Instead of providing a list of fields to snapshot, a user can instead provide a list of fields to not snapshot.	None
obj_field	ObjForeignKey	The foreign key field configuration that references the tracked object. Defaults to an unconstrained non-nullable foreign key. Use None to create a event model with no reference to the tracked object.	UNSET
context_field	Union[ContextForeignKey, ContextJSONField]	The context field configuration. Defaults to a nullable unconstrained foreign key. Use None to avoid attaching historical context altogether.	UNSET
context_id_field	ContextUUIDField	The context ID field configuration when using a ContextJSONField for the context_field. When using a denormalized context field, the ID field is used to track the UUID of the context. Use None to avoid using this field for denormalized context.	UNSET
append_only	bool	True if the event model is protected against updates and deletes.	UNSET
model_name	Union[str, None]	Use a custom model name when the event model is generated. Otherwise a default name based on the tracked model and fields will be created.	None
app_label	Union[str, None]	The app_label for the generated event model. Defaults to the app_label of the tracked model. Note, when tracking a Django model (User) or a model of a third-party app, one must manually specify the app_label of an internal app to use so that migrations work properly.	None
base_model	Type[Model]	The base model for the event model. Must inherit pghistory.models.Event.	None
attrs	Dict[str, Any]	Additional attributes to add to the event model	None
meta	Dict[str, Any]	Additional attributes to add to the Meta class of the event model.	None
abstract	bool	True if the generated model should be an abstract model.	True

Returns:

Type	Description
Type[Model]	The event model class.

Example

Create a custom event model:

class MyEventModel(create_event_model(
    TrackedModel,
    pghistory.InsertEvent(),
)):
    # Add custom indices or change default field declarations...

Source code in pghistory/core.py

def create_event_model(
    tracked_model: Type[models.Model],
    *trackers: Tracker,
    fields: Union[List[str], None] = None,
    exclude: Union[List[str], None] = None,
    obj_field: "ObjForeignKey" = constants.UNSET,
    context_field: Union["ContextForeignKey", "ContextJSONField"] = constants.UNSET,
    context_id_field: "ContextUUIDField" = constants.UNSET,
    append_only: bool = constants.UNSET,
    model_name: Union[str, None] = None,
    app_label: Union[str, None] = None,
    base_model: Type[models.Model] = None,
    attrs: Dict[str, Any] = None,
    meta: Dict[str, Any] = None,
    abstract: bool = True,
) -> Type[models.Model]:
    """
    Create an event model.

    Instead of using [pghistory.track][], which dynamically generates an event
    model, one can instead construct a event model themselves, which
    will also set up event tracking for the tracked model.

    Args:
        tracked_model: The model that is being tracked.
        *trackers: The event trackers. When using any tracker that inherits
            [pghistory.RowEvent][], such as [pghistory.InsertEvent][], a
            Postgres trigger will be installed that automatically stores the event
            into the generated event model. Trackers that do not inherit
            [pghistory.RowEvent][] must be manually created. If no events are
            supplied, defaults to `pghistory.InsertEvent` and `pghistory.UpdateEvent`.
        fields: The list of fields to snapshot when the event takes place. When
            no fields are provided, the entire model is snapshot when the event
            happens. Note that snapshotting of the OLD or NEW row is configured
            by the `snapshot` attribute of the `DatabaseTracker` object. Manual
            events must specify these fields during manual creation.
        exclude: Instead of providing a list of fields to snapshot, a user can
            instead provide a list of fields to not snapshot.
        obj_field: The foreign key field configuration that references the tracked object.
            Defaults to an unconstrained non-nullable foreign key. Use `None` to create a
            event model with no reference to the tracked object.
        context_field: The context field configuration. Defaults to a nullable
            unconstrained foreign key. Use `None` to avoid attaching historical context altogether.
        context_id_field: The context ID field configuration when using a ContextJSONField
            for the context_field. When using a denormalized context field, the ID
            field is used to track the UUID of the context. Use `None` to avoid using this
            field for denormalized context.
        append_only: True if the event model is protected against updates and deletes.
        model_name: Use a custom model name when the event model is generated. Otherwise
            a default name based on the tracked model and fields will be created.
        app_label: The app_label for the generated event model. Defaults to the app_label
            of the tracked model. Note, when tracking a Django model (User) or a model
            of a third-party app, one must manually specify the app_label of an internal
            app to use so that migrations work properly.
        base_model: The base model for the event model. Must inherit pghistory.models.Event.
        attrs: Additional attributes to add to the event model
        meta: Additional attributes to add to the Meta class of the event model.
        abstract: `True` if the generated model should be an abstract model.

    Returns:
        The event model class.

    Example:
        Create a custom event model:

            class MyEventModel(create_event_model(
                TrackedModel,
                pghistory.InsertEvent(),
            )):
                # Add custom indices or change default field declarations...
    """  # noqa
    if not trackers:
        trackers = config.default_trackers() or [InsertEvent(), UpdateEvent()]

    event_model = import_string("pghistory.models.Event")
    base_model = base_model or config.base_model()
    assert issubclass(base_model, event_model)

    obj_field = _get_obj_field(
        obj_field=obj_field,
        tracked_model=tracked_model,
        base_model=base_model,
        fields=fields,
    )
    context_field = _get_context_field(context_field)
    context_id_field = _get_context_id_field(context_id_field)
    append_only = _get_append_only(append_only)

    model_name = model_name or _generate_event_model_name(base_model, tracked_model, fields)
    app_label = app_label or tracked_model._meta.app_label
    _validate_event_model_path(app_label=app_label, model_name=model_name, abstract=abstract)
    app = apps.app_configs[app_label]
    models_module = app.module.__name__ + ".models"

    attrs = attrs or {}
    attrs.update({"pgh_trackers": trackers})
    meta = meta or {}
    exclude = exclude or []
    fields = (
        fields
        if fields is not None
        else [f.name for f in tracked_model._meta.fields if f.name not in exclude]
    )

    if append_only:
        meta["triggers"] = [
            *meta.get("triggers", []),
            pgtrigger.Protect(name="append_only", operation=pgtrigger.Update | pgtrigger.Delete),
        ]

    class_attrs = {
        "__module__": models_module,
        "Meta": type("Meta", (), {"abstract": abstract, "app_label": app_label, **meta}),
        "pgh_tracked_model": tracked_model,
        **{field: _generate_history_field(tracked_model, field) for field in fields},
        **attrs,
    }

    if isinstance(context_field, utils.JSONField) and context_id_field:
        class_attrs["pgh_context_id"] = context_id_field

    if context_field:
        class_attrs["pgh_context"] = context_field

    if obj_field:
        class_attrs["pgh_obj"] = obj_field

    event_model = type(model_name, (base_model,), class_attrs)
    if not abstract:
        setattr(sys.modules[models_module], model_name, event_model)

    return event_model

pghistory.track

track(
    *trackers: Tracker,
    fields: Union[List[str], None] = None,
    exclude: Union[List[str], None] = None,
    obj_field: ObjForeignKey = constants.UNSET,
    context_field: Union[ContextForeignKey, ContextJSONField] = constants.UNSET,
    context_id_field: ContextUUIDField = constants.UNSET,
    append_only: bool = constants.UNSET,
    model_name: Union[str, None] = None,
    app_label: Union[str, None] = None,
    base_model: Type[models.Model] = None,
    attrs: Dict[str, Any] = None,
    meta: Dict[str, Any] = None
)

A decorator for tracking events for a model.

When using this decorator, an event model is dynamically generated that snapshots the entire model or supplied fields of the model based on the events supplied. The snapshot is accompanied with the label that identifies the event.

Parameters:

Name	Type	Description	Default
*trackers	Tracker	The event trackers. When using any tracker that inherits pghistory.RowEvent, such as pghistory.InsertEvent, a Postgres trigger will be installed that automatically stores the event into the generated event model. Trackers that do not inherit pghistory.RowEvent must be manually created. If no events are supplied, defaults to pghistory.InsertEvent and pghistory.UpdateEvent.	()
fields	Union[List[str], None]	The list of fields to snapshot when the event takes place. When no fields are provided, the entire model is snapshot when the event happens. Note that snapshotting of the OLD or NEW row is configured by the snapshot attribute of the DatabaseTracker object. Manual events must specify these fields during manual creation.	None
exclude	Union[List[str], None]	Instead of providing a list of fields to snapshot, a user can instead provide a list of fields to not snapshot.	None
obj_field	ObjForeignKey	The foreign key field configuration that references the tracked object. Defaults to an unconstrained non-nullable foreign key. Use None to create a event model with no reference to the tracked object.	UNSET
context_field	Union[ContextForeignKey, ContextJSONField]	The context field configuration. Defaults to a nullable unconstrained foreign key. Use None to avoid attaching historical context altogether.	UNSET
context_id_field	ContextUUIDField	The context ID field configuration when using a ContextJSONField for the context_field. When using a denormalized context field, the ID field is used to track the UUID of the context. Use None to avoid using this field for denormalized context.	UNSET
append_only	bool	True if the event model is protected against updates and deletes.	UNSET
model_name	Union[str, None]	Use a custom model name when the event model is generated. Otherwise a default name based on the tracked model and fields will be created.	None
app_label	Union[str, None]	The app_label for the generated event model. Defaults to the app_label of the tracked model. Note, when tracking a Django model (User) or a model of a third-party app, one must manually specify the app_label of an internal app to use so that migrations work properly.	None
base_model	Type[Model]	The base model for the event model. Must inherit pghistory.models.Event.	None
attrs	Dict[str, Any]	Additional attributes to add to the event model	None
meta	Dict[str, Any]	Additional attributes to add to the Meta class of the event model.	None

Source code in pghistory/core.py

def track(
    *trackers: Tracker,
    fields: Union[List[str], None] = None,
    exclude: Union[List[str], None] = None,
    obj_field: "ObjForeignKey" = constants.UNSET,
    context_field: Union["ContextForeignKey", "ContextJSONField"] = constants.UNSET,
    context_id_field: "ContextUUIDField" = constants.UNSET,
    append_only: bool = constants.UNSET,
    model_name: Union[str, None] = None,
    app_label: Union[str, None] = None,
    base_model: Type[models.Model] = None,
    attrs: Dict[str, Any] = None,
    meta: Dict[str, Any] = None,
):
    """
    A decorator for tracking events for a model.

    When using this decorator, an event model is dynamically generated
    that snapshots the entire model or supplied fields of the model
    based on the `events` supplied. The snapshot is accompanied with
    the label that identifies the event.

    Args:
        *trackers: The event trackers. When using any tracker that inherits
            [pghistory.RowEvent][], such as [pghistory.InsertEvent][], a
            Postgres trigger will be installed that automatically stores the event
            into the generated event model. Trackers that do not inherit
            [pghistory.RowEvent][] must be manually created. If no events are
            supplied, defaults to `pghistory.InsertEvent` and `pghistory.UpdateEvent`.
        fields: The list of fields to snapshot when the event takes place. When no fields
            are provided, the entire model is snapshot when the event happens. Note that
            snapshotting of the OLD or NEW row is configured by the `snapshot`
            attribute of the `DatabaseTracker` object. Manual events must specify
            these fields during manual creation.
        exclude: Instead of providing a list of fields to snapshot, a user can instead
            provide a list of fields to not snapshot.
        obj_field: The foreign key field configuration that references the tracked object.
            Defaults to an unconstrained non-nullable foreign key. Use `None` to create a
            event model with no reference to the tracked object.
        context_field: The context field configuration. Defaults to a nullable unconstrained
            foreign key. Use `None` to avoid attaching historical context altogether.
        context_id_field: The context ID field configuration when using a ContextJSONField for
            the context_field. When using a denormalized context field, the ID field is used to
            track the UUID of the context. Use `None` to avoid using this field for denormalized
            context.
        append_only: True if the event model is protected against updates and deletes.
        model_name: Use a custom model name when the event model is generated. Otherwise a default
            name based on the tracked model and fields will be created.
        app_label: The app_label for the generated event model. Defaults to the app_label of the
            tracked model. Note, when tracking a Django model (User) or a model of a third-party
            app, one must manually specify the app_label of an internal app to
            use so that migrations work properly.
        base_model: The base model for the event model. Must inherit `pghistory.models.Event`.
        attrs: Additional attributes to add to the event model
        meta: Additional attributes to add to the Meta class of the event model.
    """

    def _model_wrapper(model_class):
        create_event_model(
            model_class,
            *trackers,
            fields=fields,
            exclude=exclude,
            obj_field=obj_field,
            context_field=context_field,
            context_id_field=context_id_field,
            append_only=append_only,
            model_name=model_name,
            app_label=app_label,
            abstract=False,
            base_model=base_model,
            attrs=attrs,
            meta=meta,
        )

        return model_class

    return _model_wrapper

pghistory.admin

pghistory.admin.EventModelAdmin

Bases: BaseEventAdmin

The base admin for event models

pghistory.admin.EventsAdmin

Bases: BaseEventAdmin

The admin for showing events across all event models

pghistory.middleware

pghistory.middleware.WSGIRequest

Bases: WSGIRequest

Although Django's auth middleware sets the user in middleware, apps like django-rest-framework set the user in the view layer. This creates issues for pghistory tracking since the context needs to be set before DB operations happen.

This special WSGIRequest updates pghistory context when the request.user attribute is updated.

pghistory.middleware.HistoryMiddleware

HistoryMiddleware(get_response)

Annotates the user/url in the pghistory context.

Source code in pghistory/middleware.py

def HistoryMiddleware(get_response):
    """
    Annotates the user/url in the pghistory context.
    """

    def middleware(request):
        if request.method in config.middleware_methods():
            user = (
                request.user.pk
                if hasattr(request, "user") and hasattr(request.user, "pk")
                else None
            )
            with pghistory.context(user=user, url=request.path):
                if isinstance(request, DjangoWSGIRequest):  # pragma: no branch
                    request.__class__ = WSGIRequest

                return get_response(request)
        else:
            return get_response(request)

    return middleware

pghistory.models

pghistory.models.Context

Bases: Model

install_pgh_attach_context_func classmethod

install_pgh_attach_context_func(using=DEFAULT_DB_ALIAS)

Installs a custom store procedure for upserting context for historical events. The upsert is aware of when tracking is enabled in the app (i.e. using pghistory.context())

This stored procedure is automatically installed in pghistory migration 0004.

Source code in pghistory/models.py

@classmethod
def install_pgh_attach_context_func(cls, using=DEFAULT_DB_ALIAS):
    """
    Installs a custom store procedure for upserting context
    for historical events. The upsert is aware of when tracking is
    enabled in the app (i.e. using pghistory.context())

    This stored procedure is automatically installed in pghistory migration 0004.
    """
    with connections[using].cursor() as cursor:
        cursor.execute(
            f"""
            CREATE OR REPLACE FUNCTION _pgh_attach_context()
            RETURNS {cls._meta.db_table}.id%TYPE AS $$
                DECLARE
                    _pgh_context_id UUID;
                    _pgh_context_metadata JSONB;
                BEGIN
                    BEGIN
                        SELECT INTO _pgh_context_id
                            CURRENT_SETTING('pghistory.context_id');
                        SELECT INTO _pgh_context_metadata
                            CURRENT_SETTING('pghistory.context_metadata');
                        EXCEPTION WHEN OTHERS THEN
                    END;
                    IF _pgh_context_id IS NOT NULL AND _pgh_context_metadata IS NOT NULL THEN
                        INSERT INTO {cls._meta.db_table} (id, metadata, created_at, updated_at)
                            VALUES (_pgh_context_id, _pgh_context_metadata, NOW(), NOW())
                            ON CONFLICT (id) DO UPDATE
                                SET metadata = EXCLUDED.metadata,
                                    updated_at = EXCLUDED.updated_at;
                        RETURN _pgh_context_id;
                    ELSE
                        RETURN NULL;
                    END IF;
                END;
            $$ LANGUAGE plpgsql;
            """
        )

pghistory.models.Event

Bases: Model

An abstract model for base elements of a event

can_revert property

can_revert

True if the event model can revert the tracked model

check classmethod

check(**kwargs)

Allow proxy models to inherit this model and define their own fields that are dynamically pulled from context

Source code in pghistory/models.py

@classmethod
def check(cls, **kwargs):
    """
    Allow proxy models to inherit this model and define their own fields
    that are dynamically pulled from context
    """
    errors = super().check(**kwargs)

    # If all local fields are proxied, ignored error E017 and allow the fields to be declared
    if any(error for error in errors if error.id == "models.E017") and all(
        hasattr(field, "pgh_proxy") for field in cls._meta.local_fields
    ):
        return [error for error in errors if error.id != "models.E017"]
    else:
        return errors

pghistory_setup classmethod

pghistory_setup()

Called when the class is prepared (see apps.py) to finalize setup of the model and register triggers

Source code in pghistory/models.py

@classmethod
def pghistory_setup(cls):
    """
    Called when the class is prepared (see apps.py)
    to finalize setup of the model and register triggers
    """
    if (
        not cls._meta.abstract and cls._meta.managed and not cls._meta.proxy
    ):  # pragma: no branch
        for tracker in cls.pgh_trackers or []:
            tracker.pghistory_setup(cls)

        # Set up the event model utility properties. Don't overwrite any
        # existing attributes
        if not hasattr(cls.pgh_tracked_model, "pgh_event_models"):
            cls.pgh_tracked_model.pgh_event_models = {}

        # Use dir here, otherwise the property is evaluated
        if "pgh_event_model" not in dir(cls.pgh_tracked_model):
            cls.pgh_tracked_model.pgh_event_model = PghEventModel()

        if isinstance(cls.pgh_tracked_model.pgh_event_models, dict):  # pragma: no branch
            cls.pgh_tracked_model.pgh_event_models.update(
                {tracker.label: cls for tracker in cls.pgh_trackers or []}
            )

revert

revert(using=DEFAULT_DB_ALIAS)

Reverts the tracked model based on the event fields.

Raises a RuntimeError if the event model doesn't track all fields

Source code in pghistory/models.py

def revert(self, using=DEFAULT_DB_ALIAS):
    """
    Reverts the tracked model based on the event fields.

    Raises a RuntimeError if the event model doesn't track all fields
    """
    if not self.can_revert:
        raise RuntimeError(
            f'Event model "{self.__class__.__name__}" cannot revert'
            f' "{self.pgh_tracked_model.__class__.__name__}" because it'
            " doesn't track every field."
        )

    qset = models.QuerySet(model=self.pgh_tracked_model, using=using)

    pk = getattr(self, self.pgh_tracked_model._meta.pk.name)
    return qset.update_or_create(
        pk=pk,
        defaults={
            field.name: getattr(self, field.name)
            for field in self.pgh_tracked_model._meta.fields
            if field != self.pgh_tracked_model._meta.pk
        },
    )[0]

pghistory.models.EventQuery

Bases: Query

A query over an event CTE when proxy fields are used

get_compiler

get_compiler(*args, **kwargs)

Overrides the Query method get_compiler in order to return an EventQueryCompiler.

Source code in pghistory/models.py

def get_compiler(self, *args, **kwargs):
    """
    Overrides the Query method get_compiler in order to return
    an EventQueryCompiler.
    """
    compiler = super().get_compiler(*args, **kwargs)
    compiler.__class__ = EventQueryCompiler
    return compiler

pghistory.models.EventQueryCompiler

Bases: SQLCompiler

as_sql

as_sql(*args, **kwargs)

If there are proxied fields on the event model, return a select from a CTE. Otherwise don't do anything special

Source code in pghistory/models.py

def as_sql(self, *args, **kwargs):
    """
    If there are proxied fields on the event model, return a select from a CTE.
    Otherwise don't do anything special
    """
    sql, params = super().as_sql(*args, **kwargs)

    if any(self.proxy_fields):
        if django.VERSION < (3, 2):  # pragma: no cover
            raise RuntimeError("Must use Django 3.2 or above to proxy fields on event models")

        cte = self._get_cte()
        sql = cte + sql.replace(f'"{self.query.model._meta.db_table}"', '"pgh_event_cte"')

    return sql, params

pghistory.models.EventQuerySet

EventQuerySet(model=None, query=None, using=None, hints=None)

Bases: QuerySet

QuerySet with support for proxy fields

Source code in pghistory/models.py

def __init__(self, model=None, query=None, using=None, hints=None):
    if query is None:
        query = EventQuery(model)

    super().__init__(model, query, using, hints)

pghistory.models.Events

Bases: Model

A proxy model for aggregating events together across tables and rendering diffs

check classmethod

check(**kwargs)

Allow proxy models to inherit this model and define their own fields that are dynamically pulled from context

Source code in pghistory/models.py

@classmethod
def check(cls, **kwargs):
    """
    Allow proxy models to inherit this model and define their own fields
    that are dynamically pulled from context
    """
    errors = super().check(**kwargs)
    return [error for error in errors if error.id != "models.E017"]

pghistory.models.EventsQuery

EventsQuery(*args, **kwargs)

Bases: Query

A query over an aggregate event CTE

Source code in pghistory/models.py

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.references = []
    self.tracks = []
    self.across = []

pghistory.models.EventsQueryCompiler

Bases: SQLCompiler

pghistory.models.EventsQuerySet

EventsQuerySet(model=None, query=None, using=None, hints=None)

Bases: QuerySet

QuerySet with support for Common Table Expressions

Source code in pghistory/models.py

def __init__(self, model=None, query=None, using=None, hints=None):
    # Only create an instance of a Query if this is the first invocation in
    # a query chain.
    if query is None:
        query = EventsQuery(model)

    super().__init__(model, query, using, hints)

across

across(*event_models)

Aggregates events across the provided event models

Source code in pghistory/models.py

def across(self, *event_models):
    """Aggregates events across the provided event models"""
    qs = self._clone()
    qs.query.across = [
        apps.get_model(model) if isinstance(model, str) else model for model in event_models
    ]
    return qs

references

references(*objs)

Query any rows that reference the objs.

If, for example, a foreign key or pgh_obj field points to the object, it will be aggregated.

Source code in pghistory/models.py

def references(self, *objs):
    """Query any rows that reference the objs.

    If, for example, a foreign key or pgh_obj field points to the
    object, it will be aggregated.
    """
    assert len(objs) >= 1

    if isinstance(objs[0], (list, tuple, models.QuerySet)):
        assert len(objs) == 1
        objs = objs[0]

    qs = self._clone()
    qs.query.references = objs
    return qs

tracks

tracks(*objs)

Query any rows with pgh_obj equal to the objs.

Source code in pghistory/models.py

def tracks(self, *objs):
    """Query any rows with pgh_obj equal to the objs."""
    assert len(objs) >= 1

    if isinstance(objs[0], (list, tuple, models.QuerySet)):
        assert len(objs) == 1
        objs = objs[0]

    qs = self._clone()
    qs.query.tracks = objs
    return qs

pghistory.models.MiddlewareEvents

Bases: Events

A proxy model for aggregating events. Includes additional fields that are captured by the pghistory middleware

pghistory.models.NoObjectsManager

Bases: Manager

Django's dumpdata and other commands will not work with Events models by default because of how they aggregate multiple tables based on objects.

We use this as the default manager for aggregate events so that dumpdata and other management commands still work with these models

pghistory.models.PghEventModel

A descriptor for accessing the pgh_event_model field on a tracked model