certified-builder-notification-py/lambda_function.py at master · PythonFloripa/certified-builder-notification-py · GitHub

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
65
66
67
68
69
70
71
72
73
74
importlogging
fromdatetimeimportdatetime
importjson
fromsrc.infrastructure.container.dependency_containerimportcontainer
fromsrc.domain.dto.certificate_notificationimportCertificateNotificationBatch, CertificateNotificationResponse
fromsrc.application.process_certificate_notificationimportProcessCertificateNotification
fromsrc.domain.dto.processed_certificate_notificationimportProcessedCertificateNotification
fromsrc.application.certificate_notification_tech_floripaimportCertificateNotificationTechFloripa
fromsrc.domain.dto.tech_floripa_notificationimportTechFloripaNotification


logger=logging.getLogger(__name__)
logger.setLevel(logging.INFO)


defget_notifications_from_event(event) ->CertificateNotificationBatch:
"""
 Extrai e valida as notificações de certificado do evento SQS.

 Args:
 event: Evento SQS contendo as notificações

 Returns:
 CertificateNotificationBatch: Batch de notificações validado
 """
body=event.get('Records')[0].get('body')
ifisinstance(body, str):
body=json.loads(body)

# Se o body é uma lista, encapsula na estrutura esperada pelo modelo
ifisinstance(body, list):
notifications_data= {"notifications": body}
else:
# Se já é um objeto, usa diretamente
notifications_data=body
logger.info(f"Notificações extraídas do evento: {notifications_data}")
returnCertificateNotificationBatch.model_validate(notifications_data)


deflambda_handler(event, context):
try:
process_certificate_notification: ProcessCertificateNotification=container.get('process_certificate_notification')
certificate_notification_tech_floripa: CertificateNotificationTechFloripa=container.get('certificate_notification_tech_floripa')

result: ProcessedCertificateNotification=process_certificate_notification.execute(
get_notifications_from_event(event)
 )

notifications: TechFloripaNotification=TechFloripaNotification.from_certificates(result.updated_certificates)
ifcertificate_notification_tech_floripa.send_notification(notifications):
logger.info(f"Notificação enviada com sucesso: {result.updated_certificates[0].product_id}")
return {
"statusCode": 200,
"body": {
"message": "Notificação enviada com sucesso",
 }
 }
else:
logger.error(f"Erro ao enviar notificação: {result.updated_certificates[0].product_id}")
return {
"statusCode": 500,
"message": "Erro ao enviar notificação - "+result.updated_certificates[0].product_id
 }
exceptExceptionase:
logger.error(f"Erro ao processar notificação de certificado: {str(e)}")
return {
"statusCode": 500,
"body": json.dumps({
"message": "Erro ao processar notificação de certificado",
"error": str(e)
 })
 }