954eab8e-e29d-4864-88d5-f814d498649d

Committed 25 Aug 2025 07:53PM UTC coverage: 42.328% (-1.0%) from 43.284%

Build # 954eab8e-e29d-4864-88d5-f814d498649d

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202508.1210'

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

1298 existing lines in 55 files now uncovered.

16981 of 40118 relevant lines covered (42.33%)

0.77 hits per line

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

8.7

src/pbn_api/tasks.py

from django.core.management import call_command

from long_running.util import wait_for_object
from pbn_api.models.queue import SendStatus

from django_bpp.celery_tasks import app


@app.task
def task_sprobuj_wyslac_do_pbn(pk):
    from pbn_api.models import PBN_Export_Queue

    p = wait_for_object(PBN_Export_Queue, pk)
    res = p.send_to_pbn()

    match res:
        case SendStatus.RETRY_LATER:
            task_sprobuj_wyslac_do_pbn.apply_async(args=[pk], countdown=5 * 60)

        case SendStatus.RETRY_SOON:
            # np. 423 Locked
            task_sprobuj_wyslac_do_pbn.apply_async(args=[pk], countdown=60)

        case SendStatus.RETRY_MUCH_LATER:
            # PraceSerwisoweException
            task_sprobuj_wyslac_do_pbn.apply_async(args=[pk], countdown=60 * 60 * 3)

        case (
            SendStatus.FINISHED_ERROR
            | SendStatus.FINISHED_OKAY
            | SendStatus.RETRY_AFTER_USER_AUTHORISED
        ):
            return

        case _:
            raise NotImplementedError(
                f"Return status for background send to PBN not supported {res=}"
            )


@app.task
def kolejka_wyczysc_wpisy_bez_rekordow():
    from pbn_api.models import PBN_Export_Queue

    for elem in PBN_Export_Queue.objects.all():
        if not elem.check_if_record_still_exists():
            elem.delete()


@app.task
def kolejka_ponow_wysylke_prac_po_zalogowaniu(pk):
    from pbn_api.models import PBN_Export_Queue

    # Użytkownik o ID pk zalogował się.
    # Odśwież do wysyłki prace które były jego po zalogowaniu
    for elem in PBN_Export_Queue.objects.filter(
        retry_after_user_authorised=True, zamowil_id=pk, wysylke_zakonczono=None
    ):
        task_sprobuj_wyslac_do_pbn.delay(elem.pk)

    # ... ale i odświez prace wszystkich użytkowników, którzy mają jego konto
    # jako konto do wysyłki:
    for elem in PBN_Export_Queue.objects.filter(
        retry_after_user_authorised=True,
        zamowil__przedstawiaj_w_pbn_jako_id=pk,
        wysylke_zakonczono=None,
    ):
        task_sprobuj_wyslac_do_pbn.delay(elem.pk)


