2277163c-118e-4992-9881-c04b22beeba1

Committed 12 Jan 2024 03:56PM CUT coverage: 73.584%. Remained the same

Build # 2277163c-118e-4992-9881-c04b22beeba1

Build Type

push

circleci

Committed by

jwhitlock

Commit Message

Switch from pycopg2 2.9.9 to psycopg 3.1.17

Django 4.2 supports psycopg 3, a new implementation of the PostgreSQL
adapter. It recommends it for new projects, and says psycopg2 may be
deprecated in the future.

Run Details

1968 of 2925 branches covered (0.0%)

Branch coverage included in aggregate %.

6319 of 8337 relevant lines covered (75.79%)

20.08 hits per line

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

0.0

/emails/management/commands/process_delayed_emails_from_sqs.py

import json
import logging
import sys
import time

import boto3
from botocore.exceptions import ClientError

from django.conf import settings
from django.core.management.base import BaseCommand

from emails.views import _sns_inbound_logic, validate_sns_arn_and_type, verify_from_sns
from emails.utils import incr_if_enabled


logger = logging.getLogger("events")
info_logger = logging.getLogger("eventsinfo")


def _verify_and_run_sns_inbound_on_message(message):
    incr_if_enabled("rerun_message_from_sqs", 1)
    json_body = json.loads(message.body)
    verified_json_body = verify_from_sns(json_body)
    topic_arn = verified_json_body["TopicArn"]
    message_type = verified_json_body["Type"]
    validate_sns_arn_and_type(topic_arn, message_type)
    try:
        _sns_inbound_logic(topic_arn, message_type, verified_json_body)
        info_logger.info(f"processed sqs message ID: {message.message_id}")
    except ClientError as e:
        incr_if_enabled("rerun_message_from_sqs_error", 1)
        logger.error("sqs_client_error: ", extra=e.response["Error"])
        temp_errors = ["throttling", "pause"]
        lower_error_code = e.response["Error"]["Code"].lower()
        if any(temp_error in lower_error_code for temp_error in temp_errors):
            incr_if_enabled("rerun_message_from_sqs_temp_error", 1)
            logger.error(
                '"temporary" error, sleeping for 1s: ', extra=e.response["Error"]
            )
            time.sleep(1)
            try:
                _sns_inbound_logic(topic_arn, message_type, verified_json_body)
                info_logger.info(f"processed sqs message ID: {message.message_id}")
            except ClientError as e:
                incr_if_enabled("rerun_message_from_sqs_error", 1)
                logger.error("sqs_client_error: ", extra=e.response["Error"])


class Command(BaseCommand):
    help = "Fetches messages from SQS dead-letter queue and processes them."

    def handle(self, *args, **options):
        self.exit_code = 0
        try:
            sqs_client = boto3.resource("sqs", region_name=settings.AWS_REGION)
            dl_queue = sqs_client.Queue(settings.AWS_SQS_QUEUE_URL)
        except ClientError as e:
            logger.error("sqs_client_error: ", extra=e.response["Error"])
            self.exit_code = 1
            sys.exit(self.exit_code)

        messages = dl_queue.receive_messages(MaxNumberOfMessages=10, WaitTimeSeconds=1)
        while len(messages) > 0:
            for message in messages:
                try:
                    _verify_and_run_sns_inbound_on_message(message)
                except:
                    exc_type, _, _ = sys.exc_info()
                    logger.exception(f"dlq_processing_error_{exc_type}")
                finally:
                    message.delete()
            messages = dl_queue.receive_messages(
                MaxNumberOfMessages=10, WaitTimeSeconds=1
            )
        sys.exit(self.exit_code)

1	import json	×
2	import logging	×
3	import sys	×
4	import time	×
5
6	import boto3	×
7	from botocore.exceptions import ClientError	×
8
9	from django.conf import settings	×
10	from django.core.management.base import BaseCommand	×
11
12	from emails.views import _sns_inbound_logic, validate_sns_arn_and_type, verify_from_sns	×
13	from emails.utils import incr_if_enabled	×
14
15
16	logger = logging.getLogger("events")	×
17	info_logger = logging.getLogger("eventsinfo")	×
18
19
20	def _verify_and_run_sns_inbound_on_message(message):	×
21	incr_if_enabled("rerun_message_from_sqs", 1)	×
22	json_body = json.loads(message.body)	×
23	verified_json_body = verify_from_sns(json_body)	×
24	topic_arn = verified_json_body["TopicArn"]	×
25	message_type = verified_json_body["Type"]	×
26	validate_sns_arn_and_type(topic_arn, message_type)	×
27	try:	×
28	_sns_inbound_logic(topic_arn, message_type, verified_json_body)	×
29	info_logger.info(f"processed sqs message ID: {message.message_id}")	×
30	except ClientError as e:	×
31	incr_if_enabled("rerun_message_from_sqs_error", 1)	×
32	logger.error("sqs_client_error: ", extra=e.response["Error"])	×
33	temp_errors = ["throttling", "pause"]	×
34	lower_error_code = e.response["Error"]["Code"].lower()	×
35	if any(temp_error in lower_error_code for temp_error in temp_errors):	×
36	incr_if_enabled("rerun_message_from_sqs_temp_error", 1)	×
37	logger.error(	×
38	'"temporary" error, sleeping for 1s: ', extra=e.response["Error"]
39	)
40	time.sleep(1)	×
41	try:	×
42	_sns_inbound_logic(topic_arn, message_type, verified_json_body)	×
43	info_logger.info(f"processed sqs message ID: {message.message_id}")	×
44	except ClientError as e:	×
45	incr_if_enabled("rerun_message_from_sqs_error", 1)	×
46	logger.error("sqs_client_error: ", extra=e.response["Error"])	×
47
48
49	class Command(BaseCommand):	×
50	help = "Fetches messages from SQS dead-letter queue and processes them."	×
51
52	def handle(self, args, *options):	×
53	self.exit_code = 0	×
54	try:	×
55	sqs_client = boto3.resource("sqs", region_name=settings.AWS_REGION)	×
56	dl_queue = sqs_client.Queue(settings.AWS_SQS_QUEUE_URL)	×
57	except ClientError as e:	×
58	logger.error("sqs_client_error: ", extra=e.response["Error"])	×
59	self.exit_code = 1	×
60	sys.exit(self.exit_code)	×
61
62	messages = dl_queue.receive_messages(MaxNumberOfMessages=10, WaitTimeSeconds=1)	×
63	while len(messages) > 0:	×
64	for message in messages:	×
65	try:	×
66	_verify_and_run_sns_inbound_on_message(message)	×
67	except:	×
68	exc_type, _, _ = sys.exc_info()	×
69	logger.exception(f"dlq_processing_error_{exc_type}")	×
70	finally:
71	message.delete()	×
72	messages = dl_queue.receive_messages(	×
73	MaxNumberOfMessages=10, WaitTimeSeconds=1
74	)
75	sys.exit(self.exit_code)	×

mozilla / fx-private-relay / 2277163c-118e-4992-9881-c04b22beeba1

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