So implementieren Sie Synchronisationstag in Django

Question

Wir machen eine Synchronisation zwischen einer Master-Datenbank und vielen Slave-Datenbanken über enge Verbindung (7kb). Um die Menge der gesendeten Daten zu minimieren, markieren wir jeden Datensatz, der erfolgreich gesendet wurde.

Dazu haben wir folgende Modelle:

class SynchronizationTag(models.Model): 
    STATUSES = ((0, "Invalid"), 
       (1, "Pending"), 
       (2, "Synchronized"), 
       ) 
    status = models.IntegerField(choices=STATUSES, default = 1) 
    storage = models.ForeignKey("Storage") 

    _content_type = models.ForeignKey(ContentType) 
    _object_id = models.PositiveIntegerField() 
    content_object = generic.GenericForeignKey('_content_type', '_object_id') 

class Storage(models.Model): 
    id = models.AutoField(primary_key=True) 

Datensätze auszuwählen, die Synchronisation benötigen schrieben wir eine allgemeine Abfrage, die Datensätze herausfiltern würde, die aus einer Abfrage als synchronisiert getaggt wurden.

Wir kamen mit folgenden suboptimale Lösung:

delta = queryset.extra(
    select={ 
     "status" : ("SELECT status FROM rv3adapter_synchronizationtag " 
        "WHERE `_content_type_id` = %d and `_object_id`= %s.id and `storage_id` = %d" 
        % (content_type.pk, table, storage.pk)) 
     }, 
    where=["`status` <> 2 or `status` is NULL") 

Die Abfrage oben jetzt gebrochen ist, wie wir einige Modelle haben, die keine Primärschlüssel namens id haben.

Wissen Sie, wie Sie obige Abfrage verbessern/korrigieren können?

Hinweis: Die Abfrage muss Django-Objekte zurückgeben.

Answer 1

Ermitteln Sie den Namen des Primärschlüsselfelds mit queryset.model._meta.pk.name.

Sie könnten auch ._meta.db_table nützlich finden.

Zugegeben, diese ist mit einer inoffiziellen API, die nicht in zukünftigen Versionen von Django kompatibel ist, aber wenn Sie ein paar Komponententests haben, sollten Sie in Ordnung sein.