#820

Committed 19 Oct 2025 06:59PM UTC coverage: 65.093% (+5.3%) from 59.791%

Build # #820

Build Type

push

coveralls-python

Committed by Michał Pasternak

Commit Message

Fixes

Run Details

4215 of 9430 branches covered (44.7%)

Branch coverage included in aggregate %.

27562 of 39388 relevant lines covered (69.98%)

0.7 hits per line

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

69.67

src/pbn_api/models/sentdata.py

from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.db import models
from django.db.models import JSONField
from django.utils import timezone

from bpp import const
from bpp.models import LinkDoPBNMixin
from pbn_api.utils import compare_dicts


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

iplweb / bpp / #820

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