How to use the c7n.utils.type_schema function in c7n

- name: rds-snapshot-retention
                resource: rds
                filters:
                  - type: value
                    key: BackupRetentionPeriod
                    value: 7
                    op: lt
                actions:
                  - type: retention
                    days: 7
                    copy-tags: true
                    enforce: exact
    """

    date_attribute = "BackupRetentionPeriod"
    schema = type_schema(
        'retention', **{'days': {'type': 'number'},
                        'copy-tags': {'type': 'boolean'},
                        'enforce': {'type': 'string', 'enum': [
                            'min', 'max', 'exact']}})
    permissions = ('rds:ModifyDBInstance',)

    def process(self, dbs):
        with self.executor_factory(max_workers=3) as w:
            futures = []
            for db in dbs:
                futures.append(w.submit(
                    self.process_snapshot_retention,
                    db))
            for f in as_completed(futures):
                if f.exception():
                    self.log.error(

"""Filter IAM users based on attached policy values

    :example:

    .. code-block:: yaml

        policies:
          - name: iam-users-with-admin-access
            resource: iam-user
            filters:
              - type: policy
                key: PolicyName
                value: AdministratorAccess
    """

    schema = type_schema('policy', rinherit=ValueFilter.schema)
    permissions = ('iam:ListAttachedUserPolicies',)

    def user_policies(self, user_set):
        client = local_session(self.manager.session_factory).client('iam')
        for u in user_set:
            if 'c7n:Policies' not in u:
                u['c7n:Policies'] = []
            aps = client.list_attached_user_policies(
                UserName=u['UserName'])['AttachedPolicies']
            for ap in aps:
                u['c7n:Policies'].append(
                    client.get_policy(PolicyArn=ap['PolicyArn'])['Policy'])

    def process(self, resources, event=None):
        user_set = chunks(resources, size=50)
        with self.executor_factory(max_workers=2) as w:

:example:

    .. code-block:: yaml

            policies:
              - name: ami-deregister-old
                resource: ami
                filters:
                  - type: image-age
                    days: 90
                actions:
                  - deregister
    """

    schema = type_schema('deregister', **{'delete-snapshots': {'type': 'boolean'}})
    permissions = ('ec2:DeregisterImage',)
    snap_expr = jmespath.compile('BlockDeviceMappings[].Ebs.SnapshotId')

    def process(self, images):
        client = local_session(self.manager.session_factory).client('ec2')
        image_count = len(images)
        images = [i for i in images if self.manager.ctx.options.account_id == i['OwnerId']]
        if len(images) != image_count:
            self.log.info("Implicitly filtered %d non owned images", image_count - len(images))

        for i in images:
            self.manager.retry(client.deregister_image, ImageId=i['ImageId'])

            if not self.data.get('delete-snapshots'):
                continue
            snap_ids = self.snap_expr.search(i) or ()

version = 'v1'
        component = 'projects.topics'
        enum_spec = ('list', 'topics[]', None)
        scope_template = "projects/{}"
        id = "name"

        @staticmethod
        def get(client, resource_info):
            return client.execute_command(
                'get', {'topic': resource_info['topic_id']})


@PubSubTopic.action_registry.register('delete')
class DeletePubSubTopic(MethodAction):

    schema = type_schema('delete')
    method_spec = {'op': 'delete'}

    def get_resource_params(self, m, r):
        return {'topic': r['name']}


@resources.register('pubsub-subscription')
class PubSubSubscription(QueryResourceManager):
    """GCP resource: https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions
    """
    class resource_type(TypeInfo):
        service = 'pubsub'
        version = 'v1'
        component = 'projects.subscriptions'
        enum_spec = ('list', 'subscriptions[]', None)
        scope_template = 'projects/{}'

"""Action to stop Opswork Stack (Stops all instances under stack)

    It is recommended to use a filter to avoid unwanted stopping of stacks

    :example:

    .. code-block:: yaml

            policies:
              - name: opswork-stop
                resource: opswork-stack
                actions:
                  - stop
    """

    schema = type_schema('stop')
    permissions = ("opsworks:StopStack",)

    def process(self, stacks):
        with self.executor_factory(max_workers=10) as w:
            list(w.map(self.process_stack, stacks))

    def process_stack(self, stack):
        client = local_session(
            self.manager.session_factory).client('opsworks')
        try:
            stack_id = stack['StackId']
            client.stop_stack(StackId=stack_id)
        except ClientError as e:
            self.log.exception(
                "Exception stopping stack:\n %s" % e)

