#819

Committed 18 Oct 2025 12:03PM UTC coverage: 59.791% (+25.6%) from 34.185%

Build # #819

Build Type

push

local-test

Committed by Michał Pasternak

Commit Message

autostash

Coverage Stats

3781 of 9280 branches covered (40.74%)

Branch coverage included in aggregate %.

24978 of 38819 relevant lines covered (64.34%)

0.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.11

src/pbn_api/models/sentdata.py

from django.core.exceptions import ObjectDoesNotExist
from django.db import models
from django.db.models import JSONField

from pbn_api.utils import compare_dicts

from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType

from django.utils import timezone

from bpp import const
from bpp.models import LinkDoPBNMixin


class SentDataManager(models.Manager):
    def get_for_rec(self, rec):
        return self.get(
            object_id=rec.pk, content_type=ContentType.objects.get_for_model(rec)
        )

    def check_if_needed(self, rec, data: dict):
        """Legacy method - kept for backward compatibility"""
        try:
            sd = self.get_for_rec(rec)
        except SentData.DoesNotExist:
            return True

        if sd.data_sent != data:
            return True

        if not sd.uploaded_okay:
            return True

        return False

    def check_if_upload_needed(self, rec, data: dict):
        """Check if upload needed based on SUCCESSFUL submissions only"""
        try:
            sd = self.get_for_rec(rec)
            # Only skip if data matches AND was successfully submitted
            if (not compare_dicts(sd.data_sent, data)) and sd.submitted_successfully:
                return False
        except SentData.DoesNotExist:
            pass
        return True

    def create_or_update_before_upload(self, rec, data: dict):
        """Create or update SentData record before API call"""
        try:
            sd = self.get_for_rec(rec)
            # Reset fields for new attempt
            sd.submitted_successfully = False
            sd.submitted_at = timezone.now()
            sd.uploaded_okay = False
            sd.api_response_status = None
            sd.exception = None
            sd.data_sent = data  # Update data if changed
            sd.save()
            return sd
        except SentData.DoesNotExist:
            # Create new record if none exists
            return self.create(
                object=rec,
                data_sent=data,
                submitted_successfully=False,
                submitted_at=timezone.now(),
                uploaded_okay=False,
            )

    def mark_as_successful(self, rec, pbn_uid_id=None, api_response_status=None):
        """Mark existing record as successful after API call"""
        sd = self.get_for_rec(rec)
        sd.submitted_successfully = True
        sd.uploaded_okay = True
        sd.pbn_uid_id = pbn_uid_id
        sd.api_response_status = api_response_status
        sd.exception = None
        sd.save()

    def mark_as_failed(self, rec, exception=None, api_response_status=None):
        """Mark existing record as failed after API call"""
        sd = self.get_for_rec(rec)
        sd.submitted_successfully = False
        sd.uploaded_okay = False
        sd.exception = str(exception) if exception else None
        sd.api_response_status = api_response_status
        sd.save()

    def updated(
        self, rec, data: dict, pbn_uid_id=None, uploaded_okay=True, exception=None
    ):
        """Legacy method - kept for backward compatibility"""
        try:
            sd = self.get_for_rec(rec)
        except SentData.DoesNotExist:
            self.create(
                object=rec,
                data_sent=data,
                uploaded_okay=uploaded_okay,
                pbn_uid_id=pbn_uid_id,
                exception=exception,
                submitted_successfully=uploaded_okay,
                submitted_at=timezone.now() if uploaded_okay else None,
            )
            return

        sd.data_sent = data
        sd.uploaded_okay = uploaded_okay
        sd.exception = exception
        sd.pbn_uid_id = pbn_uid_id
        sd.submitted_successfully = uploaded_okay
        if uploaded_okay and not sd.submitted_at:
            sd.submitted_at = timezone.now()
        sd.save()

    def ids_for_model(self, model):
        return self.filter(content_type=ContentType.objects.get_for_model(model))

    def bad_uploads(self, model):
        return (
            self.ids_for_model(model)
            .filter(uploaded_okay=False)
            .values_list("object_id", flat=True)
            .distinct()
        )


