add cert_expires_at

enrollments-events/app/events/issue_cert.py

@@ -1,5 +1,6 @@
 import json
 from datetime import datetime, timedelta
+from typing import NotRequired, TypedDict
 
 import requests
 from aws_lambda_powertools import Logger
@@ -21,8 +22,7 @@ from config import (
 )
 
 logger = Logger(__name__)
-enrollment_layer = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
-course_layer = DynamoDBPersistenceLayer(COURSE_TABLE, dynamodb_client)
+dyn = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
 
 
 @event_source(data_class=EventBridgeEvent)
@@ -32,10 +32,11 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
     now_ = now()
     enrollment_id = new_image['id']
     course_id = new_image['course']['id']
-    cert = course_layer.collection.get_item(
+    cert = dyn.collection.get_item(
         KeyPair(
             pk=course_id,
             sk=SortKey('0', path_spec='cert', rename_key='cert'),
+            table_name=COURSE_TABLE,
         ),
         raise_on_error=False,
         default=None,
@@ -49,76 +50,36 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
     started_at: datetime = fromisoformat(new_image['started_at'])  # type: ignore
     completed_at: datetime = fromisoformat(new_image['completed_at'])  # type: ignore
     # Certificate may have no expiration
-    cert_expires_at = (
+    expires_at = (
         completed_at + timedelta(days=int(cert['exp_interval']))
         if cert.get('exp_interval', 0) > 0
         else None
     )
+    s3_uri = _gen_cert(
+        enrollment_id,
+        cert=cert,
+        user=new_image['user'],
+        score=new_image['score'],
+        started_at=started_at,
+        completed_at=completed_at,
+        expires_at=expires_at,
+    )
 
-    try:
-        if 's3_uri' not in cert:
-            raise ValueError('Template URI is missing')
+    update_expr = 'SET cert = :cert, updated_at = :now'
+    expr_attr_values = {
+        ':now': now_,
+        ':cert': {'issued_at': now_} | ({'s3_uri': s3_uri} if s3_uri else {}),
+    }
 
-        # Send template URI and data to Paperforge API to generate a PDF
-        r = requests.post(
-            PAPERFORGE_API,
-            data=json.dumps(
-                {
-                    'template_uri': cert['s3_uri'],
-                    'sign_uri': ESIGN_URI,
-                    'args': {
-                        'name': new_image['user']['name'],
-                        'cpf': _cpffmt(new_image['user']['cpf']),
-                        'score': new_image['score'],
-                        'started_at': started_at.strftime('%d/%m/%Y'),
-                        'completed_at': completed_at.strftime('%d/%m/%Y'),
-                        'today': _datefmt(now_),
-                        'year': now_.strftime('%Y'),
-                    }
-                    | (
-                        {'expires_at': cert_expires_at.strftime('%d/%m/%Y')}
-                        if cert_expires_at
-                        else {}
-                    ),
-                },
-            ),
-            timeout=5,
-        )
-        r.raise_for_status()
+    if expires_at:
+        update_expr = 'SET cert = :cert, cert_expires_at = :cert_expires_at, \
+                           updated_at = :now'
+        expr_attr_values[':cert_expires_at'] = expires_at
 
-        object_key = f'certs/{enrollment_id}.pdf'
-        s3_uri = f's3://{BUCKET_NAME}/{object_key}'
-
-        s3_client.put_object(
-            Bucket=BUCKET_NAME,
-            Key=object_key,
-            Body=r.content,
-            ContentType='application/pdf',
-        )
-
-        logger.debug(f'PDF uploaded successfully to {s3_uri}')
-    except ValueError as exc:
-        # PDF generation fails if template URI is missing
-        s3_uri = None
-        logger.exception(exc)
-    except requests.exceptions.RequestException as exc:
-        logger.exception(exc)
-        raise
-
-    return enrollment_layer.update_item(
-        key=KeyPair(
-            pk=enrollment_id,
-            sk='0',
-        ),
-        update_expr='SET cert = :cert, updated_at = :now',
-        expr_attr_values={
-            ':now': now_,
-            ':cert': {
-                'issued_at': now_,
-            }
-            | ({'expires_at': cert_expires_at} if cert_expires_at else {})
-            | ({'s3_uri': s3_uri} if s3_uri else {}),
-        },
+    return dyn.update_item(
+        key=KeyPair(pk=enrollment_id, sk='0'),
+        update_expr=update_expr,
+        expr_attr_values=expr_attr_values,
         cond_expr='attribute_exists(sk)',
     )
 
@@ -143,3 +104,67 @@ def _datefmt(dt: datetime) -> str:
         'Dezembro',
     ]
     return f'{dt.day:02d} de {months[dt.month - 1]} de {dt.year}'
+
+
+User = TypedDict('User', {'name': str, 'cpf': str})
+Cert = TypedDict('Cert', {'s3_uri': NotRequired[str]})
+
+
+def _gen_cert(
+    id: str,
+    *,
+    score: int | float,
+    cert: Cert,
+    user: User,
+    started_at: datetime,
+    completed_at: datetime,
+    expires_at: datetime | None = None,
+) -> str | None:
+    now_ = now()
+
+    if 's3_uri' not in cert:
+        logger.debug('Template URI is missing')
+        return None
+
+    try:
+        # Send template URI and data to Paperforge API to generate a PDF
+        r = requests.post(
+            PAPERFORGE_API,
+            data=json.dumps(
+                {
+                    'template_uri': cert['s3_uri'],
+                    'sign_uri': ESIGN_URI,
+                    'args': {
+                        'name': user['name'],
+                        'cpf': _cpffmt(user['cpf']),
+                        'score': score,
+                        'started_at': started_at.strftime('%d/%m/%Y'),
+                        'completed_at': completed_at.strftime('%d/%m/%Y'),
+                        'today': _datefmt(now_),
+                        'year': now_.strftime('%Y'),
+                        'expires_at': expires_at.strftime('%d/%m/%Y')
+                        if expires_at
+                        else None,
+                    },
+                },
+            ),
+            timeout=5,
+        )
+        r.raise_for_status()
+
+        object_key = f'certs/{id}.pdf'
+        s3_uri = f's3://{BUCKET_NAME}/{object_key}'
+
+        s3_client.put_object(
+            Bucket=BUCKET_NAME,
+            Key=object_key,
+            Body=r.content,
+            ContentType='application/pdf',
+        )
+
+        logger.debug(f'PDF uploaded successfully to {s3_uri}')
+    except requests.exceptions.RequestException as exc:
+        logger.exception(exc)
+        raise
+
+    return s3_uri