Add download certificate functionality: Implement DownloadCertificate class and related DTOs for handling certificate download requests. Introduce FileManager for S3 operations, update dependency container, and enhance API Gateway with a new download_certificate endpoint. Add HTML templates for download success, not found, and error responses.

Author: maxsonferovante

src/application/download_certificate.py

@@ -0,0 +1,39 @@
+import logging
+import uuid
+from src.domain.repository.certificate_repository import CertificateRepository
+from src.application.dto.download_certificate_dto import DownloadCertificateRequestDto, DownloadCertificateResponseDto
+from src.infrastructure.container.dependency_container import container
+from src.infrastructure.aws.file_manager import FileManager
+from src.domain.exception.certificate_not_found import CertificateNotFound
+
+logger = logging.getLogger(__name__)
+
+class DownloadCertificate:
+    def __init__(self):
+        self.certificate_repository: CertificateRepository = container.get('certificate_repository')
+        self.file_manager: FileManager = container.get('file_manager')
+
+    def execute(self, request: DownloadCertificateRequestDto) -> DownloadCertificateResponseDto:
+        logger.info(f"Buscando certificado para download com ID: {request.id}")
+        
+        certificate = self.certificate_repository.find_by_id(request.id)
+        
+        if not certificate:
+            raise CertificateNotFound(f"Certificado não encontrado para ID: {request.id}")
+        
+        certificate_url = ""
+        
+        if certificate.success and certificate.certificate_key:
+            certificate_url = self.file_manager.get_certificate_download_url(certificate.certificate_key)
+            logger.info(f"URL pré-assinada gerada para certificado {request.id}")
+        elif certificate.success and not certificate.certificate_key:
+            logger.warning(f"Certificado {request.id} marcado como sucesso mas sem certificate_key")
+        elif not certificate.success:
+            logger.info(f"Certificado {request.id} não foi processado com sucesso")
+        
+        return DownloadCertificateResponseDto(
+            certificate_url=certificate_url,
+            email=certificate.participant_email or "",
+            product_id=certificate.product_id,
+            success=certificate.success or False
+        )

src/application/dto/download_certificate_dto.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, Field
+import uuid
+
+class DownloadCertificateRequestDto(BaseModel):
+    id: uuid.UUID = Field(..., description="UUID do certificado")
+
+class DownloadCertificateResponseDto(BaseModel):
+    certificate_url: str
+    email: str
+    product_id: int
+    success: bool

src/application/dto/fetch_certificate_dto.py

@@ -27,6 +27,7 @@ class FetchCertificateResponseDto(BaseModel):
     participant_email: Optional[str] = None
     participant_document: Optional[str] = None
     certificate_url: Optional[str] = None
+    certificate_key: Optional[str] = None
     created_at: Optional[str] = None
     updated_at: Optional[str] = None
     email: Optional[str] = None

src/application/strategy/fetch_certificate_strategy.py

