feat(serverdb): Allow to redirect attributes via alias

packages/serveradmin/serveradmin/api/views.py

@@ -18,7 +18,7 @@
 from serveradmin.api.decorators import api_view
 from serveradmin.dataset import Query
 from serveradmin.querylog.utils import log_query
-from serveradmin.serverdb.models import Attribute
+from serveradmin.serverdb.models import Attribute, AttributeRedirect
 from serveradmin.serverdb.query_committer import commit_query
 from serveradmin.serverdb.query_executer import execute_query
 from serveradmin.serverdb.query_materializer import (
@@ -67,7 +67,14 @@ def dataset_query(request, app, data):
     order_by = data.get('order_by')
 
     start = monotonic()
+
+    # Resolve alias attributes to real attributes.  The requested restrict
+    # is kept to rename the attributes in the results back to the aliases.
+    requested_restrict = restrict
+    filters, restrict, order_by = AttributeRedirect.resolve_aliases(filters, restrict, order_by)
+
     result = execute_query(filters, restrict, order_by)
+    result = AttributeRedirect.restore_aliases(requested_restrict, result)
     duration_seconds = monotonic() - start
 
     # Query(...) is instantiated here only for its repr(); it is never

packages/serveradmin/serveradmin/serverdb/admin.py

@@ -1,6 +1,6 @@
 """Serveradmin - Django Admin Setup
 
-Copyright (c) 2019 InnoGames GmbH
+Copyright (c) 2026 InnoGames GmbH
 """
 
 from django.contrib import admin
@@ -14,9 +14,8 @@
     Servertype,
     Attribute,
     ServertypeAttribute,
-    Server,
     ServerRelationAttribute,
-    ServerStringAttribute,
+    ServerStringAttribute, AttributeRedirect,
 )
 
 
@@ -106,5 +105,14 @@ def get_hovertext(self, obj):
         )
 
 
+class AttributeRedirectAdmin(admin.ModelAdmin):
+    model = AttributeRedirect
+
+    list_display = ['alias', 'target__attribute_id', ]
+    search_fields = ['alias', 'target__attribute_id', ]
+    list_filter = ['alias', 'target__attribute_id', ]
+
+
 admin.site.register(Servertype, ServertypeAdmin)
 admin.site.register(Attribute, AttributeAdmin)
+admin.site.register(AttributeRedirect, AttributeRedirectAdmin)

packages/serveradmin/serveradmin/serverdb/migrations/0026_attributeredirect.py

@@ -0,0 +1,21 @@
+# Generated by Django 5.2.17 on 2026-09-25 14:14
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('serverdb', '0025_rename_serverbooleanattribute_attribute_server_bool_attribu_25fb6c_idx_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AttributeRedirect',
+            fields=[
+                ('alias', models.CharField(help_text="The 'virtual' attribute name (e.g. old attribute)", max_length=32, primary_key=True, serialize=False)),
+                ('target', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='serverdb.attribute')),
+            ],
+        ),
+    ]

packages/serveradmin/serveradmin/serverdb/models.py

@@ -402,6 +402,141 @@ def clean(self):
         super(Attribute, self).clean()
 
 