@app.task
def download_institution_publications(user_id):
    """
    Download institution publications using PBN API management commands.
    Uses database-based locking to ensure only one instance runs at a time.

    Args:
        user_id: ID of the user initiating the download (must have valid PBN token)
    """
    from django.db import transaction

    from komparator_pbn.models import PbnDownloadTask

    from django.utils import timezone

    from bpp.models.profile import BppUser

    # Check if there's already a running task
    running_task = PbnDownloadTask.objects.filter(status="running").first()
    if running_task:
        raise ValueError(
            "Another download task is already running. Please wait for it to complete."
        )

    task_record = None
    try:
        user = BppUser.objects.get(pk=user_id)
        pbn_user = user.get_pbn_user()

        if not pbn_user.pbn_token:
            raise ValueError(
                f"User {user.username} is not authorized in PBN (no pbn_token)"
            )

        if not pbn_user.pbn_token_possibly_valid():
            raise ValueError(
                f"User {user.username} has an invalid or expired PBN token"
            )

        # Create task record with atomic transaction to prevent race conditions
        with transaction.atomic():
            # Double-check there's no running task within the transaction
            if PbnDownloadTask.objects.filter(status="running").exists():
                raise ValueError(
                    "Another download task is already running. Please wait for it to complete."
                )

            task_record = PbnDownloadTask.objects.create(
                user=user,
                status="running",
                started_at=timezone.now(),
                current_step="Inicjalizacja pobierania...",
                progress_percentage=0,
            )

        # Monkey-patch tqdm to capture progress and save to database
        from threading import Lock

        import tqdm

        original_tqdm_init = tqdm.tqdm.__init__
        original_tqdm_update = tqdm.tqdm.update
        original_tqdm_close = tqdm.tqdm.close
        progress_lock = Lock()

        def patched_tqdm_init(
            self, iterable=None, desc=None, total=None, *args, **kwargs
        ):
            # Store task record reference for progress updates
            self._task_record = task_record
            self._desc = desc or ""
            # Determine phase offset based on current step
            if task_record.current_step and "Faza 1" in task_record.current_step:
                self._phase_offset = 10  # First command: 10-45%
            elif task_record.current_step and "Faza 2" in task_record.current_step:
                self._phase_offset = 50  # Second command: 50-85%
            else:
                self._phase_offset = 10  # Default

            # Call original init
            original_tqdm_init(self, iterable, desc, total, *args, **kwargs)

        def patched_tqdm_update(self, n=1):
            result = original_tqdm_update(self, n)

            # Update database progress
            if hasattr(self, "_task_record") and self._task_record:
                with progress_lock:
                    try:
                        # Calculate progress percentage (10-45% for first command, 50-85% for second)
                        if self.total and self.total > 0:
                            command_progress = (
                                self.n / self.total
                            ) * 35  # Each command gets 35% of total progress
                            total_progress = self._phase_offset + command_progress

                            self._task_record.progress_percentage = min(
                                90, max(10, int(total_progress))
                            )

                            # Update current step with more detail
                            if self._desc:
                                if "publikacj" in self._desc.lower():
                                    self._task_record.publications_processed = self.n
                                    if self.total:
                                        self._task_record.total_publications = (
                                            self.total
                                        )
                                elif (
                                    "oświadczen" in self._desc.lower()
                                    or "statement" in self._desc.lower()
                                ):
                                    self._task_record.statements_processed = self.n
                                    if self.total:
                                        self._task_record.total_statements = self.total

                                # Update current step with progress details
                                progress_text = f"{self._desc} ({self.n}"
                                if self.total:
                                    progress_text += f"/{self.total}"
                                progress_text += ")"
                                self._task_record.current_step = progress_text

                            self._task_record.save()
                    except Exception:
                        pass  # Don't let database errors break the download

            return result

        def patched_tqdm_close(self):
            result = original_tqdm_close(self)

            # Final update when tqdm closes
            if hasattr(self, "_task_record") and self._task_record:
                with progress_lock:
                    try:
                        if self.total and self.n >= self.total:
                            # Command completed, update to end of phase
                            total_progress = self._phase_offset + 35
                            self._task_record.progress_percentage = min(
                                90, int(total_progress)
                            )
                            self._task_record.save()
                    except Exception:
                        pass

            return result

        # Apply monkey patches
        tqdm.tqdm.__init__ = patched_tqdm_init
        tqdm.tqdm.update = patched_tqdm_update
        tqdm.tqdm.close = patched_tqdm_close

        try:
            # Run the management commands with progress reporting
            task_record.current_step = "Pobieranie publikacji instytucji (Faza 1/2)"
            task_record.progress_percentage = 10
            task_record.save()

            call_command(
                "pbn_pobierz_publikacje_z_instytucji_v2", user_token=pbn_user.pbn_token
            )

            task_record.current_step = "Pobieranie oświadczeń i publikacji (Faza 2/2)"
            task_record.progress_percentage = 50
            task_record.save()

            call_command(
                "pbn_pobierz_oswiadczenia_i_publikacje_v1",
                user_token=pbn_user.pbn_token,
            )

            task_record.current_step = "Finalizowanie pobierania..."
            task_record.progress_percentage = 90
            task_record.save()

        finally:
            # Restore original tqdm methods
            tqdm.tqdm.__init__ = original_tqdm_init
            tqdm.tqdm.update = original_tqdm_update
            tqdm.tqdm.close = original_tqdm_close

        # Mark as completed
        task_record.status = "completed"
        task_record.current_step = "Pobieranie zakończone pomyślnie"
        task_record.progress_percentage = 100
        task_record.completed_at = timezone.now()
        task_record.save()

    except Exception as e:
        # Record the error
        if task_record:
            task_record.status = "failed"
            task_record.error_message = str(e)
            task_record.completed_at = timezone.now()
            task_record.save()
        raise

