def generate()

in src/services/openapi_spec_generator.py [0:0]
63 lines of code
13 McCabe index (conditional complexity)

    def generate(self) -> OpenApiV3:
        base = self._generate_base()
        paths = base['paths']
        schemas = base['components']['schemas']
        tags = set()  # todo allow to provide tags with description
        for endpoint in self._endpoints:
            data = paths.setdefault(endpoint.path, {}).setdefault(
                endpoint.method.lower(), {})
            # tags
            if self._auto_tags and not endpoint.tags:
                tag = endpoint.path.strip('/').split('/')[0].title()
                data['tags'] = [tag]
                tags.add(tag)
            else:
                data['tags'] = endpoint.tags
                tags.update(endpoint.tags)

            # path params
            for slot in re.finditer(self.dynamic_resource_regex,
                                    endpoint.path):
                data.setdefault('parameters', []).append({
                    'name': slot.group(),
                    'in': 'path',
                    'required': True,
                    'schema': {'type': 'string'}
                })

            # summary
            if summary := endpoint.summary:
                data['summary'] = summary

            # description
            if description := endpoint.description:
                data['description'] = description

            # request model / query params
            if model := endpoint.request_model:
                match endpoint.method:
                    case HTTPMethod.GET:
                        data.setdefault('parameters', []).extend(
                            self._model_to_parameters(model)
                        )
                    case _:
                        name = model.__name__
                        data['requestBody'] = {
                            'content': {
                                'application/json': {
                                    'schema': {
                                        '$ref': f'#/components/schemas/{name}'}
                                }
                            },
                            'required': True
                        }
                        schemas[name] = self._model_schema(model)

            # auth
            if endpoint.auth:
                data['security'] = [{'access_token': []}]

            # responses
            responses = data.setdefault('responses', {})
            for code, model, description in endpoint.responses:
                resp = {}
                if description:
                    resp['description'] = description
                if model:
                    name = model.__name__
                    resp['content'] = {
                        'application/json': {'schema': {
                            '$ref': f'#/components/schemas/{name}'
                        }}
                    }
                    schemas[name] = self._model_schema(model)
                responses[str(code.value)] = resp

        base['tags'].extend({'name': name} for name in sorted(tags))

        return base