diff --git a/migrations/versions/4952e040e9dd_update_changes_table.py b/migrations/versions/4952e040e9dd_update_changes_table.py
new file mode 100644
index 0000000000..7cbfa987d2
--- /dev/null
+++ b/migrations/versions/4952e040e9dd_update_changes_table.py
@@ -0,0 +1,36 @@
+"""update_changes_table
+
+Revision ID: 4952e040e9dd
+Revises: e5c7a4e2df4d
+Create Date: 2020-04-06 13:22:15.256635
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '4952e040e9dd'
+down_revision = 'e5c7a4e2df4d'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute("DELETE FROM changes")
+    op.drop_column("changes", "object_version")
+    op.add_column("changes", sa.Column("org_id", sa.Integer(), nullable=False))
+    op.create_index(
+        "ix_changes_object_ref",
+        "changes",
+        ["object_type", "object_id"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_column("changes", "org_id")
+    op.add_column("changes", sa.Column("object_version", sa.Integer(), nullable=True, default=0))
+    op.execute("UPDATE changes SET object_version = 0")
+    op.alter_column("changes", "object_version", nullable=False)
+    op.drop_index("ix_changes_object_ref")
diff --git a/redash/handlers/alerts.py b/redash/handlers/alerts.py
index 7929df9c05..c05a07a87e 100644
--- a/redash/handlers/alerts.py
+++ b/redash/handlers/alerts.py
@@ -1,5 +1,3 @@
-import time
-
 from flask import request
 from funcy import project
 
@@ -12,7 +10,6 @@
     require_permission,
     view_only,
 )
-from redash.utils import json_dumps
 
 
 class AlertResource(BaseResource):
@@ -35,6 +32,7 @@ def post(self, alert_id):
         require_admin_or_owner(alert.user.id)
 
         self.update_model(alert, params)
+        alert.record_changes(self.current_user, models.Change.Type.Modified)
         models.db.session.commit()
 
         self.record_event(
@@ -48,6 +46,7 @@ def delete(self, alert_id):
             models.Alert.get_by_id_and_org, alert_id, self.current_org
         )
         require_admin_or_owner(alert.user_id)
+        alert.record_changes(self.current_user, models.Change.Type.Deleted)
         models.db.session.delete(alert)
         models.db.session.commit()
 
@@ -60,6 +59,7 @@ def post(self, alert_id):
         require_admin_or_owner(alert.user.id)
 
         alert.options["muted"] = True
+        alert.record_changes(self.current_user, models.Change.Type.Modified)
         models.db.session.commit()
 
         self.record_event(
@@ -73,6 +73,7 @@ def delete(self, alert_id):
         require_admin_or_owner(alert.user.id)
 
         alert.options["muted"] = False
+        alert.record_changes(self.current_user, models.Change.Type.Modified)
         models.db.session.commit()
 
         self.record_event(
@@ -97,7 +98,9 @@ def post(self):
         )
 
         models.db.session.add(alert)
-        models.db.session.flush()
+        models.db.session.commit()
+
+        alert.record_changes(self.current_user, models.Change.Type.Created)
         models.db.session.commit()
 
         self.record_event(
@@ -133,6 +136,9 @@ def post(self, alert_id):
         models.db.session.add(subscription)
         models.db.session.commit()
 
+        subscription.record_changes(self.current_user, models.Change.Type.Created)
+        models.db.session.commit()
+
         self.record_event(
             {
                 "action": "subscribe",
@@ -158,6 +164,7 @@ class AlertSubscriptionResource(BaseResource):
     def delete(self, alert_id, subscriber_id):
         subscription = models.AlertSubscription.query.get_or_404(subscriber_id)
         require_admin_or_owner(subscription.user.id)
+        subscription.record_changes(self.current_user, models.Change.Type.Deleted)
         models.db.session.delete(subscription)
         models.db.session.commit()
 
diff --git a/redash/handlers/api.py b/redash/handlers/api.py
index ef479ef3c3..264722b4ee 100644
--- a/redash/handlers/api.py
+++ b/redash/handlers/api.py
@@ -83,6 +83,13 @@
     VisualizationResource,
 )
 from redash.handlers.widgets import WidgetListResource, WidgetResource
+from redash.handlers.changes import (
+    ChangesListResource,
+    QueryChangesListResource,
+    DashboardChangesListResource,
+    AlertChangesListResource,
+    DestinationChangesListResource,
+)
 from redash.utils import json_dumps
 
 
@@ -313,3 +320,13 @@ def json_representation(data, code, headers=None):
 api.add_org_resource(
     OrganizationSettings, "/api/settings/organization", endpoint="organization_settings"
 )
+
+api.add_org_resource(ChangesListResource, "/api/changes", endpoint="organization_changes")
+api.add_org_resource(QueryChangesListResource, "/api/queries/<query_id>/changes", endpoint="query_changes")
+api.add_org_resource(
+    DashboardChangesListResource, "/api/dashboards/<dashboard_id>/changes", endpoint="dashboard_changes"
+)
+api.add_org_resource(AlertChangesListResource, "/api/alerts/<alert_id>/changes", endpoint="alert_changes")
+api.add_org_resource(
+    DestinationChangesListResource, "/api/destinations/<destination_id>/changes", endpoint="destination_changes"
+)
diff --git a/redash/handlers/changes.py b/redash/handlers/changes.py
new file mode 100644
index 0000000000..3d46ecd9f5
--- /dev/null
+++ b/redash/handlers/changes.py
@@ -0,0 +1,85 @@
+from flask import jsonify, request, url_for
+from funcy import partial
+from redash.handlers.base import (
+    BaseResource,
+    paginate,
+    order_results as _order_results,
+    get_object_or_404
+)
+from redash.models import Change, Query, Dashboard, Alert, NotificationDestination
+
+
+order_map = {
+    "created_at": "created_at",
+    "-created_at": "-created_at",
+    "object_type": "object_type",
+    "-object_type": "-object_type",
+    "object_id": "object_id",
+    "-object_id": "-object_id",
+    "user_id": "user_id",
+    "-user_id": "-user_id",
+}
+
+order_results = partial(
+    _order_results, default_order="-created_at", allowed_orders=order_map
+)
+
+
+class BaseChangesListResource(BaseResource):
+    def _prepare_response(self, results_set):
+        """
+        :qparam number page_size: Number of changes to return per page
+        :qparam number page: Page number to retrieve
+        :qparam number order: Name of column to order by
+
+        Responds with an array of `Change` objects.
+        """
+
+        page = request.args.get("page", 1, type=int)
+        page_size = request.args.get("page_size", 25, type=int)
+
+        ordered_results = order_results(results_set)
+
+        response = paginate(
+            ordered_results,
+            page=page,
+            page_size=page_size,
+            serializer=lambda c: c.to_dict(),
+            with_stats=True,
+            with_last_modified_by=False,
+        )
+
+        self.record_event({"action": "list", "object_type": "change"})
+
+        return response
+
+    def _get_by_object(self, fn, *args, **kwargs):
+        obj = get_object_or_404(fn, *args, **kwargs)
+        changes_set = Change.get_by_object(obj, self.current_org)
+        return self._prepare_response(changes_set)
+
+
+class ChangesListResource(BaseChangesListResource):
+    def get(self):
+        changes_set = Change.all_changes(self.current_org)
+        return self._prepare_response(changes_set)
+
+
+class QueryChangesListResource(BaseChangesListResource):
+    def get(self, query_id):
+        return self._get_by_object(Query.get_by_id_and_org, query_id, self.current_org)
+
+
+class DashboardChangesListResource(BaseChangesListResource):
+    def get(self, dashboard_id):
+        return self._get_by_object(Dashboard.get_by_id_and_org, dashboard_id, self.current_org)
+
+
+class AlertChangesListResource(BaseChangesListResource):
+    def get(self, alert_id):
+        return self._get_by_object(Alert.get_by_id_and_org, alert_id, self.current_org)
+
+
+class DestinationChangesListResource(BaseChangesListResource):
+    def get(self, destination_id):
+        return self._get_by_object(NotificationDestination.get_by_id_and_org, destination_id, self.current_org)
diff --git a/redash/handlers/dashboards.py b/redash/handlers/dashboards.py
index c5fceab7c1..e3490c4ebc 100644
--- a/redash/handlers/dashboards.py
+++ b/redash/handlers/dashboards.py
@@ -107,6 +107,10 @@ def post(self):
         )
         models.db.session.add(dashboard)
         models.db.session.commit()
+
+        dashboard.record_changes(self.current_user, models.Change.Type.Created)
+        models.db.session.commit()
+
         return serialize_dashboard(dashboard)
 
 
@@ -210,9 +214,9 @@ def post(self, dashboard_slug):
 
         updates["changed_by"] = self.current_user
 
-        self.update_model(dashboard, updates)
-        models.db.session.add(dashboard)
         try:
+            self.update_model(dashboard, updates)
+            dashboard.record_changes(self.current_user)
             models.db.session.commit()
         except StaleDataError:
             abort(409)
@@ -240,16 +244,14 @@ def delete(self, dashboard_slug):
             dashboard_slug, self.current_org
         )
         dashboard.is_archived = True
-        dashboard.record_changes(changed_by=self.current_user)
-        models.db.session.add(dashboard)
-        d = serialize_dashboard(dashboard, with_widgets=True, user=self.current_user)
+        dashboard.record_changes(self.current_user)
         models.db.session.commit()
 
         self.record_event(
             {"action": "archive", "object_id": dashboard.id, "object_type": "dashboard"}
         )
 
-        return d
+        return serialize_dashboard(dashboard, with_widgets=True, user=self.current_user)
 
 
 class PublicDashboardResource(BaseResource):
@@ -283,7 +285,9 @@ def post(self, dashboard_id):
         dashboard = models.Dashboard.get_by_id_and_org(dashboard_id, self.current_org)
         require_admin_or_owner(dashboard.user_id)
         api_key = models.ApiKey.create_for_object(dashboard, self.current_user)
-        models.db.session.flush()
+        models.db.session.commit()
+
+        api_key.record_changes(self.current_user, models.Change.Type.Created)
         models.db.session.commit()
 
         public_url = url_for(
@@ -316,6 +320,7 @@ def delete(self, dashboard_id):
         if api_key:
             api_key.active = False
             models.db.session.add(api_key)
+            api_key.record_changes(self.current_user)
             models.db.session.commit()
 
         self.record_event(
diff --git a/redash/handlers/destinations.py b/redash/handlers/destinations.py
index e935ed84bb..bb337a46b9 100644
--- a/redash/handlers/destinations.py
+++ b/redash/handlers/destinations.py
@@ -51,6 +51,7 @@ def post(self, destination_id):
             destination.options.set_schema(schema)
             destination.options.update(req["options"])
             models.db.session.add(destination)
+            destination.record_changes(self.current_user, models.Change.Type.Modified)
             models.db.session.commit()
         except ValidationError:
             abort(400)
@@ -71,6 +72,7 @@ def delete(self, destination_id):
         destination = models.NotificationDestination.get_by_id_and_org(
             destination_id, self.current_org
         )
+        destination.record_changes(self.current_user, models.Change.Type.Deleted)
         models.db.session.delete(destination)
         models.db.session.commit()
 
@@ -131,6 +133,8 @@ def post(self):
         try:
             models.db.session.add(destination)
             models.db.session.commit()
+            destination.record_changes(self.current_user, models.Change.Type.Created)
+            models.db.session.commit()
         except IntegrityError as e:
             if "name" in str(e):
                 abort(
diff --git a/redash/handlers/queries.py b/redash/handlers/queries.py
index 129f7e3e6d..416f718712 100644
--- a/redash/handlers/queries.py
+++ b/redash/handlers/queries.py
@@ -259,6 +259,11 @@ def post(self):
         models.db.session.add(query)
         models.db.session.commit()
 
+        query.record_changes(self.current_user, models.Change.Type.Created)
+        for vis in query.visualizations:
+            vis.record_changes(self.current_user, models.Change.Type.Created)
+        models.db.session.commit()
+
         self.record_event(
             {"action": "create", "object_id": query.id, "object_type": "query"}
         )
@@ -376,6 +381,7 @@ def post(self, query_id):
 
         try:
             self.update_model(query, query_def)
+            query.record_changes(self.current_user)
             models.db.session.commit()
         except StaleDataError:
             abort(409)
@@ -416,7 +422,8 @@ def delete(self, query_id):
             models.Query.get_by_id_and_org, query_id, self.current_org
         )
         require_admin_or_owner(query.user_id)
-        query.archive(self.current_user)
+        query.archive()
+        query.record_changes(self.current_user)
         models.db.session.commit()
 
 
@@ -428,6 +435,7 @@ def post(self, query_id):
         )
         require_admin_or_owner(query.user_id)
         query.regenerate_api_key()
+        query.record_changes(self.current_user)
         models.db.session.commit()
 
         self.record_event(
@@ -459,6 +467,11 @@ def post(self, query_id):
         forked_query = query.fork(self.current_user)
         models.db.session.commit()
 
+        forked_query.record_changes(self.current_user, models.Change.Type.Created)
+        for vis in forked_query.visualizations:
+            vis.record_changes(self.current_user, models.Change.Type.Created)
+        models.db.session.commit()
+
         self.record_event(
             {"action": "fork", "object_id": query_id, "object_type": "query"}
         )
diff --git a/redash/handlers/visualizations.py b/redash/handlers/visualizations.py
index 1621ea50cd..e2f42a58b5 100644
--- a/redash/handlers/visualizations.py
+++ b/redash/handlers/visualizations.py
@@ -23,6 +23,10 @@ def post(self):
         vis = models.Visualization(**kwargs)
         models.db.session.add(vis)
         models.db.session.commit()
+
+        vis.record_changes(self.current_user, models.Change.Type.Created)
+        models.db.session.commit()
+
         return serialize_visualization(vis, with_query=False)
 
 
@@ -42,9 +46,10 @@ def post(self, visualization_id):
         kwargs.pop("query_id", None)
 
         self.update_model(vis, kwargs)
-        d = serialize_visualization(vis, with_query=False)
+        vis.record_changes(self.current_user)
         models.db.session.commit()
-        return d
+
+        return serialize_visualization(vis, with_query=False)
 
     @require_permission("edit_query")
     def delete(self, visualization_id):
@@ -59,5 +64,6 @@ def delete(self, visualization_id):
                 "object_type": "Visualization",
             }
         )
+        vis.record_changes(self.current_user, models.Change.Type.Deleted)
         models.db.session.delete(vis)
         models.db.session.commit()
diff --git a/redash/handlers/widgets.py b/redash/handlers/widgets.py
index 6907943405..491bc8a309 100644
--- a/redash/handlers/widgets.py
+++ b/redash/handlers/widgets.py
@@ -50,7 +50,9 @@ def post(self):
         models.db.session.add(widget)
         models.db.session.commit()
 
+        widget.record_changes(self.current_user, models.Change.Type.Created)
         models.db.session.commit()
+
         return serialize_widget(widget)
 
 
@@ -70,6 +72,7 @@ def post(self, widget_id):
         widget_properties = request.get_json(force=True)
         widget.text = widget_properties["text"]
         widget.options = json_dumps(widget_properties["options"])
+        widget.record_changes(self.current_user)
         models.db.session.commit()
         return serialize_widget(widget)
 
@@ -85,5 +88,6 @@ def delete(self, widget_id):
         self.record_event(
             {"action": "delete", "object_id": widget_id, "object_type": "widget"}
         )
+        widget.record_changes(self.current_user, models.Change.Type.Deleted)
         models.db.session.delete(widget)
         models.db.session.commit()
diff --git a/redash/models/__init__.py b/redash/models/__init__.py
index 70606056a5..a18a9d3560 100644
--- a/redash/models/__init__.py
+++ b/redash/models/__init__.py
@@ -42,7 +42,7 @@
 from redash.models.parameterized_query import ParameterizedQuery
 
 from .base import db, gfk_type, Column, GFKBase, SearchBaseQuery
-from .changes import ChangeTrackingMixin, Change  # noqa
+from .changes import Change, track_changes  # noqa
 from .mixins import BelongsToOrgMixin, TimestampMixin
 from .organizations import Organization
 from .types import (
@@ -451,7 +451,20 @@ def should_schedule_next(
     "schedule",
     "schedule_failures",
 )
-class Query(ChangeTrackingMixin, TimestampMixin, BelongsToOrgMixin, db.Model):
+@track_changes(
+    attributes=[
+        "name",
+        "description",
+        "query_text",
+        "is_archived",
+        "is_draft",
+        "api_key",
+        "options",
+        "schedule",
+        "tags",
+    ]
+)
+class Query(TimestampMixin, BelongsToOrgMixin, db.Model):
     id = Column(db.Integer, primary_key=True)
     version = Column(db.Integer, default=1)
     org_id = Column(db.Integer, db.ForeignKey("organizations.id"))
@@ -500,7 +513,7 @@ class Query(ChangeTrackingMixin, TimestampMixin, BelongsToOrgMixin, db.Model):
     def __str__(self):
         return str(self.id)
 
-    def archive(self, user=None):
+    def archive(self):
         db.session.add(self)
         self.is_archived = True
         self.schedule = None
@@ -512,8 +525,6 @@ def archive(self, user=None):
         for a in self.alerts:
             db.session.delete(a)
 
-        if user:
-            self.record_changes(user)
 
     def regenerate_api_key(self):
         self.api_key = generate_token(40)
@@ -940,6 +951,7 @@ def next_state(op, value, threshold):
 @generic_repr(
     "id", "name", "query_id", "user_id", "state", "last_triggered_at", "rearm"
 )
+@track_changes(attributes=("name", "query_id", "options", "rearm"))
 class Alert(TimestampMixin, BelongsToOrgMixin, db.Model):
     UNKNOWN_STATE = "unknown"
     OK_STATE = "ok"
@@ -1055,7 +1067,8 @@ def generate_slug(ctx):
 @generic_repr(
     "id", "name", "slug", "user_id", "org_id", "version", "is_archived", "is_draft"
 )
-class Dashboard(ChangeTrackingMixin, TimestampMixin, BelongsToOrgMixin, db.Model):
+@track_changes(attributes=["name", "is_archived", "is_draft", "tags", "dashboard_filters_enabled"])
+class Dashboard(TimestampMixin, BelongsToOrgMixin, db.Model):
     id = Column(db.Integer, primary_key=True)
     version = Column(db.Integer)
     org_id = Column(db.Integer, db.ForeignKey("organizations.id"))
@@ -1161,6 +1174,10 @@ def lowercase_name(cls):
 
 
 @generic_repr("id", "name", "type", "query_id")
+@track_changes(
+    parent=(Query, "query_id"),
+    attributes=["type", "name", "description", "options"]
+)
 class Visualization(TimestampMixin, BelongsToOrgMixin, db.Model):
     id = Column(db.Integer, primary_key=True)
     type = Column(db.String(100))
@@ -1178,7 +1195,7 @@ def __str__(self):
 
     @classmethod
     def get_by_id_and_org(cls, object_id, org):
-        return super(Visualization, cls).get_by_id_and_org(object_id, org, Query)
+        return super().get_by_id_and_org(object_id, org, Query)
 
     def copy(self):
         return {
@@ -1190,6 +1207,10 @@ def copy(self):
 
 
 @generic_repr("id", "visualization_id", "dashboard_id")
+@track_changes(
+    parent=(Dashboard, "dashboard_id"),
+    attributes=["text", "visualization_id", "options"]
+)
 class Widget(TimestampMixin, BelongsToOrgMixin, db.Model):
     id = Column(db.Integer, primary_key=True)
     visualization_id = Column(
@@ -1210,7 +1231,7 @@ def __str__(self):
 
     @classmethod
     def get_by_id_and_org(cls, object_id, org):
-        return super(Widget, cls).get_by_id_and_org(object_id, org, Dashboard)
+        return super().get_by_id_and_org(object_id, org, Dashboard)
 
 
 @generic_repr(
@@ -1275,6 +1296,10 @@ def record(cls, event):
 
 
 @generic_repr("id", "created_by_id", "org_id", "active")
+@track_changes(
+    parent=("object_type", "object_id"),
+    attributes=["api_key", "created_by_id", "active"]
+)
 class ApiKey(TimestampMixin, GFKBase, db.Model):
     id = Column(db.Integer, primary_key=True)
     org_id = Column(db.Integer, db.ForeignKey("organizations.id"))
@@ -1310,6 +1335,7 @@ def create_for_object(cls, object, user):
 
 
 @generic_repr("id", "name", "type", "user_id", "org_id", "created_at")
+@track_changes(attributes=("name", "type", "options"))
 class NotificationDestination(BelongsToOrgMixin, db.Model):
     id = Column(db.Integer, primary_key=True)
     org_id = Column(db.Integer, db.ForeignKey("organizations.id"))
@@ -1367,6 +1393,10 @@ def notify(self, alert, query, user, new_state, app, host):
 
 
 @generic_repr("id", "user_id", "destination_id", "alert_id")
+@track_changes(
+    parent=(Alert, "alert_id"),
+    attributes=("destination_id",)
+)
 class AlertSubscription(TimestampMixin, db.Model):
     id = Column(db.Integer, primary_key=True)
     user_id = Column(db.Integer, db.ForeignKey("users.id"))
diff --git a/redash/models/changes.py b/redash/models/changes.py
index a670703514..ea9146ec2f 100644
--- a/redash/models/changes.py
+++ b/redash/models/changes.py
@@ -1,15 +1,46 @@
 from sqlalchemy.inspection import inspect
 from sqlalchemy_utils.models import generic_repr
+from sqlalchemy.orm import object_session
+from sqlalchemy.ext.mutable import MutableBase
+
+from inspect import isclass
+from enum import Enum
+
+from redash.utils.configuration import ConfigurationContainer
 
 from .base import GFKBase, db, Column
 from .types import PseudoJSON
 
+# Structure of Change object
+# user_id
+#   reference to a user that modified object
+# created_at
+#   time when modification was made
+# object_type/object_id
+#   reference to modified object (e.g. query, dashboard)
+# changes.object_type/changes.object_id
+#   reference to a real target. When there are relation between entities, object_type/object_id contains
+#   a reference to "parent", and changes.object_type/changes.object_id contains a reference to "child".
+#   For example, when modifying a visualization, changes.object_type/changes.object_id will be the reference
+#   to that visualization, and object_type/object_id will be the reference to a query for this visualization.
+# changes.changes
+#   a dictionary where keys are names of modified fields and values are tuples with field values.
+#   Tuple may contain one value if field wasn't changes OR if value was mutated (MutableDict/List/Set),
+#   or two values - previous and new one. In both cases tuple is used to avoid situations when field
+#   contains an JSON array and may be wrongly interpreted.
+
 
 @generic_repr("id", "object_type", "object_id", "created_at")
 class Change(GFKBase, db.Model):
+    class Type(str, Enum):  # `str` to make it json-serializable
+        Created = "created"
+        Modified = "modified"
+        Deleted = "deleted"
+
     id = Column(db.Integer, primary_key=True)
     # 'object' defined in GFKBase
-    object_version = Column(db.Integer, default=0)
+    org_id = Column(db.Integer, db.ForeignKey("organizations.id"))
+    org = db.relationship("Organization", backref="changes")
     user_id = Column(db.Integer, db.ForeignKey("users.id"))
     user = db.relationship("User", backref="changes")
     change = Column(PseudoJSON)
@@ -17,77 +48,189 @@ class Change(GFKBase, db.Model):
 
     __tablename__ = "changes"
 
-    def to_dict(self, full=True):
-        d = {
+    def to_dict(self, extended=True):
+        result = {
             "id": self.id,
-            "object_id": self.object_id,
-            "object_type": self.object_type,
-            "change_type": self.change_type,
-            "object_version": self.object_version,
-            "change": self.change,
             "created_at": self.created_at,
+            # TODO: return an object (and target?) as a sub-structure, but carefully - they may be removed
+            "object_type": self.object_type,
+            "object_id": self.object_id,
+            "target_type": self.change["object_type"],
+            "target_id": self.change["object_id"],
+            "change_type": self.change["change_type"],
+            "changes": self.change["changes"],
         }
 
-        if full:
-            d["user"] = self.user.to_dict()
+        # TODO: user may be deleted - need to handle this
+        if extended:
+            result["user"] = self.user.to_dict()
         else:
-            d["user_id"] = self.user_id
+            result["user_id"] = self.user_id
 
-        return d
+        return result
 
     @classmethod
-    def last_change(cls, obj):
-        return (
-            cls.query.filter(
-                cls.object_id == obj.id, cls.object_type == obj.__class__.__tablename__
-            )
-            .order_by(cls.object_version.desc())
-            .first()
-        )
+    def all_changes(cls, org):
+        return cls.query.filter(cls.org == org).order_by(cls.id.desc())
 
+    @classmethod
+    def get_by_object(cls, target_object, org):
+        return cls.query.filter(
+            cls.org == org,
+            cls.object_type == target_object.__table__.name,
+            cls.object_id == target_object.id,
+        ).order_by(cls.id.desc())
+
+
+def _to_json(value):
+    if isinstance(value, ConfigurationContainer):
+        return value.to_dict()
+    return value
+
+
+def _detect_mutations(obj, attributes):
+    if not hasattr(obj, "__changes__"):
+        obj.__changes__ = {}
+
+    state = inspect(obj)
+    for key in attributes:
+        value = getattr(obj, key, None)
+        if isinstance(value, MutableBase):
+            attr = getattr(state.attrs, key, None)
+            if attr:
+                attr.load_history()
+                if attr.history and attr.history.has_changes():
+                    obj.__changes__[key] = {
+                        "previous": value,
+                        "current": value,
+                    }
+
+
+def _get_object_changes(obj, reset=True):
+    result = {}
+    changes = getattr(obj, "__changes__", None)
+
+    if changes:
+        columns = {}
+        for attr in inspect(obj.__class__).column_attrs:
+            col, = attr.columns
+            columns[attr.key] = col.name
 
-class ChangeTrackingMixin(object):
-    skipped_fields = ("id", "created_at", "updated_at", "version")
-    _clean_values = None
+        for key, change in changes.items():
+            if change["current"] != change["previous"]:
+                col = columns.get(key, key)
+                result[col] = (_to_json(change["previous"]), _to_json(change["current"]))
+            elif isinstance(change["current"], MutableBase):
+                col = columns.get(key, key)
+                result[col] = (_to_json(change["current"]),)
 
-    def __init__(self, *a, **kw):
-        super(ChangeTrackingMixin, self).__init__(*a, **kw)
-        self.record_changes(self.user)
+        if reset:
+            changes.clear()
 
-    def prep_cleanvalues(self):
-        self.__dict__["_clean_values"] = {}
-        for attr in inspect(self.__class__).column_attrs:
-            col, = attr.columns
-            # 'query' is col name but not attr name
-            self._clean_values[col.name] = None
+    return result if result else None
 
-    def __setattr__(self, key, value):
-        if self._clean_values is None:
-            self.prep_cleanvalues()
-        for attr in inspect(self.__class__).column_attrs:
-            col, = attr.columns
-            previous = getattr(self, attr.key, None)
-            self._clean_values[col.name] = previous
 
-        super(ChangeTrackingMixin, self).__setattr__(key, value)
+def _collect_all_attributes(obj, attributes, reset=True):
+    result = {}
+
+    columns = {}
+    for attr in inspect(obj.__class__).column_attrs:
+        col, = attr.columns
+        columns[attr.key] = col.name
+
+    for key in attributes:
+        col = columns.get(key, key)
+        value = getattr(obj, key, None)
+        result[col] = (_to_json(value),)
+
+    changes = getattr(obj, "__changes__", None)
+    if changes and reset:
+        changes.clear()
+
+    return result
+
+
+def _patch_setattr_method(cls, attributes):
+    original_setattr = cls.__setattr__
+
+    def new_setattr(self, key, value):
+        if key in attributes:
+            if not hasattr(self, "__changes__"):
+                self.__changes__ = {}
+            change = self.__changes__.get(key, {
+                "previous": getattr(self, key, None),
+                "current": None,
+            })
+            change["current"] = value
+            self.__changes__[key] = change
+
+        original_setattr(self, key, value)
+
+    cls.__setattr__ = new_setattr
+
+
+def _patch_clear_detected_changes_method(cls):
+    def clear_detected_changes(self):
+        changes = getattr(self, "__changes__", None)
+        if changes:
+            changes.clear()
+
+    cls.clear_detected_changes = clear_detected_changes
+
+
+def _patch_record_changes_method(cls, attributes, parent):
+    def record_changes(self, changed_by, change_type=Change.Type.Modified):
+        session = object_session(self)
+        if not session:
+            return
 
-    def record_changes(self, changed_by):
-        db.session.add(self)
-        db.session.flush()
         changes = {}
-        for attr in inspect(self.__class__).column_attrs:
-            col, = attr.columns
-            if attr.key not in self.skipped_fields:
-                changes[col.name] = {
-                    "previous": self._clean_values[col.name],
-                    "current": getattr(self, attr.key),
-                }
-
-        db.session.add(
-            Change(
-                object=self,
-                object_version=self.version,
-                user=changed_by,
-                change=changes,
-            )
+        if change_type == Change.Type.Created:
+            changes = _collect_all_attributes(self, attributes)
+        if change_type == Change.Type.Modified:
+            _detect_mutations(self, attributes)
+            changes = _get_object_changes(self)
+
+        if changes is None:
+            return
+
+        self_type = self.__table__.name
+        self_id = self.id
+
+        changes = Change(
+            object_type=self_type,
+            object_id=self_id,
+            org=changed_by.org,
+            user=changed_by,
+            change={
+                "object_type": self_type,
+                "object_id": self_id,
+                "change_type": change_type,
+                "changes": changes,
+            },
         )
+
+        if parent:
+            parent_type, parent_id = parent
+            if isclass(parent_type):  # SQLAlchemy model
+                changes.object_type = parent_type.__table__.name
+            else:
+                changes.object_type = getattr(self, parent_type, self_type)
+            changes.object_id = getattr(self, parent_id, self_id)
+
+        session.add(changes)
+
+    cls.record_changes = record_changes
+
+
+def track_changes(attributes, parent=None):
+    attributes = set(attributes) - {"id", "created_at", "updated_at", "version"}
+
+    # monkey-patch class because inheritance will break SQLAlchemy
+    def decorator(cls):
+        _patch_setattr_method(cls, attributes)
+        _patch_record_changes_method(cls, attributes, parent)
+        _patch_clear_detected_changes_method(cls)
+        return cls
+
+    return decorator
diff --git a/tests/handlers/test_changes.py b/tests/handlers/test_changes.py
new file mode 100644
index 0000000000..2d3ec30902
--- /dev/null
+++ b/tests/handlers/test_changes.py
@@ -0,0 +1,80 @@
+from tests import BaseTestCase
+from redash.models import db, Change
+
+
+class TestChangesGet(BaseTestCase):
+    def setUp(self):
+        super().setUp()
+
+        query = self.factory.create_query()
+        query.record_changes(self.factory.user, Change.Type.Created)
+        query.query_text = "SELECT 'test'"
+        query.record_changes(self.factory.user, Change.Type.Modified)
+        query.name = "Test Query"
+        query.record_changes(self.factory.user, Change.Type.Modified)
+        db.session.commit()
+
+        widget = self.factory.create_widget()
+        dashboard = widget.dashboard
+        dashboard.record_changes(self.factory.user, Change.Type.Created)
+        widget.record_changes(self.factory.user, Change.Type.Created)
+        dashboard.name = "Test Dashboard"
+        dashboard.record_changes(self.factory.user, Change.Type.Modified)
+        widget.visualization_id = None
+        widget.text = "Text of widget"
+        widget.record_changes(self.factory.user, Change.Type.Modified)
+        db.session.commit()
+
+        self._objects = (query, dashboard, widget)
+
+    def _assert_changes_structure(self, changes, expected_count=None):
+        if expected_count is not None:
+            self.assertEquals(len(changes), expected_count, "Expected %i items" % expected_count)
+        expected_keys = {
+            "created_at", "object_type", "object_id", "target_type", "target_id", "change_type", "changes",
+        }
+        for c in changes:
+            keys = set(c.keys())
+            self.assertTrue(expected_keys.issubset(keys), "Change structure mismatch")
+            self.assertIsInstance(c["changes"], dict, "Change structure mismatch")
+
+    def _assert_changes_ownership(self, changes, refs):
+        for c in changes:
+            ref = (c["object_type"], c["object_id"], c["target_type"], c["target_id"])
+            self.assertTrue(ref in refs, "Change does not belong to any of specified objects")
+
+    def test_get_all_changes(self):
+        query, dashboard, widget = self._objects
+        rv = self.make_request("get", "/api/changes")
+        self.assertEqual(rv.status_code, 200)
+
+        changes = rv.json["results"]
+        self._assert_changes_structure(changes, expected_count=7)
+        self._assert_changes_ownership(changes, {
+            (query.__table__.name, query.id, query.__table__.name, query.id),
+            (dashboard.__table__.name, dashboard.id, dashboard.__table__.name, dashboard.id),
+            (dashboard.__table__.name, dashboard.id, widget.__table__.name, widget.id),
+        })
+
+    def test_get_all_changes_for_object(self):
+        query, dashboard, widget = self._objects
+        rv = self.make_request("get", "/api/queries/{0}/changes".format(query.id))
+        self.assertEqual(rv.status_code, 200)
+
+        changes = rv.json["results"]
+        self._assert_changes_structure(changes, expected_count=3)
+        self._assert_changes_ownership(changes, {
+            (query.__table__.name, query.id, query.__table__.name, query.id),
+        })
+
+    def test_get_all_changes_for_object_with_relations(self):
+        query, dashboard, widget = self._objects
+        rv = self.make_request("get", "/api/dashboards/{0}/changes".format(dashboard.id))
+        self.assertEqual(rv.status_code, 200)
+
+        changes = rv.json["results"]
+        self._assert_changes_structure(changes, expected_count=4)
+        self._assert_changes_ownership(changes, {
+            (dashboard.__table__.name, dashboard.id, dashboard.__table__.name, dashboard.id),
+            (dashboard.__table__.name, dashboard.id, widget.__table__.name, widget.id),
+        })
diff --git a/tests/models/test_changes.py b/tests/models/test_changes.py
index e9847b0a0a..2e17d17ec1 100644
--- a/tests/models/test_changes.py
+++ b/tests/models/test_changes.py
@@ -1,87 +1,187 @@
 from tests import BaseTestCase
 
-from redash.models import db, Query, Change, ChangeTrackingMixin
-
-
-def create_object(factory):
-    obj = Query(
-        name="Query",
-        description="",
-        query_text="SELECT 1",
-        user=factory.user,
-        data_source=factory.data_source,
-        org=factory.org,
-    )
-
-    return obj
-
-
-class TestChangesProperty(BaseTestCase):
-    def test_returns_initial_state(self):
-        obj = create_object(self.factory)
-
-        for change in Change.query.filter(Change.object == obj):
-            self.assertIsNone(change.change["previous"])
-
-
-class TestLogChange(BaseTestCase):
-    def obj(self):
-        obj = Query(
-            name="Query",
-            description="",
-            query_text="SELECT 1",
-            user=self.factory.user,
-            data_source=self.factory.data_source,
-            org=self.factory.org,
-        )
-
-        return obj
-
-    def test_properly_logs_first_creation(self):
-        obj = create_object(self.factory)
-        obj.record_changes(changed_by=self.factory.user)
-        change = Change.last_change(obj)
-
-        self.assertIsNotNone(change)
-        self.assertEqual(change.object_version, 1)
-
-    def test_skips_unnecessary_fields(self):
-        obj = create_object(self.factory)
-        obj.record_changes(changed_by=self.factory.user)
-        change = Change.last_change(obj)
-
-        self.assertIsNotNone(change)
-        self.assertEqual(change.object_version, 1)
-        for field in ChangeTrackingMixin.skipped_fields:
-            self.assertNotIn(field, change.change)
-
-    def test_properly_log_modification(self):
-        obj = create_object(self.factory)
-        obj.record_changes(changed_by=self.factory.user)
-        obj.name = "Query 2"
-        obj.description = "description"
-        db.session.flush()
-        obj.record_changes(changed_by=self.factory.user)
-
-        change = Change.last_change(obj)
-
-        self.assertIsNotNone(change)
-        # TODO: https://github.com/getredash/redash/issues/1550
-        # self.assertEqual(change.object_version, 2)
-        self.assertEqual(change.object_version, obj.version)
-        self.assertIn("name", change.change)
-        self.assertIn("description", change.change)
-
-    def test_logs_create_method(self):
-        q = Query(
-            name="Query",
-            description="",
-            query_text="",
-            user=self.factory.user,
-            data_source=self.factory.data_source,
-            org=self.factory.org,
-        )
-        change = Change.last_change(q)
-
-        self.assertIsNotNone(change)
-        self.assertEqual(q.user, change.user)
+from redash.models import db, Change
+
+
+class BaseChangeTestCase(BaseTestCase):
+    # updates an object and returns changes in the same form as they're stored in DB
+    def _update_object(self, obj, new_values):
+        result = {}
+        for key, new_value in new_values.items():
+            old_value = getattr(obj, key, None)
+            setattr(obj, key, new_value)
+            if old_value == new_value:
+                result[key] = [new_value]
+            else:
+                result[key] = [old_value, new_value]
+        return result
+
+    def _record_changes(self, obj, change_type=Change.Type.Modified):
+        obj.record_changes(self.factory.user, change_type)
+        db.session.commit()
+
+    def _get_changes_by_object(self, obj, expected_count=None):
+        changes = list(Change.get_by_object(obj, self.factory.org))
+        if expected_count is not None:
+            self.assertEquals(len(changes), expected_count, "Expected %i change(s)" % expected_count)
+        return changes
+
+    def _assert_change_ownership(self, change, target, parent):
+        if parent:
+            self.assertEquals(change.object_type, parent.__table__.name, "Parent reference mismatch")
+            self.assertEquals(change.object_id, parent.id, "Parent reference mismatch")
+
+        if target:
+            self.assertEquals(change.change["object_type"], target.__table__.name, "Target reference mismatch")
+            self.assertEquals(change.change["object_id"], target.id, "Target reference mismatch")
+
+    def _test_creation(self, target, parent):
+        self._record_changes(target, Change.Type.Created)
+
+        changes = self._get_changes_by_object(parent, expected_count=1)
+        change = changes[0]
+        self._assert_change_ownership(change, target, parent)
+
+        self.assertTrue(change.change["changes"], "There should be some changes")
+
+    def _test_modification(self, target, parent, new_values):
+        target.clear_detected_changes()
+
+        updates = self._update_object(target, new_values)
+        self._record_changes(target, Change.Type.Modified)
+
+        changes = self._get_changes_by_object(parent, expected_count=1)
+        change = changes[0]
+        self._assert_change_ownership(change, target, parent)
+
+        # this will also check if only modified fields were stored to DB
+        self.assertDictEqual(change.change["changes"], updates)
+
+    def _test_deletion(self, target, parent):
+        target.clear_detected_changes()
+
+        self._record_changes(target, Change.Type.Deleted)
+
+        changes = self._get_changes_by_object(parent, expected_count=1)
+        change = changes[0]
+        self._assert_change_ownership(change, target, parent)
+
+        # when deleting an object, no changes are logged, just a fact of deletion
+        self.assertDictEqual(change.change["changes"], {})
+
+
+class TestQueryChanges(BaseChangeTestCase):
+    def test_creation(self):
+        query = self.factory.create_query()
+        self._test_creation(query, query)
+
+    def test_modification(self):
+        query = self.factory.create_query()
+        self._test_modification(query, query, {"name": "New Name"})
+
+    def test_deletion(self):
+        query = self.factory.create_query()
+        self._test_deletion(query, query)
+
+
+class TestVisualizationChanges(BaseChangeTestCase):
+    def test_creation(self):
+        visualization = self.factory.create_visualization()
+        self._test_creation(visualization, visualization.query_rel)
+
+    def test_modification(self):
+        visualization = self.factory.create_visualization()
+        self._test_modification(visualization, visualization.query_rel, {"name": "New Name"})
+
+    def test_deletion(self):
+        visualization = self.factory.create_visualization()
+        self._test_deletion(visualization, visualization.query_rel)
+
+
+class TestDashboardChanges(BaseChangeTestCase):
+    def test_creation(self):
+        dashboard = self.factory.create_dashboard()
+        self._test_creation(dashboard, dashboard)
+
+    def test_modification(self):
+        dashboard = self.factory.create_dashboard()
+        self._test_modification(dashboard, dashboard, {"name": "New Name"})
+
+    def test_deletion(self):
+        dashboard = self.factory.create_dashboard()
+        self._test_deletion(dashboard, dashboard)
+
+
+class TestWidgetChanges(BaseChangeTestCase):
+    def test_creation(self):
+        widget = self.factory.create_widget()
+        self._test_creation(widget, widget.dashboard)
+
+    def test_modification(self):
+        widget = self.factory.create_widget()
+        self._test_modification(widget, widget.dashboard, {
+            "visualization_id": None,
+            "text": "Text of widget"
+        })
+
+    def test_deletion(self):
+        widget = self.factory.create_widget()
+        self._test_deletion(widget, widget.dashboard)
+
+
+class TestApiKeyChanges(BaseChangeTestCase):
+    def test_creation(self):
+        dashboard = self.factory.create_dashboard()
+        api_key = self.factory.create_api_key(object=dashboard, active=True)
+        self._test_creation(api_key, dashboard)
+
+    def test_modification(self):
+        dashboard = self.factory.create_dashboard()
+        api_key = self.factory.create_api_key(object=dashboard, active=True)
+        self._test_modification(api_key, dashboard, {"active": False})
+
+    def test_deletion(self):
+        dashboard = self.factory.create_dashboard()
+        api_key = self.factory.create_api_key(object=dashboard, active=True)
+        self._test_deletion(api_key, dashboard)
+
+
+class TestNotificationDestinationChanges(BaseChangeTestCase):
+    def test_creation(self):
+        destination = self.factory.create_destination()
+        self._test_creation(destination, destination)
+
+    def test_modification(self):
+        destination = self.factory.create_destination()
+        self._test_modification(destination, destination, {"name": "New Name"})
+
+    def test_deletion(self):
+        destination = self.factory.create_destination()
+        self._test_deletion(destination, destination)
+
+
+class TestAlertChanges(BaseChangeTestCase):
+    def test_creation(self):
+        alert = self.factory.create_alert()
+        self._test_creation(alert, alert)
+
+    def test_modification(self):
+        alert = self.factory.create_alert()
+        self._test_modification(alert, alert, {"name": "New Name"})
+
+    def test_deletion(self):
+        alert = self.factory.create_alert()
+        self._test_deletion(alert, alert)
+
+
+# Alert subscriptions can only be added and removed, so not testing for modifications
+class TestAlertSubscriptionChanges(BaseChangeTestCase):
+    def test_creation(self):
+        alert = self.factory.create_alert()
+        subscription = self.factory.create_alert_subscription(alert=alert)
+        self._test_creation(subscription, alert)
+
+    def test_deletion(self):
+        alert = self.factory.create_alert()
+        subscription = self.factory.create_alert_subscription(alert=alert)
+        self._test_deletion(subscription, alert)