+class AttributeRedirect(models.Model):
+    """Redirect alias to an existing attribute
+
+    Purpose of this is to allow graceful renaming of attributes.
+
+    One can delete and attribute create a new one and set up a redirect
+    from the old name to the new name.
+    """
+
+    alias = models.CharField(
+        max_length=32,
+        help_text="The 'virtual' attribute name (e.g. old attribute)",
+        primary_key=True,
+    )
+    target = models.ForeignKey(Attribute, on_delete=models.CASCADE)
+
+    def clean(self):
+        super().clean()
+
+        if Attribute.objects.filter(attribute_id=self.alias).exists():
+            raise ValidationError({
+                "alias": "Creating a alias that matches an existing attribute not allowed!"
+            })
+
+    @classmethod
+    def get_mapping(cls):
+        """Return a dictionary mapping alias names to real attribute_ids"""
+        return dict(cls.objects.values_list('alias', 'target_id'))
+
+    @classmethod
+    def resolve_aliases(cls, *objs):
+        """Resolve alias names in the given objects to real attribute_ids."""
+        mapping = cls.get_mapping()
+        return tuple(_apply_aliases(mapping, obj) for obj in objs)
+
+    @classmethod
+    def restore_aliases(cls, restrict, results):
+        """Rename attributes in query results back to the requested aliases
+
+        ``restrict`` must be the clause as the client sent it, before the
+        aliases were resolved.  ``results`` are modified in place and
+        returned for convenience.
+        """
+        if restrict is None:
+            return results
+
+        mapping = cls.get_mapping()
+        if mapping:
+            _restore_aliases(mapping, restrict, results)
+
+        return results
+
+
+def _apply_aliases(mapping, obj):
+    """Recursively replace alias attribute names in a query argument
+
+    Supported shapes are the ones passed to a query:
+
+    * ``filters``: ``{attribute_id: filter}``, only keys are attribute names,
+      the filter objects are left untouched.
+    * ``restrict``: ``[attribute_id, {attribute_id: [restrict, ...]}, ...]``,
+      where a dictionary item is a join into a related object whose value is
+      again a restrict clause.
+    * ``order_by``: ``[attribute_id, ...]``
+    """
+    if obj is None:
+        return None
+    if isinstance(obj, str):
+        return mapping.get(obj, obj)
+    if isinstance(obj, list):
+        return [
+            _apply_aliases(mapping, item)
+            if isinstance(item, (str, list, dict)) else item
+            for item in obj
+        ]
+    if isinstance(obj, dict):
+        # Only keys are attribute names.  A value is either a nested restrict
+        # clause of a join (list) or a filter value, which must stay as is.
+        return {
+            mapping.get(key, key): _apply_aliases(mapping, value)
+            if isinstance(value, (list, dict)) else value
+            for key, value in obj.items()
+        }
+    raise TypeError(f'Unsupported type {type(obj).__name__}')
+
+
+def _restore_aliases(mapping, restrict, results):
+    """Rename real attribute_ids in results back to the names in restrict
+
+    The query is executed with the resolved attribute_ids, so the results
+    are keyed by the real names.  The client however expects the names it
+    asked for.  Joins are followed recursively.  If both the alias and the
+    real attribute are requested, the value ends up under both names.
+
+    The results are dictionaries, usually DatasetObjects.  Their base dict
+    methods are used on purpose, so the objects are not marked as changed.
+    """
+    # real attribute_id -> [(requested name, sub restrict or None), ...]
+    requested = {}
+    for item in restrict:
+        if isinstance(item, dict):
+            for name, sub_restrict in item.items():
+                real_id = mapping.get(name, name)
+                requested.setdefault(real_id, []).append((name, sub_restrict))
+        else:
+            real_id = mapping.get(item, item)
+            requested.setdefault(real_id, []).append((item, None))
+
+    # Nothing to do for attributes requested by their real name without join
+    requested = {
+        real_id: names for real_id, names in requested.items()
+        if any(name != real_id or sub is not None for name, sub in names)
+    }
+    if not requested:
+        return
+
+    for obj in results:
+        for real_id, names in requested.items():
+            if real_id not in obj:
+                continue
+
+            if any(name == real_id for name, _ in names):
+                value = dict.__getitem__(obj, real_id)
+            else:
+                value = dict.pop(obj, real_id)
+
+            for name, sub_restrict in names:
+                if sub_restrict is not None and value is not None:
+                    if isinstance(value, (set, frozenset, list, tuple)):
+                        _restore_aliases(mapping, sub_restrict, value)
+                    else:
+                        _restore_aliases(mapping, sub_restrict, [value])
+                dict.__setitem__(obj, name, value)
+
+
 class ServerTableSpecial(object):
     def __init__(self, field, unique=False):
         self.field = field

packages/serveradmin/serveradmin/serverdb/tests/test_attribute_redirect.py

