8766121206

Committed 20 Apr 2024 03:45PM UTC coverage: 83.956% (-2.9%) from 86.892%

Build # 8766121206

Build Type

push

github

Committed by

andgineer

Commit Message

upgrade reqs

Run Details

2 of 2 new or added lines in 2 files covered. (100.0%)

89 existing lines in 23 files now uncovered.

1057 of 1259 relevant lines covered (83.96%)

0.84 hits per line

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

60.34

/src/cloud_services.py

from typing import Any, Dict, List, Optional

import boto3
from botocore.exceptions import ClientError, NoCredentialsError

import settings
from journaling import log

MAX_MESSAGES = 10  # Amazon limits max number of requested messages
CHARSET = "UTF-8"


def aws_session() -> Any:
    """AWS Session."""
    return boto3

    # Authorization by local credentials. Not necessary if AWS box we run on has appropriate IAM role.
    # return boto3.Session(
    #     aws_access_key_id=settings.config.aws_key_id,
    #     aws_secret_access_key=settings.config.aws_secret_key,
    # )


def ses() -> Any:
    """Amazon simple email service."""
    assert settings.config
    return aws_session().client("ses", region_name=settings.config.aws_region)


def send_email(
    recipients: List[Any], subject: str, html: Optional[str] = None, text: Optional[str] = None
) -> None:
    """Send email via Amazon SES."""
    try:
        message: Dict[str, Dict[str, Any]] = {
            "Body": {},
            "Subject": {
                "Charset": CHARSET,
                "Data": subject,
            },
        }
        if html:
            message["Body"]["Html"] = {
                "Charset": CHARSET,
                "Data": html,
            }
        if text:
            message["Body"]["Text"] = {
                "Charset": CHARSET,
                "Data": text,
            }
        response = ses().send_email(
            Destination={
                "ToAddresses": recipients,
            },
            Message=message,
            Source=settings.config.sender_email,  # type: ignore
            # ConfigurationSetName=CONFIGURATION_SET,
        )
    except NoCredentialsError:
        log.error(
            "Error sending email: Amazon engine unable to locate credentials. May be we are running in dev/test environment."
        )
    except ClientError as e:
        log.error(e.response["Error"]["Message"])
    else:
        log.info("Email sent! Message ID:")
        log.debug(response["MessageId"])


def sqs() -> Any:
    """Amazon simple queue service."""
    assert settings.config
    return aws_session().client("sqs", region_name=settings.config.aws_region)


def queue_url() -> str:
    """URL of the queue we use in our application."""
    assert settings.config
    return sqs().get_queue_url(QueueName=settings.config.aws_queue)["QueueUrl"]  # type: ignore


def queue() -> Any:
    """Specific queue we use in our application."""
    assert settings.config
    return sqs().get_queue_by_name(QueueName=settings.config.aws_queue)  # type: ignore


def queue_messages_list() -> list[Dict[str, Any]]:
    """List of messages in the queue."""
    result = []
    while True:
        messages = queue().receive_messages(
            MaxNumberOfMessages=MAX_MESSAGES,
            VisibilityTimeout=1,
            WaitTimeSeconds=1,
        )
        if len(messages) == 0:
            break
        result.extend(messages)
    return result


def get_queue_message() -> Optional[Dict[str, Any]]:
    """Get one message from the queue."""
    messages = sqs().receive_message(
        QueueUrl=queue_url(),
        MaxNumberOfMessages=1,
        VisibilityTimeout=1,
        WaitTimeSeconds=1,
    )
    if "Messages" not in messages or len(messages["Messages"]) == 0:
        return None
    log.debug(f'Got SQS messages: {messages["Messages"]}')
    return messages["Messages"][0]  # type: ignore


def send_queue_message(body: str) -> None:
    """Send message to the queue."""
    response = sqs().send_message(
        QueueUrl=queue_url(), DelaySeconds=10, MessageAttributes={}, MessageBody=body
    )
    log.debug(f'Sent SQS message "{body}",\n\nresponse: {response}')


def delete_queue_message(message: Dict[str, Any]) -> None:
    """Delete message from the queue."""
    log.debug(f"Delete message {message}")
    sqs().delete_message(
        QueueUrl=queue_url(),
        ReceiptHandle=message["ReceiptHandle"],
    )


if __name__ == "__main__":
    settings.config = settings.ConfigProd()
    send_email(["support@example.com"], "Some subject", text="Some text")

