import json import boto3 sqs_client boto3 client sqs us-east-1 queue_u

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import json
import boto3
sqs_client = boto3.client('sqs', 'us-east-1')
queue_url= f'https://sqs.us-east-1.amazonaws.com/641113671333/lesson1_vitalii_senchenko'\
def add_messages_to_queue(amount):
for item in range(amount):
sqs_client.send_message(
QueueUrl=queue_url,
MessageBody=f'hi item {item}'
)
def read_from_queue_to_file():
messages = []
while True:
try:
message = sqs_client.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, AttributeNames=['All'])
messages.append(message['Messages'][0]['Body'])
messages.append('\n')
sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=message['Messages'][0]['ReceiptHandle'])
except KeyError:
break
with open ('sqs_data.txt', 'w') as file:
file.writelines(messages)
print(messages)
def create_file_based_on_json_message(amount):
for item in range(amount):
sqs_client.send_message(QueueUrl=queue_url,
MessageBody=f'{{"name": "{item}_new_name.txt", "data": "Number {item}"}}')
while True:
try:
message = sqs_client.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, AttributeNames=['All'])
with open (f"{json.loads(message['Messages'][0]['Body'])['name']}", 'w') as file:
file.writelines(json.loads(message['Messages'][0]['Body'])['data'])
sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=message['Messages'][0]['ReceiptHandle'])
except KeyError:
break
def create_file_based_on_json_message_to_s3(amount):
for item in range(amount):
sqs_client.send_message(QueueUrl=queue_url,
MessageBody=f'{{"s3path": "s3://edge-research-tmp/vitalii.senchenko/file{item}.txt", "data": "Number {item}"}}')
s3_client = boto3.client('s3')
while True:
try:
message = sqs_client.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, AttributeNames=['All'])
full_path = json.loads(message['Messages'][0]['Body'])['s3path']
bucket_name = full_path.split('/')
sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=message['Messages'][0]['ReceiptHandle'])
s3_client.put_object(Bucket=bucket_name[2], Key='/'.join(bucket_name[3:]), Body=json.loads(message['Messages'][0]['Body'])['data'])
except KeyError:
break
add_messages_to_queue(10)
read_from_queue_to_file()
create_file_based_on_json_message_to_s3(3)
create_file_based_on_json_message(1)