@@ -0,0 +1,173 @@
+from django.contrib.auth.models import User
+from django.test import SimpleTestCase, TransactionTestCase
+
+from adminapi.dataset import DatasetObject
+from adminapi.filters import Regexp
+from serveradmin.api.views import dataset_query
+from serveradmin.apps.models import Application
+from serveradmin.serverdb.models import (
+    Attribute,
+    AttributeRedirect,
+    _apply_aliases,
+    _restore_aliases,
+)
+
+
+class ApplyAliasesTestCase(SimpleTestCase):
+    mapping = {'old_name': 'new_name', 'old_code': 'short_code'}
+
+    def test_none(self):
+        self.assertIsNone(_apply_aliases(self.mapping, None))
+
+    def test_string(self):
+        self.assertEqual(_apply_aliases(self.mapping, 'old_name'), 'new_name')
+        self.assertEqual(_apply_aliases(self.mapping, 'hostname'), 'hostname')
+
+    def test_flat_restrict(self):
+        self.assertEqual(
+            _apply_aliases(self.mapping, ['hostname', 'old_name']),
+            ['hostname', 'new_name'],
+        )
+
+    def test_restrict_with_joins(self):
+        restrict = [{'old_name': ['old_code', 'object_id']}, 'hostname']
+        self.assertEqual(
+            _apply_aliases(self.mapping, restrict),
+            [{'new_name': ['short_code', 'object_id']}, 'hostname'],
+        )
+
+    def test_restrict_with_nested_joins(self):
+        restrict = [{'project': [{'old_name': ['old_code']}, 'hostname']}]
+        self.assertEqual(
+            _apply_aliases(self.mapping, restrict),
+            [{'project': [{'new_name': ['short_code']}, 'hostname']}],
+        )
+
+    def test_filters_keep_filter_objects(self):
+        regexp = Regexp('^foo')
+        # A plain string value is a filter value, not an attribute name,
+        # so it must not be rewritten even if it matches an alias.
+        filters = {'old_name': regexp, 'hostname': 'old_code'}
+        result = _apply_aliases(self.mapping, filters)
+        self.assertEqual(set(result), {'new_name', 'hostname'})
+        self.assertIs(result['new_name'], regexp)
+        self.assertEqual(result['hostname'], 'old_code')
+
+    def test_unsupported_type(self):
+        with self.assertRaises(TypeError):
+            _apply_aliases(self.mapping, 42)
+
+
+class RestoreAliasesTestCase(SimpleTestCase):
+    mapping = {'operating_system': 'os', 'hv': 'hypervisor', 'name': 'hostname'}
+
+    def test_flat_alias_is_renamed(self):
+        results = [{'hostname': 'test0', 'os': 'wheezy'}]
+        _restore_aliases(self.mapping, ['hostname', 'operating_system'], results)
+        self.assertEqual(results, [{'hostname': 'test0', 'operating_system': 'wheezy'}])
+
+    def test_real_name_is_untouched(self):
+        results = [{'hostname': 'test0', 'os': 'wheezy'}]
+        _restore_aliases(self.mapping, ['hostname', 'os'], results)
+        self.assertEqual(results, [{'hostname': 'test0', 'os': 'wheezy'}])
+
+    def test_alias_and_real_name_both_requested(self):
+        results = [{'os': 'wheezy'}]
+        _restore_aliases(self.mapping, ['os', 'operating_system'], results)
+        self.assertEqual(results, [{'os': 'wheezy', 'operating_system': 'wheezy'}])
+
+    def test_missing_attribute_is_not_added(self):
+        results = [{'hostname': 'test0'}]
+        _restore_aliases(self.mapping, ['hostname', 'operating_system'], results)
+        self.assertEqual(results, [{'hostname': 'test0'}])
+
+    def test_join_single_relation(self):
+        results = [{'hostname': 'vm-1', 'hypervisor': {'hostname': 'hv-1', 'os': 'wheezy'}}]
+        _restore_aliases(self.mapping, ['hostname', {'hv': ['name', 'operating_system']}], results)
+        self.assertEqual(results, [
+            {'hostname': 'vm-1', 'hv': {'name': 'hv-1', 'operating_system': 'wheezy'}},
+        ])
+
+    def test_join_with_none_value(self):
+        results = [{'hostname': 'vm-1', 'hypervisor': None}]
+        _restore_aliases(self.mapping, [{'hv': ['name']}], results)
+        self.assertEqual(results, [{'hostname': 'vm-1', 'hv': None}])
+
+    def test_join_multi_relation(self):
+        vm1 = DatasetObject({'hostname': 'vm-1', 'os': 'wheezy'}, 7)
+        vm2 = DatasetObject({'hostname': 'vm-2', 'os': 'squeeze'}, 8)
+        results = [DatasetObject({'hostname': 'hv-1', 'vms': [vm1, vm2]}, 6)]
+        _restore_aliases(self.mapping, [{'vms': ['name', 'operating_system']}], results)
+
+        self.assertEqual(set(results[0]), {'hostname', 'vms'})
+        self.assertEqual(
+            sorted(sorted(vm.items()) for vm in results[0]['vms']),
+            [
+                [('name', 'vm-1'), ('operating_system', 'wheezy')],
+                [('name', 'vm-2'), ('operating_system', 'squeeze')],
+            ],
+        )
+
+    def test_join_real_name_with_nested_alias(self):
+        results = [{'hypervisor': {'os': 'wheezy'}}]
+        _restore_aliases(self.mapping, [{'hypervisor': ['operating_system']}], results)
+        self.assertEqual(results, [{'hypervisor': {'operating_system': 'wheezy'}}])
+
+    def test_dataset_objects_stay_clean(self):
+        nested = DatasetObject({'hostname': 'hv-1', 'os': 'wheezy'}, 6)
+        obj = DatasetObject({'hostname': 'vm-1', 'os': 'squeeze', 'hypervisor': nested}, 7)
+        _restore_aliases(
+            self.mapping,
+            ['operating_system', {'hv': ['operating_system']}],
+            [obj],
+        )
+        self.assertEqual(
+            obj, {'hostname': 'vm-1', 'operating_system': 'squeeze', 'hv': nested},
+        )
+        self.assertEqual(nested, {'hostname': 'hv-1', 'operating_system': 'wheezy'})
+        self.assertFalse(obj.is_dirty())
+        self.assertFalse(nested.is_dirty())
+
+
+class DatasetQueryAliasTestCase(TransactionTestCase):
+    fixtures = ['test_dataset.json']
+
+    def setUp(self):
+        super().setUp()
+        user = User.objects.create_user('alice')
+        self.app = Application.objects.create(name='app', owner=user, location='')
+        AttributeRedirect.objects.create(
+            alias='operating_system', target=Attribute.objects.get(pk='os'),
+        )
+        AttributeRedirect.objects.create(
+            alias='hv', target=Attribute.objects.get(pk='hypervisor'),
+        )
+
+    def _query(self, filters, restrict):
+        response = dataset_query.__wrapped__(
+            None, self.app, {'filters': filters, 'restrict': restrict},
+        )
+        self.assertEqual(response['status'], 'success')
+        return response['result']
+
+    def test_restrict_alias_is_returned_as_alias(self):
+        result = self._query({'hostname': 'test0'}, ['hostname', 'operating_system'])
+        self.assertEqual(result, [{'hostname': 'test0', 'operating_system': 'wheezy'}])
+
+    def test_restrict_alias_and_real_name(self):
+        result = self._query({'hostname': 'test0'}, ['os', 'operating_system'])
+        self.assertEqual(result, [{'os': 'wheezy', 'operating_system': 'wheezy'}])
+
+    def test_restrict_join_alias(self):
+        result = self._query({'hostname': 'vm-1'}, ['hostname', {'hv': ['hostname']}])
+        self.assertEqual(result, [{'hostname': 'vm-1', 'hv': {'hostname': 'hv-1'}}])
+
+    def test_filter_alias_resolves_objects(self):
+        result = self._query({'operating_system': 'wheezy'}, ['hostname'])
+        self.assertEqual(result, [{'hostname': 'test0'}])
+
+    def test_no_restrict_returns_real_names(self):
+        result = self._query({'hostname': 'test0'}, [])
+        self.assertEqual(len(result), 1)
+        self.assertIn('os', result[0])
+        self.assertNotIn('operating_system', result[0])