Find users that can create other users

    .. code-block:: yaml

        policies:
          - name: super-users
            resource: iam-user
            filters:
              - type: check-permissions
                match: allowed
                actions:
                 - iam:CreateUser
    """

    schema = type_schema(
        'check-permissions', **{
            'match': {'oneOf': [
                {'enum': ['allowed', 'denied']},
                {'$ref': '#/definitions/filters/valuekv'},
                {'$ref': '#/definitions/filters/value'}]},
            'match-operator': {'enum': ['and', 'or']},
            'actions': {'type': 'array', 'items': {'type': 'string'}},
            'required': ('actions', 'match')})
    schema_alias = True
    policy_annotation = 'c7n:policy'
    eval_annotation = 'c7n:perm-matches'

    def get_permissions(self):
        if self.manager.type == 'iam-policy':
            return ('iam:SimulateCustomPolicy',)
        return ('iam:SimulatePrincipalPolicy',)

:example:

    .. code-block:: yaml

            policies:
              - name: s3-enable-logging
                resource: s3
                filters:
                  - "tag:Testing": present
                actions:
                  - type: toggle-logging
                    target_bucket: log-bucket
                    target_prefix: logs123
    """
    schema = type_schema(
        'toggle-logging',
        enabled={'type': 'boolean'},
        target_bucket={'type': 'string'},
        target_prefix={'type': 'string'})

    permissions = ("s3:PutBucketLogging", "iam:ListAccountAliases")

    def validate(self):
        if self.data.get('enabled', True):
            if not self.data.get('target_bucket'):
                raise PolicyValidationError(
                    "target_bucket must be specified on %s" % (
                        self.manager.data,))
        return self

    def process(self, resources):

policies:
            - name: elb-is-logging-test
              resource: elb
              filters:
                - type: is-logging

            - name: elb-is-logging-bucket-and-prefix-test
              resource: elb
              filters:
                - type: is-logging
                  bucket: prodlogs
                  prefix: elblogs
    """

    permissions = ("elasticloadbalancing:DescribeLoadBalancerAttributes",)
    schema = type_schema('is-logging',
                         bucket={'type': 'string'},
                         prefix={'type': 'string'}
                         )

    def process(self, resources, event=None):
        self.initialize(resources)
        bucket_name = self.data.get('bucket', None)
        bucket_prefix = self.data.get('prefix', None)

        return [elb for elb in resources
                if elb['Attributes']['AccessLog']['Enabled'] and
                (not bucket_name or bucket_name == elb['Attributes'][
                    'AccessLog'].get('S3BucketName', None)) and
                (not bucket_prefix or bucket_prefix == elb['Attributes'][
                    'AccessLog'].get('S3BucketPrefix', None))
                ]

root device)

    :Example:

    .. code-block:: yaml

        policies:
          - name: ec2-ephemeral-instances
            resource: ec2
            filters:
              - type: ephemeral

    http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/InstanceStorage.html
    """

    schema = type_schema('ephemeral')

    def __call__(self, i):
        return self.is_ephemeral(i)

    @staticmethod
    def is_ephemeral(i):
        for bd in i.get('BlockDeviceMappings', []):
            if bd['DeviceName'] in ('/dev/sda1', '/dev/xvda', 'xvda'):
                if 'Ebs' in bd:
                    return False
                return True
        return True


@filters.register('instance-uptime')
class UpTimeFilter(AgeFilter):

if state and key:
            client.modify_ebs_default_kms_key_id(
                KmsKeyId=self.data['key'])


@filters.register('s3-public-block')
class S3PublicBlock(ValueFilter):
    """Check for s3 public blocks on an account.

    https://docs.aws.amazon.com/AmazonS3/latest/dev/access-control-block-public-access.html
    """

    annotation_key = 'c7n:s3-public-block'
    annotate = False  # no annotation from value filter
    schema = type_schema('s3-public-block', rinherit=ValueFilter.schema)
    schema_alias = False
    permissions = ('s3:GetAccountPublicAccessBlock',)

    def process(self, resources, event=None):
        self.augment([r for r in resources if self.annotation_key not in r])
        return super(S3PublicBlock, self).process(resources, event)

    def augment(self, resources):
        client = local_session(self.manager.session_factory).client('s3control')
        for r in resources:
            try:
                r[self.annotation_key] = client.get_public_access_block(
                    AccountId=r['account_id']).get('PublicAccessBlockConfiguration', {})
            except client.exceptions.NoSuchPublicAccessBlockConfiguration:
                r[self.annotation_key] = {}