diff --git a/README.md b/README.md index dd49752..3d6feca 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,29 @@ Options: - `--model app_label.ModelName` — only process a single configured model. - `--batch-size N` — rows examined per transaction (default 1000). +## Inspecting a single object + +The `gc_inspect` command reports the garbage collection status of one object +without deleting anything. It's useful for answering "why is this row still +here?" or "would this row be collected on the next run?": + +``` +python manage.py gc_inspect myapp.Attachment 42 +``` + +The model must be configured for garbage collection. The reported status is +one of: + +- *object not found* — no row with that primary key exists. +- *object excluded by filter* — the row exists but the model's `gc_filter` + excludes it, so it is not currently a deletion candidate (a time-based + filter may still let it through later). +- *object referenced by N other objects* — the row is still referenced by + foreign keys (each referencing `model.field (pk=...)` is listed), so it is + retained. +- *object is unreferenced and would be collected* — the row passes the filter + and nothing references it, so a `gc --delete` run would remove it. + A typical deployment runs it from a daily cron job: ``` diff --git a/django_gc/core.py b/django_gc/core.py new file mode 100644 index 0000000..84b588b --- /dev/null +++ b/django_gc/core.py @@ -0,0 +1,63 @@ +"""Shared helpers used by more than one garbage collection command.""" + +from typing import Any + +from django.apps import apps +from django.conf import settings +from django.core.management.base import CommandError +from django.db import models +from django.db.models import Model + +from django_gc.registry import get_gc_registry + + +def get_combined_config() -> dict[str, dict[str, Any]]: + """Return the garbage collection config for all configured models. + + Combines ``settings.GARBAGE_COLLECTION_CONFIG`` with the registry-discovered + config from model class attributes, raising if a model is configured in both + places. + """ + hardcoded_config = getattr(settings, 'GARBAGE_COLLECTION_CONFIG', {}) + registry_config = get_gc_registry() + + collisions = set(hardcoded_config.keys()) & set(registry_config.keys()) + if collisions: + raise CommandError( + f"Configuration collision detected for models: {', '.join(sorted(collisions))}. " + "Models cannot be configured both in settings.GARBAGE_COLLECTION_CONFIG " + "and via model class attributes." + ) + + combined_config = hardcoded_config.copy() + combined_config.update(registry_config) + return combined_config + + +def find_fk_fields( + target_model: type[Model], + ignored_fields: list[str], +) -> list[tuple[type[Model], models.ForeignKey[Model, Model]]]: + """Find all ForeignKey fields pointing to target_model, excluding ignored fields.""" + fk_fields = [] + ignored_set = set(ignored_fields) + matched_ignored = set() + + for model in apps.get_models(): + for field in model._meta.get_fields(): + if isinstance(field, models.ForeignKey) and field.related_model == target_model: + field_path = f"{model._meta.label}.{field.name}" + if field_path in ignored_set: + matched_ignored.add(field_path) + else: + fk_fields.append((model, field)) + + unmatched_ignored = ignored_set - matched_ignored + if unmatched_ignored: + raise CommandError( + f"Ignored referencing fields for {target_model._meta.label} do not match " + f"any ForeignKey pointing to it: {', '.join(sorted(unmatched_ignored))}. " + "The configuration is stale or contains a typo." + ) + + return fk_fields diff --git a/django_gc/management/commands/gc.py b/django_gc/management/commands/gc.py index 8a7ccf5..3f33203 100644 --- a/django_gc/management/commands/gc.py +++ b/django_gc/management/commands/gc.py @@ -2,12 +2,11 @@ from typing import Any, Callable from django.apps import apps -from django.conf import settings -from django.core.management.base import BaseCommand, CommandError +from django.core.management.base import BaseCommand from django.db import models, transaction from django.db.models import Model, QuerySet -from django_gc.registry import get_gc_registry +from django_gc.core import find_fk_fields, get_combined_config class Command(BaseCommand): @@ -29,22 +28,7 @@ def handle(self, *args: Any, **options: Any) -> None: total_deleted = 0 - # Combine hardcoded config with registry-discovered config - # Registry takes precedence over hardcoded config - hardcoded_config = getattr(settings, 'GARBAGE_COLLECTION_CONFIG', {}) - registry_config = get_gc_registry() - - # Check for collisions between hardcoded and registry config - collisions = set(hardcoded_config.keys()) & set(registry_config.keys()) - if collisions: - raise CommandError( - f"Configuration collision detected for models: {', '.join(sorted(collisions))}. " - "Models cannot be configured both in settings.GARBAGE_COLLECTION_CONFIG " - "and via model class attributes." - ) - - combined_config = hardcoded_config.copy() - combined_config.update(registry_config) + combined_config = get_combined_config() for model_label, config in combined_config.items(): if specific_model and model_label != specific_model: @@ -100,35 +84,6 @@ def filter_func(qs: QuerySet[Model]) -> QuerySet[Model]: return total_deleted -def find_fk_fields( - target_model: type[Model], - ignored_fields: list[str], -) -> list[tuple[type[Model], models.ForeignKey[Model, Model]]]: - """Find all ForeignKey fields pointing to target_model, excluding ignored fields.""" - fk_fields = [] - ignored_set = set(ignored_fields) - matched_ignored = set() - - for model in apps.get_models(): - for field in model._meta.get_fields(): - if isinstance(field, models.ForeignKey) and field.related_model == target_model: - field_path = f"{model._meta.label}.{field.name}" - if field_path in ignored_set: - matched_ignored.add(field_path) - else: - fk_fields.append((model, field)) - - unmatched_ignored = ignored_set - matched_ignored - if unmatched_ignored: - raise CommandError( - f"Ignored referencing fields for {target_model._meta.label} do not match " - f"any ForeignKey pointing to it: {', '.join(sorted(unmatched_ignored))}. " - "The configuration is stale or contains a typo." - ) - - return fk_fields - - @transaction.atomic def process_batch( model: type[Model], diff --git a/django_gc/management/commands/gc_inspect.py b/django_gc/management/commands/gc_inspect.py new file mode 100644 index 0000000..cbb333b --- /dev/null +++ b/django_gc/management/commands/gc_inspect.py @@ -0,0 +1,103 @@ +from argparse import ArgumentParser +from dataclasses import dataclass, field +from typing import Any, Callable + +from django.apps import apps +from django.core.management.base import BaseCommand, CommandError +from django.db.models import Model, QuerySet + +from django_gc.core import find_fk_fields, get_combined_config + + +class Command(BaseCommand): + """Report the garbage collection status of a single object. + + Read-only: explains why a specific row is or is not garbage collected + without deleting anything. + """ + + help = 'Report the garbage collection status of a single object' + + def add_arguments(self, parser: ArgumentParser) -> None: + parser.add_argument('model', type=str, help='Model to inspect (app_label.model_name)') + parser.add_argument('pk', type=str, help='Primary key of the object to inspect') + + def handle(self, *args: Any, **options: Any) -> None: + model_label = options['model'] + pk = options['pk'] + + combined_config = get_combined_config() + if model_label not in combined_config: + raise CommandError( + f'{model_label} is not configured for garbage collection.' + ) + + model = apps.get_model(model_label) + result = inspect_object(model, combined_config[model_label], pk) + + print(result.message) + for reference in result.references: + print(f' - {reference}') + + +@dataclass +class InspectResult: + """The garbage collection status of a single object. + + ``status`` is one of ``'not_found'``, ``'excluded_by_filter'``, + ``'referenced'`` (with the blocking foreign keys in ``references``), or + ``'collectable'``. + """ + + status: str + message: str + references: list[str] = field(default_factory=list) + + +def inspect_object( + model: type[Model], + config: dict[str, Any], + pk: Any, +) -> InspectResult: + """Report the garbage collection status of a single object. + + Mirrors the decisions made by the gc command (filter, then foreign key + references) without deleting anything, so it can be used to explain why a + specific row is or is not collected. + """ + label = model._meta.label + fk_fields = find_fk_fields(model, config.get('ignored_referencing_fields', [])) + + # Model.objects is present at runtime but mypy doesn't recognize it on type[Model] + if not model.objects.filter(pk=pk).exists(): # type: ignore[attr-defined] + return InspectResult('not_found', f'{label} pk={pk}: object not found') + + filter_func: Callable[[QuerySet[Model]], QuerySet[Model]] | None = config.get('filter') + if filter_func is not None: + candidates = filter_func(model.objects.all()) # type: ignore[attr-defined] + if not candidates.filter(pk=pk).exists(): + return InspectResult( + 'excluded_by_filter', + f'{label} pk={pk}: object excluded by filter (not currently a collection candidate)', + ) + + references = [] + for ref_model, fk_field in fk_fields: + ref_pks = ref_model.objects.filter( # type: ignore[attr-defined] + **{fk_field.name: pk} + ).values_list('pk', flat=True) + for ref_pk in ref_pks: + references.append(f'{ref_model._meta.label}.{fk_field.name} (pk={ref_pk})') + + if references: + plural = 's' if len(references) != 1 else '' + return InspectResult( + 'referenced', + f'{label} pk={pk}: object referenced by {len(references)} other object{plural}', + references, + ) + + return InspectResult( + 'collectable', + f'{label} pk={pk}: object is unreferenced and would be collected', + ) diff --git a/tests/test_gc_command.py b/tests/test_gc_command.py index 45cb531..546902f 100644 --- a/tests/test_gc_command.py +++ b/tests/test_gc_command.py @@ -2,6 +2,7 @@ from django.core.management import call_command from django.core.management.base import CommandError +from django_gc.management.commands.gc_inspect import inspect_object from tests.testapp.models import Cache, CacheLog, Document, Reference @@ -75,3 +76,82 @@ def test_collision_between_settings_and_model_config(settings) -> None: with pytest.raises(CommandError, match='collision'): call_command('gc') + + +@pytest.mark.django_db +def test_inspect_object_not_found() -> None: + result = inspect_object(Document, {}, 999999) + + assert result.status == 'not_found' + assert result.references == [] + + +@pytest.mark.django_db +def test_inspect_object_collectable() -> None: + document = Document.objects.create() + + result = inspect_object(Document, {}, document.pk) + + assert result.status == 'collectable' + assert result.references == [] + + +@pytest.mark.django_db +def test_inspect_object_referenced_lists_referencing_objects() -> None: + document = Document.objects.create() + reference = Reference.objects.create(document=document) + + result = inspect_object(Document, {}, document.pk) + + assert result.status == 'referenced' + assert result.references == [f'testapp.Reference.document (pk={reference.pk})'] + assert 'referenced by 1 other object' in result.message + + +@pytest.mark.django_db +def test_inspect_object_excluded_by_filter() -> None: + document = Document.objects.create() + config = {'filter': lambda qs: qs.none()} + + result = inspect_object(Document, config, document.pk) + + assert result.status == 'excluded_by_filter' + + +@pytest.mark.django_db +def test_inspect_ignores_ignored_referencing_fields() -> None: + cache = Cache.objects.create() + CacheLog.objects.create(cache=cache) + config = {'ignored_referencing_fields': ['testapp.CacheLog.cache']} + + result = inspect_object(Cache, config, cache.pk) + + # The ignored reference does not keep the row alive. + assert result.status == 'collectable' + + +@pytest.mark.django_db +def test_inspect_command_unconfigured_model_fails() -> None: + with pytest.raises(CommandError, match='not configured'): + call_command('gc_inspect', 'testapp.Reference', '1') + + +@pytest.mark.django_db +def test_inspect_command_does_not_delete() -> None: + document = Document.objects.create() + + call_command('gc_inspect', 'testapp.Document', str(document.pk)) + + assert Document.objects.filter(pk=document.pk).exists() + + +@pytest.mark.django_db +def test_inspect_command_renders_references(capsys) -> None: + document = Document.objects.create() + reference = Reference.objects.create(document=document) + + call_command('gc_inspect', 'testapp.Document', str(document.pk)) + + out = capsys.readouterr().out + assert 'object referenced by 1 other object' in out + assert f'- testapp.Reference.document (pk={reference.pk})' in out