class SentData(LinkDoPBNMixin, models.Model):
    url_do_pbn = const.LINK_PBN_DO_PUBLIKACJI
    content_type = models.ForeignKey(
        "contenttypes.ContentType", on_delete=models.CASCADE
    )
    object_id = models.PositiveIntegerField(db_index=True)

    object = GenericForeignKey()

    data_sent = JSONField("Wysłane dane")
    last_updated_on = models.DateTimeField("Data operacji", auto_now=True)

    uploaded_okay = models.BooleanField(
        "Wysłano poprawnie", default=True, db_index=True
    )
    exception = models.TextField("Kod błędu", max_length=65535, blank=True, null=True)

    # New fields for success tracking
    submitted_successfully = models.BooleanField(
        "Wysłano pomyślnie",
        default=False,
        db_index=True,
        help_text="True gdy API call zakończył się sukcesem",
    )
    submitted_at = models.DateTimeField(
        "Data wysyłki",
        null=True,
        blank=True,
        help_text="Kiedy dane zostały wysłane do PBN",
    )
    api_response_status = models.TextField(
        "Status odpowiedzi API", null=True, blank=True, help_text="Odpowiedź z PBN API"
    )

    pbn_uid = models.ForeignKey(
        "pbn_api.Publication",
        verbose_name="Publikacja z PBN",
        blank=True,
        null=True,
        on_delete=models.SET_NULL,
    )

    typ_rekordu = models.CharField(max_length=50, blank=True, null=True)

    objects = SentDataManager()

    class Meta:
        verbose_name = "Informacja o wysłanych danych"
        verbose_name_plural = "Informacje o wysłanych danych"

    object.verbose_name = "Rekord"

    def __str__(self):
        return (
            f"Informacja o wysłanych do PBN danych dla rekordu ({self.content_type_id},{self.object_id}) "
            f"z dnia {self.last_updated_on} (status: {'OK' if self.uploaded_okay else 'ERR'})"
        )

    def link_do_pbn_wartosc_id(self):
        return self.pbn_uid_id

    def rekord_w_bpp(self):
        try:
            return self.object
        except ObjectDoesNotExist:
            pass

    def save(
        self, force_insert=False, force_update=False, using=None, update_fields=None
    ):

        if update_fields and "data_sent" in update_fields:
            if self.typ_rekordu != self.data_sent.get("type"):
                update_fields.append("typ_rekordu")

        self.typ_rekordu = self.data_sent.get("type")
        return super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )

