[OAS3] OpenApi v3 Schema Generator compliant to Italian Agid guidelines

Regading this: https://github.com/encode/django-rest-framework/issues/7452 .
It come with some minor changes and specialized files/classes that would handle these specifications.
This commit is contained in:
peppelinux 2020-08-04 15:52:26 +02:00
parent 559088463b
commit d0ef9d4660
6 changed files with 225 additions and 8 deletions

View File

@ -6,6 +6,7 @@ from rest_framework.schemas import coreapi
from rest_framework.schemas.openapi import SchemaGenerator from rest_framework.schemas.openapi import SchemaGenerator
OPENAPI_MODE = 'openapi' OPENAPI_MODE = 'openapi'
OPENAPI_AGID_MODE = 'openapi-agid' # WiP
COREAPI_MODE = 'coreapi' COREAPI_MODE = 'coreapi'
@ -20,9 +21,9 @@ class Command(BaseCommand):
parser.add_argument('--url', dest="url", default=None, type=str) parser.add_argument('--url', dest="url", default=None, type=str)
parser.add_argument('--description', dest="description", default=None, type=str) parser.add_argument('--description', dest="description", default=None, type=str)
if self.get_mode() == COREAPI_MODE: if self.get_mode() == COREAPI_MODE:
parser.add_argument('--format', dest="format", choices=['openapi', 'openapi-json', 'corejson'], default='openapi', type=str) parser.add_argument('--format', dest="format", choices=['openapi', 'openapi-json', 'openapi-agid', 'openapi-agid-json', 'corejson'], default='openapi', type=str)
else: else:
parser.add_argument('--format', dest="format", choices=['openapi', 'openapi-json'], default='openapi', type=str) parser.add_argument('--format', dest="format", choices=['openapi', 'openapi-json', 'openapi-agid', 'openapi-agid-json'], default='openapi', type=str)
parser.add_argument('--urlconf', dest="urlconf", default=None, type=str) parser.add_argument('--urlconf', dest="urlconf", default=None, type=str)
parser.add_argument('--generator_class', dest="generator_class", default=None, type=str) parser.add_argument('--generator_class', dest="generator_class", default=None, type=str)
parser.add_argument('--file', dest="file", default=None, type=str) parser.add_argument('--file', dest="file", default=None, type=str)
@ -54,12 +55,16 @@ class Command(BaseCommand):
'corejson': renderers.CoreJSONRenderer, 'corejson': renderers.CoreJSONRenderer,
'openapi': renderers.CoreAPIOpenAPIRenderer, 'openapi': renderers.CoreAPIOpenAPIRenderer,
'openapi-json': renderers.CoreAPIJSONOpenAPIRenderer, 'openapi-json': renderers.CoreAPIJSONOpenAPIRenderer,
'openapi-agid-json': renderers.CoreAPIJSONOpenAPIRenderer,
'openapi-agid': renderers.CoreAPIOpenAPIRenderer,
}[format] }[format]
return renderer_cls() return renderer_cls()
renderer_cls = { renderer_cls = {
'openapi': renderers.OpenAPIRenderer, 'openapi': renderers.OpenAPIRenderer,
'openapi-json': renderers.JSONOpenAPIRenderer, 'openapi-json': renderers.JSONOpenAPIRenderer,
'openapi-agid-json': renderers.JSONOpenAPIRenderer,
'openapi-agid': renderers.OpenAPIRenderer,
}[format] }[format]
return renderer_cls() return renderer_cls()

View File