1	from django.core.management import call_command	2✔
2
3	from long_running.util import wait_for_object	2✔
4	from pbn_api.models.queue import SendStatus	2✔
5
6	from django_bpp.celery_tasks import app	2✔
7
8
9	@app.task	2✔
10	def task_sprobuj_wyslac_do_pbn(pk):	2✔
11	from pbn_api.models import PBN_Export_Queue	×
12
UNCOV 13	p = wait_for_object(PBN_Export_Queue, pk)	×
14	res = p.send_to_pbn()	×
15
16	match res:	×
UNCOV 17	case SendStatus.RETRY_LATER:	×
18	task_sprobuj_wyslac_do_pbn.apply_async(args=[pk], countdown=5 * 60)	×
19
20	case SendStatus.RETRY_SOON:	×
21	# np. 423 Locked
22	task_sprobuj_wyslac_do_pbn.apply_async(args=[pk], countdown=60)	×
23
24	case SendStatus.RETRY_MUCH_LATER:	×
25	# PraceSerwisoweException
26	task_sprobuj_wyslac_do_pbn.apply_async(args=[pk], countdown=60 * 60 * 3)	×
27
UNCOV 28	case (	×
29	SendStatus.FINISHED_ERROR
30	\| SendStatus.FINISHED_OKAY
31	\| SendStatus.RETRY_AFTER_USER_AUTHORISED
32	):
33	return	×
34
UNCOV 35	case _:	×
UNCOV 36	raise NotImplementedError(	×
37	f"Return status for background send to PBN not supported {res=}"
38	)
39
40
41	@app.task	2✔
42	def kolejka_wyczysc_wpisy_bez_rekordow():	2✔
43	from pbn_api.models import PBN_Export_Queue	×
44
45	for elem in PBN_Export_Queue.objects.all():	×
UNCOV 46	if not elem.check_if_record_still_exists():	×
UNCOV 47	elem.delete()	×
48
49
50	@app.task	2✔
51	def kolejka_ponow_wysylke_prac_po_zalogowaniu(pk):	2✔
UNCOV 52	from pbn_api.models import PBN_Export_Queue	×
53
54	# Użytkownik o ID pk zalogował się.
55	# Odśwież do wysyłki prace które były jego po zalogowaniu
UNCOV 56	for elem in PBN_Export_Queue.objects.filter(	×
57	retry_after_user_authorised=True, zamowil_id=pk, wysylke_zakonczono=None
58	):
UNCOV 59	task_sprobuj_wyslac_do_pbn.delay(elem.pk)	×
60
61	# ... ale i odświez prace wszystkich użytkowników, którzy mają jego konto
62	# jako konto do wysyłki:
UNCOV 63	for elem in PBN_Export_Queue.objects.filter(	×
64	retry_after_user_authorised=True,
65	zamowil__przedstawiaj_w_pbn_jako_id=pk,
66	wysylke_zakonczono=None,
67	):
UNCOV 68	task_sprobuj_wyslac_do_pbn.delay(elem.pk)	×
69
70
71	@app.task	2✔
72	def download_institution_publications(user_id):	2✔
73	"""
74	Download institution publications using PBN API management commands.
75	Uses database-based locking to ensure only one instance runs at a time.
76
77	Args:
78	user_id: ID of the user initiating the download (must have valid PBN token)
79	"""
UNCOV 80	from django.db import transaction	×
81
UNCOV 82	from komparator_pbn.models import PbnDownloadTask	×
83
UNCOV 84	from django.utils import timezone	×
85
UNCOV 86	from bpp.models.profile import BppUser	×
87
88	# Check if there's already a running task
UNCOV 89	running_task = PbnDownloadTask.objects.filter(status="running").first()	×
UNCOV 90	if running_task:	×
UNCOV 91	raise ValueError(	×
92	"Another download task is already running. Please wait for it to complete."
93	)
94
UNCOV 95	task_record = None	×
UNCOV 96	try:	×
UNCOV 97	user = BppUser.objects.get(pk=user_id)	×
UNCOV 98	pbn_user = user.get_pbn_user()	×
99
UNCOV 100	if not pbn_user.pbn_token:	×
UNCOV 101	raise ValueError(	×
102	f"User {user.username} is not authorized in PBN (no pbn_token)"
103	)
104
UNCOV 105	if not pbn_user.pbn_token_possibly_valid():	×
UNCOV 106	raise ValueError(	×
107	f"User {user.username} has an invalid or expired PBN token"
108	)
109
110	# Create task record with atomic transaction to prevent race conditions
UNCOV 111	with transaction.atomic():	×
112	# Double-check there's no running task within the transaction
UNCOV 113	if PbnDownloadTask.objects.filter(status="running").exists():	×
UNCOV 114	raise ValueError(	×
115	"Another download task is already running. Please wait for it to complete."
116	)
117
UNCOV 118	task_record = PbnDownloadTask.objects.create(	×
119	user=user,
120	status="running",
121	started_at=timezone.now(),
122	current_step="Inicjalizacja pobierania...",
123	progress_percentage=0,
124	)
125
126	# Monkey-patch tqdm to capture progress and save to database
UNCOV 127	from threading import Lock	×
128
UNCOV 129	import tqdm	×
130
UNCOV 131	original_tqdm_init = tqdm.tqdm.__init__	×
UNCOV 132	original_tqdm_update = tqdm.tqdm.update	×
UNCOV 133	original_tqdm_close = tqdm.tqdm.close	×
UNCOV 134	progress_lock = Lock()	×
135
UNCOV 136	def patched_tqdm_init(	×
137	self, iterable=None, desc=None, total=None, args, *kwargs
138	):
139	# Store task record reference for progress updates
UNCOV 140	self._task_record = task_record	×
UNCOV 141	self._desc = desc or ""	×
142	# Determine phase offset based on current step
UNCOV 143	if task_record.current_step and "Faza 1" in task_record.current_step:	×
UNCOV 144	self._phase_offset = 10 # First command: 10-45%	×
UNCOV 145	elif task_record.current_step and "Faza 2" in task_record.current_step:	×
UNCOV 146	self._phase_offset = 50 # Second command: 50-85%	×
147	else:
UNCOV 148	self._phase_offset = 10 # Default	×
149
150	# Call original init
UNCOV 151	original_tqdm_init(self, iterable, desc, total, args, *kwargs)	×
152
UNCOV 153	def patched_tqdm_update(self, n=1):	×
UNCOV 154	result = original_tqdm_update(self, n)	×
155
156	# Update database progress
UNCOV 157	if hasattr(self, "_task_record") and self._task_record:	×
UNCOV 158	with progress_lock:	×
UNCOV 159	try:	×
160	# Calculate progress percentage (10-45% for first command, 50-85% for second)
UNCOV 161	if self.total and self.total > 0:	×
UNCOV 162	command_progress = (	×
163	self.n / self.total
164	) * 35 # Each command gets 35% of total progress
UNCOV 165	total_progress = self._phase_offset + command_progress	×
166
UNCOV 167	self._task_record.progress_percentage = min(	×
168	90, max(10, int(total_progress))
169	)
170
171	# Update current step with more detail
UNCOV 172	if self._desc:	×
UNCOV 173	if "publikacj" in self._desc.lower():	×
UNCOV 174	self._task_record.publications_processed = self.n	×
UNCOV 175	if self.total:	×
UNCOV 176	self._task_record.total_publications = (	×
177	self.total
178	)
UNCOV 179	elif (	×
180	"oświadczen" in self._desc.lower()
181	or "statement" in self._desc.lower()
182	):
UNCOV 183	self._task_record.statements_processed = self.n	×
UNCOV 184	if self.total:	×
UNCOV 185	self._task_record.total_statements = self.total	×
186
187	# Update current step with progress details
UNCOV 188	progress_text = f"{self._desc} ({self.n}"	×
UNCOV 189	if self.total:	×
UNCOV 190	progress_text += f"/{self.total}"	×
UNCOV 191	progress_text += ")"	×
UNCOV 192	self._task_record.current_step = progress_text	×
193
UNCOV 194	self._task_record.save()	×
UNCOV 195	except Exception:	×
UNCOV 196	pass # Don't let database errors break the download	×
197
UNCOV 198	return result	×
199
UNCOV 200	def patched_tqdm_close(self):	×
UNCOV 201	result = original_tqdm_close(self)	×
202
203	# Final update when tqdm closes
UNCOV 204	if hasattr(self, "_task_record") and self._task_record:	×
UNCOV 205	with progress_lock:	×
UNCOV 206	try:	×
UNCOV 207	if self.total and self.n >= self.total:	×
208	# Command completed, update to end of phase
UNCOV 209	total_progress = self._phase_offset + 35	×
UNCOV 210	self._task_record.progress_percentage = min(	×
211	90, int(total_progress)
212	)
UNCOV 213	self._task_record.save()	×
UNCOV 214	except Exception:	×
UNCOV 215	pass	×
216
UNCOV 217	return result	×
218
219	# Apply monkey patches
UNCOV 220	tqdm.tqdm.__init__ = patched_tqdm_init	×
UNCOV 221	tqdm.tqdm.update = patched_tqdm_update	×
UNCOV 222	tqdm.tqdm.close = patched_tqdm_close	×
223
UNCOV 224	try:	×
225	# Run the management commands with progress reporting
UNCOV 226	task_record.current_step = "Pobieranie publikacji instytucji (Faza 1/2)"	×
UNCOV 227	task_record.progress_percentage = 10	×
UNCOV 228	task_record.save()	×
229
UNCOV 230	call_command(	×
231	"pbn_pobierz_publikacje_z_instytucji_v2", user_token=pbn_user.pbn_token
232	)
233
UNCOV 234	task_record.current_step = "Pobieranie oświadczeń i publikacji (Faza 2/2)"	×
UNCOV 235	task_record.progress_percentage = 50	×
UNCOV 236	task_record.save()	×
237
UNCOV 238	call_command(	×
239	"pbn_pobierz_oswiadczenia_i_publikacje_v1",
240	user_token=pbn_user.pbn_token,
241	)
242
UNCOV 243	task_record.current_step = "Finalizowanie pobierania..."	×
UNCOV 244	task_record.progress_percentage = 90	×
UNCOV 245	task_record.save()	×
246
247	finally:
248	# Restore original tqdm methods
UNCOV 249	tqdm.tqdm.__init__ = original_tqdm_init	×
UNCOV 250	tqdm.tqdm.update = original_tqdm_update	×
UNCOV 251	tqdm.tqdm.close = original_tqdm_close	×
252
253	# Mark as completed
UNCOV 254	task_record.status = "completed"	×
UNCOV 255	task_record.current_step = "Pobieranie zakończone pomyślnie"	×
UNCOV 256	task_record.progress_percentage = 100	×
UNCOV 257	task_record.completed_at = timezone.now()	×
UNCOV 258	task_record.save()	×
259
UNCOV 260	except Exception as e:	×
261	# Record the error
UNCOV 262	if task_record:	×
UNCOV 263	task_record.status = "failed"	×
UNCOV 264	task_record.error_message = str(e)	×
UNCOV 265	task_record.completed_at = timezone.now()	×
UNCOV 266	task_record.save()	×
UNCOV 267	raise	×

iplweb / bpp / 954eab8e-e29d-4864-88d5-f814d498649d

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