Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:

```
Expand Down
63 changes: 63 additions & 0 deletions django_gc/core.py
Original file line number Diff line number Diff line change
@@ -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
51 changes: 3 additions & 48 deletions django_gc/management/commands/gc.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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:
Expand Down Expand Up @@ -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],
Expand Down
103 changes: 103 additions & 0 deletions django_gc/management/commands/gc_inspect.py
Original file line number Diff line number Diff line change
@@ -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',
)
80 changes: 80 additions & 0 deletions tests/test_gc_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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