1	from typing import Any, Dict, List, Optional	1✔
2
3	import boto3	1✔
4	from botocore.exceptions import ClientError, NoCredentialsError	1✔
5
6	import settings	1✔
7	from journaling import log	1✔
8
9	MAX_MESSAGES = 10 # Amazon limits max number of requested messages	1✔
10	CHARSET = "UTF-8"	1✔
11
12
13	def aws_session() -> Any:	1✔
14	"""AWS Session."""
UNCOV 15	return boto3	×
16
17	# Authorization by local credentials. Not necessary if AWS box we run on has appropriate IAM role.
18	# return boto3.Session(
19	# aws_access_key_id=settings.config.aws_key_id,
20	# aws_secret_access_key=settings.config.aws_secret_key,
21	# )
22
23
24	def ses() -> Any:	1✔
25	"""Amazon simple email service."""
26	assert settings.config	×
27	return aws_session().client("ses", region_name=settings.config.aws_region)	×
28
29
30	def send_email(	1✔
31	recipients: List[Any], subject: str, html: Optional[str] = None, text: Optional[str] = None
32	) -> None:
33	"""Send email via Amazon SES."""
34	try:	1✔
35	message: Dict[str, Dict[str, Any]] = {	1✔
36	"Body": {},
37	"Subject": {
38	"Charset": CHARSET,
39	"Data": subject,
40	},
41	}
42	if html:	1✔
43	message["Body"]["Html"] = {	1✔
44	"Charset": CHARSET,
45	"Data": html,
46	}
47	if text:	1✔
48	message["Body"]["Text"] = {	1✔
49	"Charset": CHARSET,
50	"Data": text,
51	}
52	response = ses().send_email(	1✔
53	Destination={
54	"ToAddresses": recipients,
55	},
56	Message=message,
57	Source=settings.config.sender_email, # type: ignore
58	# ConfigurationSetName=CONFIGURATION_SET,
59	)
UNCOV 60	except NoCredentialsError:	×
UNCOV 61	log.error(	×
62	"Error sending email: Amazon engine unable to locate credentials. May be we are running in dev/test environment."
63	)
UNCOV 64	except ClientError as e:	×
UNCOV 65	log.error(e.response["Error"]["Message"])	×
66	else:
67	log.info("Email sent! Message ID:")	1✔
68	log.debug(response["MessageId"])	1✔
69
70
71	def sqs() -> Any:	1✔
72	"""Amazon simple queue service."""
UNCOV 73	assert settings.config	×
UNCOV 74	return aws_session().client("sqs", region_name=settings.config.aws_region)	×
75
76
77	def queue_url() -> str:	1✔
78	"""URL of the queue we use in our application."""
UNCOV 79	assert settings.config	×
UNCOV 80	return sqs().get_queue_url(QueueName=settings.config.aws_queue)["QueueUrl"] # type: ignore	×
81
82
83	def queue() -> Any:	1✔
84	"""Specific queue we use in our application."""
UNCOV 85	assert settings.config	×
UNCOV 86	return sqs().get_queue_by_name(QueueName=settings.config.aws_queue) # type: ignore	×
87
88
89	def queue_messages_list() -> list[Dict[str, Any]]:	1✔
90	"""List of messages in the queue."""
UNCOV 91	result = []	×
UNCOV 92	while True:	×
UNCOV 93	messages = queue().receive_messages(	×
94	MaxNumberOfMessages=MAX_MESSAGES,
95	VisibilityTimeout=1,
96	WaitTimeSeconds=1,
97	)
UNCOV 98	if len(messages) == 0:	×
UNCOV 99	break	×
UNCOV 100	result.extend(messages)	×
UNCOV 101	return result	×
102
103
104	def get_queue_message() -> Optional[Dict[str, Any]]:	1✔
105	"""Get one message from the queue."""
106	messages = sqs().receive_message(	1✔
107	QueueUrl=queue_url(),
108	MaxNumberOfMessages=1,
109	VisibilityTimeout=1,
110	WaitTimeSeconds=1,
111	)
112	if "Messages" not in messages or len(messages["Messages"]) == 0:	1✔
UNCOV 113	return None	×
114	log.debug(f'Got SQS messages: {messages["Messages"]}')	1✔
115	return messages["Messages"][0] # type: ignore	1✔
116
117
118	def send_queue_message(body: str) -> None:	1✔
119	"""Send message to the queue."""
120	response = sqs().send_message(	1✔
121	QueueUrl=queue_url(), DelaySeconds=10, MessageAttributes={}, MessageBody=body
122	)
123	log.debug(f'Sent SQS message "{body}",\n\nresponse: {response}')	1✔
124
125
126	def delete_queue_message(message: Dict[str, Any]) -> None:	1✔
127	"""Delete message from the queue."""
128	log.debug(f"Delete message {message}")	1✔
129	sqs().delete_message(	1✔
130	QueueUrl=queue_url(),
131	ReceiptHandle=message["ReceiptHandle"],
132	)
133
134
135	if __name__ == "__main__":	1✔
UNCOV 136	settings.config = settings.ConfigProd()	×
UNCOV 137	send_email(["support@example.com"], "Some subject", text="Some text")	×

andgineer / api-db-prototype / 8766121206

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