@ -32,7 +32,7 @@ def get_schema_view(
public=False, patterns=None, generator_class=None, public=False, patterns=None, generator_class=None,
authentication_classes=api_settings.DEFAULT_AUTHENTICATION_CLASSES, authentication_classes=api_settings.DEFAULT_AUTHENTICATION_CLASSES,
permission_classes=api_settings.DEFAULT_PERMISSION_CLASSES, permission_classes=api_settings.DEFAULT_PERMISSION_CLASSES,
version=None): version=None, **kwargs):
""" """
Return a schema view. Return a schema view.
""" """
@ -44,7 +44,7 @@ def get_schema_view(
generator = generator_class( generator = generator_class(
title=title, url=url, description=description, title=title, url=url, description=description,
urlconf=urlconf, patterns=patterns, version=version urlconf=urlconf, patterns=patterns, version=version, **kwargs
) )
# Avoid import cycle on APIView # Avoid import cycle on APIView

View File

@ -0,0 +1,56 @@
"""
rest_framework.schemas
schemas:
__init__.py
generators.py # Top-down schema generation
inspectors.py # Per-endpoint view introspection
utils.py # Shared helper functions
views.py # Houses `SchemaView`, `APIView` subclass.
We expose a minimal "public" API directly from `schemas`. This covers the
basic use-cases:
from rest_framework.schemas import (
AutoSchema,
ManualSchema,
get_schema_view,
SchemaGenerator,
)
Other access should target the submodules directly
"""
from rest_framework.settings import api_settings
api_settings.defaults['DEFAULT_SCHEMA_CLASS'] = \
'rest_framework.schemas.openapi_agid.AgidAutoSchema'
from . import openapi_agid
# from .openapi_agid import AgidAutoSchema, AgidSchemaGenerator # noqa
from .inspectors import DefaultSchema # noqa
def get_schema_view(
title=None, url=None, description=None, urlconf=None, renderer_classes=None,
public=False, patterns=None,
authentication_classes=api_settings.DEFAULT_AUTHENTICATION_CLASSES,
permission_classes=api_settings.DEFAULT_PERMISSION_CLASSES,
version=None, **kwargs):
"""
Return a schema view.
"""
generator_class = openapi_agid.AgidSchemaGenerator
generator = generator_class(
title=title, url=url, description=description,
urlconf=urlconf, patterns=patterns, version=version, **kwargs
)
# Avoid import cycle on APIView
from .views import SchemaView
return SchemaView.as_view(
renderer_classes=renderer_classes,
schema_generator=generator,
public=public,
authentication_classes=authentication_classes,
permission_classes=permission_classes,
)

View File

@ -151,7 +151,9 @@ class BaseSchemaGenerator:
# Set by 'SCHEMA_COERCE_PATH_PK'. # Set by 'SCHEMA_COERCE_PATH_PK'.
coerce_path_pk = None coerce_path_pk = None
def __init__(self, title=None, url=None, description=None, patterns=None, urlconf=None, version=None): def __init__(self, title=None, url=None, description=None,
patterns=None, urlconf=None, version=None, **kwargs):
if url and not url.endswith('/'): if url and not url.endswith('/'):
url += '/' url += '/'
@ -165,6 +167,9 @@ class BaseSchemaGenerator:
self.url = url self.url = url
self.endpoints = None self.endpoints = None
for k,v in kwargs.items():
setattr(self, k, v)
def _initialise_endpoints(self): def _initialise_endpoints(self):
if self.endpoints is None: if self.endpoints is None:
inspector = self.endpoint_inspector_cls(self.patterns, self.urlconf) inspector = self.endpoint_inspector_cls(self.patterns, self.urlconf)

View File

@ -113,9 +113,8 @@ class SchemaGenerator(BaseSchemaGenerator):
return schema return schema
# View Inspectors # View Inspectors
class AutoSchema(ViewInspector): class AutoSchema(ViewInspector):
def __init__(self, tags=None, operation_id_base=None, component_name=None): def __init__(self, tags=None, operation_id_base=None, component_name=None):
@ -471,13 +470,15 @@ class AutoSchema(ViewInspector):
if isinstance(field, serializers.FloatField): if isinstance(field, serializers.FloatField):
content = { content = {
'type': 'number', 'type': 'number',
'format': 'float'
} }
self._map_min_max(field, content) self._map_min_max(field, content)
return content return content
if isinstance(field, serializers.IntegerField): if isinstance(field, serializers.IntegerField):
content = { content = {
'type': 'integer' 'type': 'integer',
'format': 'int64'
} }
self._map_min_max(field, content) self._map_min_max(field, content)
# 2147483647 is max for int32_size, so we use int64 for format # 2147483647 is max for int32_size, so we use int64 for format

View File

@ -0,0 +1,150 @@
from rest_framework import (
RemovedInDRF314Warning, exceptions, renderers, serializers
)
from urllib.parse import urljoin
from . openapi import SchemaGenerator, AutoSchema
from . utils import get_pk_description, is_list_view
class AgidSchemaGenerator(SchemaGenerator):
def get_info(self):
# Title and version are required by openapi specification 3.x
info = {
'title': self.title or '',
'version': self.version or '',
'contact': getattr(self, 'contact', {}),
'termsOfService': getattr(self, 'termsOfService', {}),
'license': getattr(self, 'license', {}),
'x-api-id': getattr(self, 'x-api-id', {}),
'x-summary': getattr(self, 'x-summary', {}),
}
if self.description is not None:
info['description'] = self.description
return info
def get_servers(self):
servers = getattr(self, 'servers', {})
return servers
def get_tags(self):
tags = getattr(self, 'tags', {}),
return tags[0] if isinstance(tags, tuple) else tags
def get_schema(self, request=None, public=False):
"""
Generate a OpenAPI schema.
"""
self._initialise_endpoints()
components_schemas = {}
# Iterate endpoints generating per method path operations.
paths = {}
_, view_endpoints = self._get_paths_and_endpoints(None if public else request)
for path, method, view in view_endpoints:
if not self.has_view_permissions(path, method, view):
continue
operation = view.schema.get_operation(path, method)
components = view.schema.get_components(path, method)
for k in components.keys():
if k not in components_schemas:
continue
if components_schemas[k] == components[k]:
continue
warnings.warn('Schema component "{}" has been overriden with a different value.'.format(k))
components_schemas.update(components)
# Normalise path for any provided mount url.
if path.startswith('/'):
path = path[1:]
# AGID - remove trailing / from urls
if path.endswith('/'):
path = path[:-1]
path = urljoin(self.url or '/', path)
paths.setdefault(path, {})
paths[path][method.lower()] = operation
self.check_duplicate_operation_id(paths)
# Compile final schema.
schema = {
'openapi': '3.0.2',
'info': self.get_info(),
'tags': self.get_tags(),
'servers': self.get_servers(),
'paths': paths,
}
if len(components_schemas) > 0:
schema['components'] = {
'schemas': components_schemas
}
return schema
class AgidAutoSchema(AutoSchema):
def get_responses(self, path, method):
if method == 'DELETE':
return {
'204': {
'description': ''
}
}
self.response_media_types = self.map_renderers(path, method)
serializer = self.get_serializer(path, method)
if not isinstance(serializer, serializers.Serializer):
item_schema = {}
else:
item_schema = self._get_reference(serializer)
if is_list_view(path, method, self.view):
response_schema = {
'type': 'object',
'items': item_schema,
}
paginator = self.get_paginator()
if paginator:
response_schema = paginator.get_paginated_response_schema(response_schema)
else:
response_schema = item_schema
status_code = '201' if method == 'POST' else '200'
return {
status_code: {
'content': {
ct: {'schema': response_schema}
for ct in self.response_media_types
},
# description is a mandatory property,
# https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#responseObject
# TODO: put something meaningful into it
'description': ""
}
}
def get_tags(self, path, method):
# If user have specified tags, use them.
if self._tags:
return self._tags
# First element of a specific path could be valid tag. This is a fallback solution.
# PUT, PATCH, GET(Retrieve), DELETE: /user_profile/{id}/ tags = [user-profile]
# POST, GET(List): /user_profile/ tags = [user-profile]
if path.startswith('/'):
path = path[1:]
return [path.split('/')[0].replace('_', '-')]