1	from django.core.exceptions import ObjectDoesNotExist	1✔
2	from django.db import models	1✔
3	from django.db.models import JSONField	1✔
4
5	from pbn_api.utils import compare_dicts	1✔
6
7	from django.contrib.contenttypes.fields import GenericForeignKey	1✔
8	from django.contrib.contenttypes.models import ContentType	1✔
9
10	from django.utils import timezone	1✔
11
12	from bpp import const	1✔
13	from bpp.models import LinkDoPBNMixin	1✔
14
15
16	class SentDataManager(models.Manager):	1✔
17	def get_for_rec(self, rec):	1✔
18	return self.get(	1✔
19	object_id=rec.pk, content_type=ContentType.objects.get_for_model(rec)
20	)
21
22	def check_if_needed(self, rec, data: dict):	1✔
23	"""Legacy method - kept for backward compatibility"""
24	try:	×
25	sd = self.get_for_rec(rec)	×
26	except SentData.DoesNotExist:	×
27	return True	×
28
29	if sd.data_sent != data:	×
30	return True	×
31
32	if not sd.uploaded_okay:	×
33	return True	×
34
35	return False	×
36
37	def check_if_upload_needed(self, rec, data: dict):	1✔
38	"""Check if upload needed based on SUCCESSFUL submissions only"""
39	try:	1✔
40	sd = self.get_for_rec(rec)	1✔
41	# Only skip if data matches AND was successfully submitted
42	if (not compare_dicts(sd.data_sent, data)) and sd.submitted_successfully:	1!
43	return False	1✔
44	except SentData.DoesNotExist:	1✔
45	pass	1✔
46	return True	1✔
47
48	def create_or_update_before_upload(self, rec, data: dict):	1✔
49	"""Create or update SentData record before API call"""
50	try:	1✔
51	sd = self.get_for_rec(rec)	1✔
52	# Reset fields for new attempt
53	sd.submitted_successfully = False	×
54	sd.submitted_at = timezone.now()	×
55	sd.uploaded_okay = False	×
56	sd.api_response_status = None	×
57	sd.exception = None	×
58	sd.data_sent = data # Update data if changed	×
59	sd.save()	×
60	return sd	×
61	except SentData.DoesNotExist:	1✔
62	# Create new record if none exists
63	return self.create(	1✔
64	object=rec,
65	data_sent=data,
66	submitted_successfully=False,
67	submitted_at=timezone.now(),
68	uploaded_okay=False,
69	)
70
71	def mark_as_successful(self, rec, pbn_uid_id=None, api_response_status=None):	1✔
72	"""Mark existing record as successful after API call"""
73	sd = self.get_for_rec(rec)	1✔
74	sd.submitted_successfully = True	1✔
75	sd.uploaded_okay = True	1✔
76	sd.pbn_uid_id = pbn_uid_id	1✔
77	sd.api_response_status = api_response_status	1✔
78	sd.exception = None	1✔
79	sd.save()	1✔
80
81	def mark_as_failed(self, rec, exception=None, api_response_status=None):	1✔
82	"""Mark existing record as failed after API call"""
83	sd = self.get_for_rec(rec)	1✔
84	sd.submitted_successfully = False	1✔
85	sd.uploaded_okay = False	1✔
86	sd.exception = str(exception) if exception else None	1✔
87	sd.api_response_status = api_response_status	1✔
88	sd.save()	1✔
89
90	def updated(	1✔
91	self, rec, data: dict, pbn_uid_id=None, uploaded_okay=True, exception=None
92	):
93	"""Legacy method - kept for backward compatibility"""
94	try:	1✔
95	sd = self.get_for_rec(rec)	1✔
96	except SentData.DoesNotExist:	1✔
97	self.create(	1✔
98	object=rec,
99	data_sent=data,
100	uploaded_okay=uploaded_okay,
101	pbn_uid_id=pbn_uid_id,
102	exception=exception,
103	submitted_successfully=uploaded_okay,
104	submitted_at=timezone.now() if uploaded_okay else None,
105	)
106	return	1✔
107
108	sd.data_sent = data	×
109	sd.uploaded_okay = uploaded_okay	×
110	sd.exception = exception	×
111	sd.pbn_uid_id = pbn_uid_id	×
112	sd.submitted_successfully = uploaded_okay	×
113	if uploaded_okay and not sd.submitted_at:	×
114	sd.submitted_at = timezone.now()	×
115	sd.save()	×
116
117	def ids_for_model(self, model):	1✔
118	return self.filter(content_type=ContentType.objects.get_for_model(model))	×
119
120	def bad_uploads(self, model):	1✔
121	return (	×
122	self.ids_for_model(model)
123	.filter(uploaded_okay=False)
124	.values_list("object_id", flat=True)
125	.distinct()
126	)
127
128
129	class SentData(LinkDoPBNMixin, models.Model):	1✔
130	url_do_pbn = const.LINK_PBN_DO_PUBLIKACJI	1✔
131	content_type = models.ForeignKey(	1✔
132	"contenttypes.ContentType", on_delete=models.CASCADE
133	)
134	object_id = models.PositiveIntegerField(db_index=True)	1✔
135
136	object = GenericForeignKey()	1✔
137
138	data_sent = JSONField("Wysłane dane")	1✔
139	last_updated_on = models.DateTimeField("Data operacji", auto_now=True)	1✔
140
141	uploaded_okay = models.BooleanField(	1✔
142	"Wysłano poprawnie", default=True, db_index=True
143	)
144	exception = models.TextField("Kod błędu", max_length=65535, blank=True, null=True)	1✔
145
146	# New fields for success tracking
147	submitted_successfully = models.BooleanField(	1✔
148	"Wysłano pomyślnie",
149	default=False,
150	db_index=True,
151	help_text="True gdy API call zakończył się sukcesem",
152	)
153	submitted_at = models.DateTimeField(	1✔
154	"Data wysyłki",
155	null=True,
156	blank=True,
157	help_text="Kiedy dane zostały wysłane do PBN",
158	)
159	api_response_status = models.TextField(	1✔
160	"Status odpowiedzi API", null=True, blank=True, help_text="Odpowiedź z PBN API"
161	)
162
163	pbn_uid = models.ForeignKey(	1✔
164	"pbn_api.Publication",
165	verbose_name="Publikacja z PBN",
166	blank=True,
167	null=True,
168	on_delete=models.SET_NULL,
169	)
170
171	typ_rekordu = models.CharField(max_length=50, blank=True, null=True)	1✔
172
173	objects = SentDataManager()	1✔
174
175	class Meta:	1✔
176	verbose_name = "Informacja o wysłanych danych"	1✔
177	verbose_name_plural = "Informacje o wysłanych danych"	1✔
178
179	object.verbose_name = "Rekord"	1✔
180
181	def __str__(self):	1✔
182	return (	×
183	f"Informacja o wysłanych do PBN danych dla rekordu ({self.content_type_id},{self.object_id}) "
184	f"z dnia {self.last_updated_on} (status: {'OK' if self.uploaded_okay else 'ERR'})"
185	)
186
187	def link_do_pbn_wartosc_id(self):	1✔
188	return self.pbn_uid_id	×
189
190	def rekord_w_bpp(self):	1✔
191	try:	×
192	return self.object	×
193	except ObjectDoesNotExist:	×
194	pass	×
195
196	def save(	1✔
197	self, force_insert=False, force_update=False, using=None, update_fields=None
198	):
199
200	if update_fields and "data_sent" in update_fields:	1!
201	if self.typ_rekordu != self.data_sent.get("type"):	×
202	update_fields.append("typ_rekordu")	×
203
204	self.typ_rekordu = self.data_sent.get("type")	1✔
205	return super().save(	1✔
206	force_insert=force_insert,
207	force_update=force_update,
208	using=using,
209	update_fields=update_fields,
210	)

iplweb / bpp / #819

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous