def __call__()

in src/main.py [0:0]

• 71 lines of code
• 10 McCabe index (conditional complexity)

    def __call__(self, **kwargs):
        from validators import registry

        api_def = self.custodian_api_definition
        for model in registry.iter_models(without_get=True):
            schema = model.model_json_schema()
            dereference_json(schema)
            schema.pop('$defs', None)
            api_def[self.custodian_api_gateway_name]['models'].update(
                {
                    model.__name__: {
                        'content_type': 'application/json',
                        'schema': schema,
                    }
                }
            )
        path = self.models_deployment_resources
        _LOG.info(f'Updating {path}')
        with open(path, 'w') as file:
            json.dump(api_def, file, indent=2, sort_keys=True)
        _LOG.info(f'{path} has been updated')

        # here we update api gateway inside main deployment resources.
        # We don't remove existing endpoints, only add new in case they are
        # defined in RequestModelRegistry and are absent inside deployment
        # resources. Also, we update request and response models. Default
        # lambda is configuration-api-handler. Change it if it's wrong
        path = self.mail_deployment_resources
        _LOG.info(f'Updating {path}')
        with open(path, 'r') as file:
            deployment_resources = json.load(file)
        api = deployment_resources.get(self.custodian_api_gateway_name)
        if not api:
            _LOG.warning('Api gateway not found in deployment_resources')
            return
        resources = api.setdefault('resources', {})
        for item in registry.iter_all():
            # if endpoint & method are defined, just update models.
            # otherwise add configuration
            data = resources.setdefault(
                item.path,
                {'policy_statement_singleton': True, 'enable_cors': True},
            ).setdefault(
                item.method.value,
                {
                    'integration_type': 'lambda',
                    'enable_proxy': True,
                    'lambda_alias': '${lambdas_alias_name}',
                    'authorization_type': 'authorizer'
                    if item.auth
                    else 'NONE',
                    'lambda_name': 'caas-configuration-api-handler',
                },
            )
            data.pop('method_request_models', None)
            data.pop('responses', None)
            data.pop('method_request_parameters', None)
            if model := item.request_model:
                match item.method:
                    case HTTPMethod.GET:
                        params = {}
                        for name, info in model.model_fields.items():
                            params[f'method.request.querystring.{name}'] = (
                                info.is_required()
                            )
                        data['method_request_parameters'] = params
                    case _:
                        data['method_request_models'] = {
                            'application/json': model.__name__
                        }
            responses = []
            for st, m, description in item.responses:
                resp = {'status_code': str(st.value)}
                if m:
                    resp['response_models'] = {'application/json': m.__name__}
                responses.append(resp)
            data['responses'] = responses

        with open(path, 'w') as file:
            json.dump(deployment_resources, file, indent=2)
        _LOG.info(f'{path} has been updated')