As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

# Modelagem de CloudFormation ganchos personalizados usando Python
<a name="hooks-model-python"></a>

A modelagem de CloudFormation Hooks personalizados envolve a criação de um esquema que define o Hook, suas propriedades e seus atributos. Este tutorial mostra como modelar Hooks personalizados usando Python.

## Etapa 1: gerar o pacote do projeto Hook
<a name="model-hook-project-package-python"></a>

Gere seu pacote de projeto Hook. A CloudFormation CLI cria funções de manipulador vazias que correspondem a ações específicas do Hook no ciclo de vida de destino, conforme definido na especificação do Hook.

```
cfn generate
```

O comando retorna a seguinte saída.

```
Generated files for MyCompany::Testing::MyTestHook
```

**nota**  
Certifique-se de que seus tempos de execução do Lambda evitem up-to-date o uso de uma versão obsoleta. Para obter mais informações, consulte [Atualização de tempos de execução do Lambda para tipos de recursos](https://docs.aws.amazon.com/cloudformation-cli/latest/userguide/runtime-update.html) e Hooks.

## Etapa 2: Adicionar manipuladores Hook
<a name="model-hook-project-add-handler"></a>

Adicione seu próprio código de tempo de execução do manipulador Hook aos manipuladores que você escolher implementar. Por exemplo, você pode adicionar o código a seguir para registro.

```
LOG.setLevel(logging.INFO)
LOG.info("Internal testing Hook triggered for target: " + request.hookContext.targetName);
```

A CloudFormation CLI gera o `src/models.py` arquivo a partir do. [Referência de sintaxe de esquema de configuração de hook](hook-configuration-schema.md)

**Example models.py**  

```
import sys
from dataclasses import dataclass
from inspect import getmembers, isclass
from typing import (
    AbstractSet,
    Any,
    Generic,
    Mapping,
    MutableMapping,
    Optional,
    Sequence,
    Type,
    TypeVar,
)

from cloudformation_cli_python_lib.interface import (
    BaseModel,
    BaseHookHandlerRequest,
)
from cloudformation_cli_python_lib.recast import recast_object
from cloudformation_cli_python_lib.utils import deserialize_list

T = TypeVar("T")


def set_or_none(value: Optional[Sequence[T]]) -> Optional[AbstractSet[T]]:
    if value:
        return set(value)
    return None


@dataclass
class HookHandlerRequest(BaseHookHandlerRequest):
    pass


@dataclass
class TypeConfigurationModel(BaseModel):
    limitSize: Optional[str]
    cidr: Optional[str]
    encryptionAlgorithm: Optional[str]

    @classmethod
    def _deserialize(
        cls: Type["_TypeConfigurationModel"],
        json_data: Optional[Mapping[str, Any]],
    ) -> Optional["_TypeConfigurationModel"]:
        if not json_data:
            return None
        return cls(
            limitSize=json_data.get("limitSize"),
            cidr=json_data.get("cidr"),
            encryptionAlgorithm=json_data.get("encryptionAlgorithm"),
        )


_TypeConfigurationModel = TypeConfigurationModel
```

## Etapa 3: Implementar manipuladores de gancho
<a name="model-hook-project-code-handler-python"></a>

Com as classes de dados do Python geradas, você pode escrever os manipuladores que realmente implementam a funcionalidade do Hook. Neste exemplo, você implementará os pontos de `preDelete` invocação `preCreate``preUpdate`, e para os manipuladores.

**Topics**
+ [Implemente o manipulador PreCreate](#model-hook-project-code-handler-python-precreate)
+ [Implemente o manipulador PreUpdate](#model-hook-project-code-handler-python-preupdate)
+ [Implemente o manipulador PreDelete](#model-hook-project-code-handler-python-predelete)
+ [Implemente um manipulador de Hook](#model-hook-project-code-handler-python-hook-handler)

### Implemente o manipulador PreCreate
<a name="model-hook-project-code-handler-python-precreate"></a>

O `preCreate` manipulador verifica as configurações de criptografia do lado do servidor para um recurso ou. `AWS::S3::Bucket ` `AWS::SQS::Queue`
+ Para um `AWS::S3::Bucket` recurso, o Hook só passará se o seguinte for verdadeiro.
  + A criptografia do bucket do Amazon S3 está definida.
  + A chave do bucket do Amazon S3 está habilitada para o bucket.
  + O algoritmo de criptografia definido para o bucket do Amazon S3 é o algoritmo correto necessário.
  + O ID da AWS Key Management Service chave está definido.
+ Para um `AWS::SQS::Queue` recurso, o Hook só passará se o seguinte for verdadeiro.
  + O ID da AWS Key Management Service chave está definido.

### Implemente o manipulador PreUpdate
<a name="model-hook-project-code-handler-python-preupdate"></a>

Implemente um `preUpdate` manipulador, que é iniciado antes das operações de atualização para todos os destinos especificados no manipulador. O `preUpdate` manipulador realiza o seguinte:
+ Para um `AWS::S3::Bucket` recurso, o Hook só passará se o seguinte for verdadeiro:
  + O algoritmo de criptografia de bucket para um bucket do Amazon S3 não foi modificado.

### Implemente o manipulador PreDelete
<a name="model-hook-project-code-handler-python-predelete"></a>

Implemente um `preDelete` manipulador, que é iniciado antes das operações de exclusão para todos os destinos especificados no manipulador. O `preDelete` manipulador realiza o seguinte:
+ Para um `AWS::S3::Bucket` recurso, o Hook só passará se o seguinte for verdadeiro:
  + Verifica se os recursos compatíveis mínimos necessários existirão na conta após a exclusão do recurso.
  + A quantidade mínima necessária de recursos compatíveis é definida na configuração do Hook.

### Implemente um manipulador de Hook
<a name="model-hook-project-code-handler-python-hook-handler"></a>

1. No seu IDE, abra o `handlers.py` arquivo, localizado na `src` pasta.

1. Substitua todo o conteúdo do `handlers.py` arquivo pelo código a seguir.  
**Example handlers.py**  

   ```
   import logging
   from typing import Any, MutableMapping, Optional
   import botocore
   
   from cloudformation_cli_python_lib import (
       BaseHookHandlerRequest,
       HandlerErrorCode,
       Hook,
       HookInvocationPoint,
       OperationStatus,
       ProgressEvent,
       SessionProxy,
       exceptions,
   )
   
   from .models import HookHandlerRequest, TypeConfigurationModel
   
   # Use this logger to forward log messages to CloudWatch Logs.
   LOG = logging.getLogger(__name__)
   TYPE_NAME = "MyCompany::Testing::MyTestHook"
   
   LOG.setLevel(logging.INFO)
   
   hook = Hook(TYPE_NAME, TypeConfigurationModel)
   test_entrypoint = hook.test_entrypoint
   
   
   def _validate_s3_bucket_encryption(
       bucket: MutableMapping[str, Any], required_encryption_algorithm: str
   ) -> ProgressEvent:
       status = None
       message = ""
       error_code = None
   
       if bucket:
           bucket_name = bucket.get("BucketName")
   
           bucket_encryption = bucket.get("BucketEncryption")
           if bucket_encryption:
               server_side_encryption_rules = bucket_encryption.get(
                   "ServerSideEncryptionConfiguration"
               )
               if server_side_encryption_rules:
                   for rule in server_side_encryption_rules:
                       bucket_key_enabled = rule.get("BucketKeyEnabled")
                       if bucket_key_enabled:
                           server_side_encryption_by_default = rule.get(
                               "ServerSideEncryptionByDefault"
                           )
   
                           encryption_algorithm = server_side_encryption_by_default.get(
                               "SSEAlgorithm"
                           )
                           kms_key_id = server_side_encryption_by_default.get(
                               "KMSMasterKeyID"
                           )  # "KMSMasterKeyID" is name of the property for an AWS::S3::Bucket
   
                           if encryption_algorithm == required_encryption_algorithm:
                               if encryption_algorithm == "aws:kms" and not kms_key_id:
                                   status = OperationStatus.FAILED
                                   message = f"KMS Key ID not set for bucket with name: f{bucket_name}"
                               else:
                                   status = OperationStatus.SUCCESS
                                   message = f"Successfully invoked PreCreateHookHandler for AWS::S3::Bucket with name: {bucket_name}"
                           else:
                               status = OperationStatus.FAILED
                               message = f"SSE Encryption Algorithm is incorrect for bucket with name: {bucket_name}"
                       else:
                           status = OperationStatus.FAILED
                           message = f"Bucket key not enabled for bucket with name: {bucket_name}"
   
                       if status == OperationStatus.FAILED:
                           break
               else:
                   status = OperationStatus.FAILED
                   message = f"No SSE Encryption configurations for bucket with name: {bucket_name}"
           else:
               status = OperationStatus.FAILED
               message = (
                   f"Bucket Encryption not enabled for bucket with name: {bucket_name}"
               )
       else:
           status = OperationStatus.FAILED
           message = "Resource properties for S3 Bucket target model are empty"
   
       if status == OperationStatus.FAILED:
           error_code = HandlerErrorCode.NonCompliant
   
       return ProgressEvent(status=status, message=message, errorCode=error_code)
   
   
   def _validate_sqs_queue_encryption(queue: MutableMapping[str, Any]) -> ProgressEvent:
       if not queue:
           return ProgressEvent(
               status=OperationStatus.FAILED,
               message="Resource properties for SQS Queue target model are empty",
               errorCode=HandlerErrorCode.NonCompliant,
           )
       queue_name = queue.get("QueueName")
   
       kms_key_id = queue.get(
           "KmsMasterKeyId"
       )  # "KmsMasterKeyId" is name of the property for an AWS::SQS::Queue
       if not kms_key_id:
           return ProgressEvent(
               status=OperationStatus.FAILED,
               message=f"Server side encryption turned off for queue with name: {queue_name}",
               errorCode=HandlerErrorCode.NonCompliant,
           )
   
       return ProgressEvent(
           status=OperationStatus.SUCCESS,
           message=f"Successfully invoked PreCreateHookHandler for targetAWS::SQS::Queue with name: {queue_name}",
       )
   
   
   @hook.handler(HookInvocationPoint.CREATE_PRE_PROVISION)
   def pre_create_handler(
       session: Optional[SessionProxy],
       request: HookHandlerRequest,
       callback_context: MutableMapping[str, Any],
       type_configuration: TypeConfigurationModel,
   ) -> ProgressEvent:
       target_name = request.hookContext.targetName
       if "AWS::S3::Bucket" == target_name:
           return _validate_s3_bucket_encryption(
               request.hookContext.targetModel.get("resourceProperties"),
               type_configuration.encryptionAlgorithm,
           )
       elif "AWS::SQS::Queue" == target_name:
           return _validate_sqs_queue_encryption(
               request.hookContext.targetModel.get("resourceProperties")
           )
       else:
           raise exceptions.InvalidRequest(f"Unknown target type: {target_name}")
   
   
   def _validate_bucket_encryption_rules_not_updated(
       resource_properties, previous_resource_properties
   ) -> ProgressEvent:
       bucket_encryption_configs = resource_properties.get("BucketEncryption", {}).get(
           "ServerSideEncryptionConfiguration", []
       )
       previous_bucket_encryption_configs = previous_resource_properties.get(
           "BucketEncryption", {}
       ).get("ServerSideEncryptionConfiguration", [])
   
       if len(bucket_encryption_configs) != len(previous_bucket_encryption_configs):
           return ProgressEvent(
               status=OperationStatus.FAILED,
               message=f"Current number of bucket encryption configs does not match previous. Current has {str(len(bucket_encryption_configs))} configs while previously there were {str(len(previous_bucket_encryption_configs))} configs",
               errorCode=HandlerErrorCode.NonCompliant,
           )
   
       for i in range(len(bucket_encryption_configs)):
           current_encryption_algorithm = (
               bucket_encryption_configs[i]
               .get("ServerSideEncryptionByDefault", {})
               .get("SSEAlgorithm")
           )
           previous_encryption_algorithm = (
               previous_bucket_encryption_configs[i]
               .get("ServerSideEncryptionByDefault", {})
               .get("SSEAlgorithm")
           )
   
           if current_encryption_algorithm != previous_encryption_algorithm:
               return ProgressEvent(
                   status=OperationStatus.FAILED,
                   message=f"Bucket Encryption algorithm can not be changed once set. The encryption algorithm was changed to {current_encryption_algorithm} from {previous_encryption_algorithm}.",
                   errorCode=HandlerErrorCode.NonCompliant,
               )
   
       return ProgressEvent(
           status=OperationStatus.SUCCESS,
           message="Successfully invoked PreUpdateHookHandler for target: AWS::SQS::Queue",
       )
   
   
   def _validate_queue_encryption_not_disabled(
       resource_properties, previous_resource_properties
   ) -> ProgressEvent:
       if previous_resource_properties.get(
           "KmsMasterKeyId"
       ) and not resource_properties.get("KmsMasterKeyId"):
           return ProgressEvent(
               status=OperationStatus.FAILED,
               errorCode=HandlerErrorCode.NonCompliant,
               message="Queue encryption can not be disable",
           )
       else:
           return ProgressEvent(status=OperationStatus.SUCCESS)
   
   
   @hook.handler(HookInvocationPoint.UPDATE_PRE_PROVISION)
   def pre_update_handler(
       session: Optional[SessionProxy],
       request: BaseHookHandlerRequest,
       callback_context: MutableMapping[str, Any],
       type_configuration: MutableMapping[str, Any],
   ) -> ProgressEvent:
       target_name = request.hookContext.targetName
       if "AWS::S3::Bucket" == target_name:
           resource_properties = request.hookContext.targetModel.get("resourceProperties")
           previous_resource_properties = request.hookContext.targetModel.get(
               "previousResourceProperties"
           )
   
           return _validate_bucket_encryption_rules_not_updated(
               resource_properties, previous_resource_properties
           )
       elif "AWS::SQS::Queue" == target_name:
           resource_properties = request.hookContext.targetModel.get("resourceProperties")
           previous_resource_properties = request.hookContext.targetModel.get(
               "previousResourceProperties"
           )
   
           return _validate_queue_encryption_not_disabled(
               resource_properties, previous_resource_properties
           )
       else:
           raise exceptions.InvalidRequest(f"Unknown target type: {target_name}")
   ```

Continue no próximo tópico [Registrando um gancho personalizado com CloudFormation](registering-hooks.md).