Implement logging

Author: pfouque

.gitignore

@@ -133,3 +133,6 @@ test.db
 
 # django fsm command tests
 exports/*
+
+# codex
+.codex/*

README.md

@@ -418,6 +418,45 @@ executed transitions, make sure:
 Following these recommendations, `ConcurrentTransitionMixin` will cause a
 rollback of all changes executed in an inconsistent state.
 
+## Transition tracking
+
+Use `@django_fsm.track()` to write state changes to a log table.
+By default, it writes to `django_fsm.GenericTransitionLog` (single table).
+If you prefer one table per model, define your own log model and pass it in.
+You can also capture `author` and `description` for each transition.
+
+```python
+import django_fsm
+from django_fsm.log import fsm_log_by
+from django_fsm.log import fsm_log_description
+from django.db import models
+
+
+@django_fsm.track()
+class BlogPost(models.Model):
+    state = django_fsm.FSMField(default="new")
+
+    @fsm_log_by
+    @fsm_log_description
+    @django_fsm.transition(field=state, source="new", target="published")
+    def publish(self):
+        pass
+```
+
+```python
+import django_fsm
+from django.db import models
+
+
+class BlogPostLog(django_fsm.TransitionLogBase):
+    post = models.ForeignKey("BlogPost", on_delete=models.CASCADE, related_name="transition_logs")
+
+
+@django_fsm.track(log_model=BlogPostLog, relation_field="post")
+class BlogPost(models.Model):
+    state = django_fsm.FSMField(default="new")
+```
+
 ## Drawing transitions
 
 Render a graphical overview of your model transitions.
@@ -460,7 +499,6 @@ INSTALLED_APPS = (
 ## Extensions
 
 - Admin integration: <https://github.com/coral-li/django-fsm-2-admin>
-- Transition logging: <https://github.com/gizmag/django-fsm-log>
 
 ## Contributing
 

django_fsm/apps.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from django.apps import AppConfig
+
+
+class DjangoFSMAppConfig(AppConfig):
+    name = "django_fsm"
+    verbose_name = "Django FSM"

django_fsm/log.py

@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+import contextlib
+import typing
+from dataclasses import dataclass
+from functools import partial
+from functools import wraps
+
+from django.contrib.contenttypes.models import ContentType
+from django.db import models
+
+from .models import GenericTransitionLog
+from .models import TransitionLogBase
+from .signals import post_transition
+
+if typing.TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from . import _Field
+
+
+__all__ = [
+    "GenericTransitionLog",
+    "TransitionLogBase",
+    "fsm_log_author",
+    "fsm_log_by",
+    "fsm_log_description",
+    "track",
+]
+
+
+@dataclass(frozen=True)
+class TrackConfig:
+    log_model: type[TransitionLogBase] | None
+    relation_field: str | None
+
+
+_registry: dict[type[models.Model], TrackConfig] = {}
+NOTSET = object()
+
+
+def track(
+    *,
+    log_model: type[TransitionLogBase] | None = None,
+    relation_field: str | None = None,
+) -> Callable[[type[models.Model]], type[models.Model]]:
+    def decorator(model_cls: type[models.Model]) -> type[models.Model]:
+        if model_cls._meta.abstract:
+            raise TypeError("django_fsm.track cannot be used with abstract models")
+        config = TrackConfig(log_model=log_model, relation_field=relation_field)
+        _registry[model_cls] = config
+        _connect_signal(model_cls)
+        return model_cls
+
+    return decorator
+
+
+class FSMLogDescriptor:
+    ATTR_PREFIX = "__django_fsm_log_attr_"
+
+    def __init__(self, instance: models.Model, attribute: str, value: typing.Any = NOTSET):
+        self.instance = instance
+        self.attribute = attribute
+        if value is not NOTSET:
+            self.set(value)
+
+    def get(self) -> typing.Any:
+        return getattr(self.instance, self.ATTR_PREFIX + self.attribute)
+
+    def set(self, value: typing.Any) -> None:
+        setattr(self.instance, self.ATTR_PREFIX + self.attribute, value)
+
+    def __enter__(self) -> typing.Self:
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        with contextlib.suppress(AttributeError):
+            delattr(self.instance, self.ATTR_PREFIX + self.attribute)
+
+
+def fsm_log_by(func: typing.Callable[..., typing.Any]) -> typing.Callable[..., typing.Any]:
+    @wraps(func)
+    def wrapped(instance: models.Model, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
+        if "author" in kwargs:
+            author = kwargs.pop("author")
+        elif "by" in kwargs:
+            author = kwargs.pop("by")
+        else:
+            return func(instance, *args, **kwargs)
+
+        with FSMLogDescriptor(instance, "author", author):
+            return func(instance, *args, **kwargs)
+
+    return wrapped
+
+
+def fsm_log_author(func: typing.Callable[..., typing.Any]) -> typing.Callable[..., typing.Any]:
+    return fsm_log_by(func)
+
+
+def fsm_log_description(
+    func: typing.Callable[..., typing.Any] | None = None,
+    *,
+    allow_inline: bool = False,
+    description: str | None = None,
+) -> typing.Callable[..., typing.Any]:
+    if func is None:
+        return partial(fsm_log_description, allow_inline=allow_inline, description=description)
+
+    @wraps(func)
+    def wrapped(instance: models.Model, *args: typing.Any, **kwargs: typing.Any) -> typing.Any:
+        with FSMLogDescriptor(instance, "description") as descriptor:
+            if "description" in kwargs:
+                descriptor.set(kwargs.pop("description"))
+            elif allow_inline:
+                kwargs["description"] = descriptor
+            else:
+                descriptor.set(description)
+            return func(instance, *args, **kwargs)
+
+    return wrapped
+
+
+def _connect_signal(model_cls: type[models.Model]) -> None:
+    dispatch_uid = f"django_fsm.track.{model_cls._meta.label_lower}"
+    post_transition.connect(
+        _log_transition, sender=model_cls, dispatch_uid=dispatch_uid, weak=False
+    )
+
+
+def _log_transition(
+    sender: type[models.Model],
+    instance: models.Model,
+    name: str,
+    source: typing.Any,
+    target: typing.Any,
+    field: _Field,
+    **kwargs: typing.Any,
+) -> None:
+    config = _registry.get(sender)
+    if not config or instance.pk is None:
+        return
+
+    log_model = config.log_model or GenericTransitionLog
+    method_kwargs = kwargs.get("method_kwargs") or {}
+    author = _extract_log_value(instance, method_kwargs, "author", ("author", "by"))
+    description = _extract_log_value(instance, method_kwargs, "description", ("description",))
+    log_kwargs: dict[str, typing.Any] = {
+        "transition": name,
+        "state_field": field.name,
+        "source": _coerce_state(source),
+        "target": _coerce_state(target),
+    }
+    if author is not None:
+        log_kwargs["author"] = author
+    if description is not None:
+        log_kwargs["description"] = description
+    if issubclass(log_model, GenericTransitionLog):
+        log_kwargs["content_type"] = ContentType.objects.get_for_model(sender)
+        log_kwargs["object_id"] = str(instance.pk)
+    else:
+        relation_field = config.relation_field or _resolve_relation_field(log_model, sender)
+        log_kwargs[relation_field] = instance
+
+    log_model._default_manager.using(instance._state.db).create(**log_kwargs)
+
+
+def _resolve_relation_field(
+    log_model: type[TransitionLogBase], model_cls: type[models.Model]
+) -> str:
+    relation_fields = [
+        field.name
+        for field in log_model._meta.fields
+        if isinstance(field, models.ForeignKey)
+        and _matches_model(field.remote_field.model, model_cls)
+    ]
+    if len(relation_fields) == 1:
+        return relation_fields[0]
+    if not relation_fields:
+        raise ValueError(
+            f"{log_model.__name__} does not define a ForeignKey to {model_cls.__name__}"
+        )
+    raise ValueError(
+        f"{log_model.__name__} has multiple ForeignKey fields to {model_cls.__name__}; "
+        "set relation_field when calling track()"
+    )
+
+
+def _coerce_state(value: typing.Any) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, models.Model):
+        return str(value.pk)
+    return str(value)
+
+
+def _matches_model(remote_model: typing.Any, model_cls: type[models.Model]) -> bool:
+    if remote_model == model_cls:
+        return True
+    if isinstance(remote_model, str):
+        return remote_model == model_cls.__name__ or remote_model.endswith(f".{model_cls.__name__}")
+    return False
+
+
+def _extract_log_value(
+    instance: models.Model,
+    method_kwargs: dict[str, typing.Any],
+    attribute: str,
+    keys: tuple[str, ...],
+) -> typing.Any:
+    for key in keys:
+        if key in method_kwargs:
+            value = method_kwargs[key]
+            if isinstance(value, FSMLogDescriptor):
+                try:
+                    return value.get()
+                except AttributeError:
+                    return None
+            return value
+    try:
+        return FSMLogDescriptor(instance, attribute).get()
+    except AttributeError:
+        return None

django_fsm/migrations/0001_initial.py

@@ -0,0 +1,60 @@
+# Generated by Django 4.2.16 on 2026-01-28 16:49
+from __future__ import annotations
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("contenttypes", "0002_remove_content_type_name"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="GenericTransitionLog",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("transition", models.CharField(max_length=255)),
+                ("state_field", models.CharField(max_length=255)),
+                ("source", models.TextField(blank=True, null=True)),
+                ("target", models.TextField(blank=True, null=True)),
+                ("description", models.TextField(blank=True, null=True)),
+                ("object_id", models.TextField()),
+                (
+                    "author",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="+",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "content_type",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="contenttypes.contenttype"
+                    ),
+                ),
+            ],
+            options={
+                "indexes": [
+                    models.Index(
+                        fields=["content_type", "object_id"], name="django_fsm__content_cda773_idx"
+                    )
+                ],
+            },
+        ),
+    ]

django_fsm/migrations/__init__.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from django.conf import settings
+from django.contrib.contenttypes.fields import GenericForeignKey
+from django.contrib.contenttypes.models import ContentType
+from django.db import models
+
+
+class TransitionLogBase(models.Model):
+    created_at = models.DateTimeField(auto_now_add=True)
+    transition = models.CharField(max_length=255)
+    state_field = models.CharField(max_length=255)
+    source = models.TextField(null=True, blank=True)  # noqa: DJ001
+    target = models.TextField(null=True, blank=True)  # noqa: DJ001
+    author = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="+",
+    )
+    description = models.TextField(null=True, blank=True)  # noqa: DJ001
+
+    class Meta:
+        abstract = True
+
+
+class GenericTransitionLog(TransitionLogBase):  # noqa: DJ008
+    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
+    object_id = models.TextField()
+    content_object = GenericForeignKey("content_type", "object_id")
+
+    class Meta:
+        indexes = [
+            models.Index(fields=["content_type", "object_id"]),
+        ]

django_fsm/models.py

@@ -103,6 +103,7 @@ extend-ignore = [
 ]
 fixable = [
     "I",  # isort
+    "RUF022",
     "RUF100",  # Unused `noqa` directive
     "E501",
 ]