@@ -55,6 +55,7 @@ def execute(self, request: FetchCertificateRequestDto) -> List[FetchCertificateR
             participant_email=certificate.participant_email,
             participant_document=certificate.participant_cpf,
             certificate_url=certificate.certificate_url,
+            certificate_key=certificate.certificate_key,
             created_at=certificate.generated_date,
             updated_at=certificate.generated_date,  # Assumindo que não temos updated_at separado
             email=certificate.participant_email,

src/domain/repository/base_repository.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
-from typing import List, Optional, TypeVar, Generic
+from typing import List, Optional, TypeVar, Generic, Union
 from pydantic import BaseModel
+import uuid
 
 # Tipo genérico para as entidades
 T = TypeVar('T', bound=BaseModel)
@@ -21,6 +22,11 @@ def get_by_id(self, entity_id: str) -> Optional[T]:
         """Busca uma entidade pelo ID"""
         pass
 
+    @abstractmethod
+    def find_by_id(self, entity_id: Union[str, uuid.UUID]) -> Optional[T]:
+        """Busca uma entidade pelo UUID"""
+        pass
+    
     @abstractmethod
     def get_all(self) -> List[T]:
         """Busca todas as entidades"""

src/domain/repository/certificate_repository.py

@@ -1,7 +1,8 @@
 from abc import abstractmethod
-from typing import List, Optional
+from typing import List, Optional, Union
 from src.domain.entity.certificate import Certificate
 from src.domain.repository.base_repository import BaseRepository
+import uuid
 
 class CertificateRepository(BaseRepository[Certificate]):
     """

src/infrastructure/aws/boto_aws.py

@@ -5,6 +5,7 @@
 class ServiceNameAWS(Enum):
     SQS = 'sqs'
     DYNAMODB = 'dynamodb'
+    S3 = 's3'
 
 def get_instance_aws(service_name: ServiceNameAWS):
     return client(

src/infrastructure/aws/file_manager.py

@@ -0,0 +1,65 @@
+import logging
+from src.infrastructure.aws.boto_aws import get_instance_aws, ServiceNameAWS
+from src.infrastructure.config.config import config
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+class FileManager:
+    """
+    Gerencia operações de arquivos no S3.
+    """
+
+    def __init__(self):
+        self.aws = get_instance_aws(ServiceNameAWS.S3)
+        self.bucket_name = config.S3_BUCKET_NAME
+
+    def get_url(self, key: str, expires_in: int = 604800) -> str:
+        """
+        Gera uma URL pré-assinada para acessar (GET) um objeto no S3.
+        Configura response-content-disposition=inline para visualização direta no navegador.
+        
+        Args:
+            key: Chave do objeto no S3
+            expires_in: Tempo de expiração em segundos (padrão: 7 dias)
+            
+        Returns:
+            URL pré-assinada para acesso ao objeto
+            
+        Raises:
+            Exception: Se houver erro ao gerar a URL
+        """
+        try:
+            logger.info(f"Getting URL for key {key} with expiration {expires_in} seconds")
+            
+            # Gera URL pré-assinada para GET com response-content-disposition=inline
+            presigned_url = self.aws.generate_presigned_url(
+                'get_object',
+                Params={
+                    'Bucket': self.bucket_name,
+                    'Key': key,
+                    'ResponseContentDisposition': 'inline'
+                },
+                ExpiresIn=expires_in
+            )
+            
+            logger.info(f"Successfully retrieved URL for key {key}")
+            return presigned_url
+            
+        except Exception as e:
+            logger.error(f"Error generating presigned URL for key {key}: {e}")
+            return None
+    
+    def get_certificate_download_url(self, certificate_key: str) -> str:
+        """
+        Gera URL pré-assinada específica para download de certificados.
+        URL válida por 30 minutos.
+        
+        Args:
+            certificate_key: Chave do certificado no S3
+            
+        Returns:
+            URL pré-assinada para download do certificado (válida por 30 minutos)
+        """
+        return self.get_url(certificate_key, expires_in=1800)  # 30 minutos em segundos
+    

src/infrastructure/config/config.py

@@ -8,6 +8,7 @@
 class Config(BaseSettings):
     REGION: str
     BUILDER_QUEUE_URL: str
+    S3_BUCKET_NAME: str
     ENVIRONMENT: str = Field(default="dev")
     PROJECT_NAME: str = Field(default="certified-builder-api-py")
     URL_SERVICE_TECH: str

src/infrastructure/container/dependency_container.py

@@ -6,11 +6,12 @@
 from typing import Dict, Any
 from src.infrastructure.aws.dynamodb_service import DynamoDBService
 from src.infrastructure.aws.sqs_service import SQSService
+from src.infrastructure.aws.file_manager import FileManager
 from src.infrastructure.repository.certificate_repository_impl import CertificateRepositoryImpl
 from src.infrastructure.repository.participant_repository_impl import ParticipantRepositoryImpl
 from src.infrastructure.repository.product_repository_impl import ProductRepositoryImpl
 from src.infrastructure.repository.order_repository_impl import OrderRepositoryImpl
-from src.application.fetch_order_tech_floripa import FetchOrderTechFloripa
+
 
 class DependencyContainer:
     """
@@ -37,6 +38,7 @@ def _register_services(self):
         Define como cada dependência deve ser criada.
         """
         # Registra serviços de infraestrutura
+        self._services['file_manager'] = self._create_file_manager
         self._services['dynamodb_service'] = self._create_dynamodb_service
         self._services['sqs_service'] = self._create_sqs_service
         # Registra repositórios
@@ -49,6 +51,7 @@ def _register_services(self):
         self._services['create_certificate'] = self._create_create_certificate
         self._services['fetch_certificate'] = self._create_fetch_certificate
         self._services['fetch_order_tech_floripa'] = self._create_fetch_order_tech_floripa
+        self._services['download_certificate'] = self._create_download_certificate
 
     def get(self, service_name: str) -> Any:
         """
@@ -83,6 +86,10 @@ def reset(self):
         """
         self._singletons.clear()
 
+    def _create_file_manager(self) -> FileManager:
+        """Cria uma instância do FileManager."""
+        return FileManager()
+    
     def _create_sqs_service(self) -> SQSService:
         """Cria uma instância do SQSService."""
         return SQSService()
@@ -133,7 +140,11 @@ def _create_fetch_certificate(self):
         from src.application.fetch_certificate import FetchCertificate
         return FetchCertificate()
 
-
+    def _create_download_certificate(self):
+        """Cria uma instância do DownloadCertificate."""
+        from src.application.download_certificate import DownloadCertificate
+        return DownloadCertificate()
+        
     def _create_fetch_order_tech_floripa(self):
         """
         Cria uma instância do FetchOrderTechFloripa.