Operations more fields

This commit is contained in:
p1c2u 2019-10-21 23:16:24 +01:00
parent 4c4636a263
commit d7840d0d61
10 changed files with 114 additions and 2 deletions

View file

@ -0,0 +1,14 @@
"""OpenAPI core external docs factories module"""
from openapi_core.schema.external_docs.models import ExternalDocumentation
class ExternalDocumentationFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, external_doc_spec):
url = external_doc_spec['url']
description = external_doc_spec.get('description')
return ExternalDocumentation(url, description=description)

View file

@ -0,0 +1,9 @@
"""OpenAPI core external docs models module"""
class ExternalDocumentation(object):
"""Represents an OpenAPI External Documentation."""
def __init__(self, url, description=None):
self.url = url
self.description = description

View file

@ -4,10 +4,14 @@ from six import iteritems
from openapi_spec_validator.validators import PathItemValidator
from openapi_core.compat import lru_cache
from openapi_core.schema.external_docs.factories import (
ExternalDocumentationFactory,
)
from openapi_core.schema.operations.models import Operation
from openapi_core.schema.parameters.generators import ParametersGenerator
from openapi_core.schema.request_bodies.factories import RequestBodyFactory
from openapi_core.schema.responses.generators import ResponsesGenerator
from openapi_core.schema.security.factories import SecurityRequirementFactory
class OperationsGenerator(object):
@ -32,6 +36,21 @@ class OperationsGenerator(object):
parameters_list)
operation_id = operation_deref.get('operationId')
tags_list = operation_deref.get('tags', [])
summary = operation_deref.get('summary')
description = operation_deref.get('description')
security_requirements_list = operation_deref.get('security', [])
security = None
if security_requirements_list:
security = list(map(
self.security_requirement_factory.create,
security_requirements_list))
external_docs = None
if 'externalDocs' in operation_deref:
external_docs_spec = operation_deref.get('externalDocs')
external_docs = self.external_docs_factory.create(
external_docs_spec)
request_body = None
if 'requestBody' in operation_deref:
@ -43,8 +62,10 @@ class OperationsGenerator(object):
http_method,
Operation(
http_method, path_name, responses, list(parameters),
summary=summary, description=description,
external_docs=external_docs, security=security,
request_body=request_body, deprecated=deprecated,
operation_id=operation_id, tags=list(tags_list)
operation_id=operation_id, tags=list(tags_list),
),
)
@ -58,7 +79,17 @@ class OperationsGenerator(object):
def parameters_generator(self):
return ParametersGenerator(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def external_docs_factory(self):
return ExternalDocumentationFactory(self.dereferencer)
@property
@lru_cache()
def request_body_factory(self):
return RequestBodyFactory(self.dereferencer, self.schemas_registry)
@property
@lru_cache()
def security_requirement_factory(self):
return SecurityRequirementFactory(self.dereferencer)

View file

@ -8,15 +8,22 @@ class Operation(object):
def __init__(
self, http_method, path_name, responses, parameters,
request_body=None, deprecated=False, operation_id=None, tags=None):
summary=None, description=None, external_docs=None, security=None,
request_body=None, deprecated=False, operation_id=None, tags=None,
servers=None):
self.http_method = http_method
self.path_name = path_name
self.responses = dict(responses)
self.parameters = dict(parameters)
self.summary = summary
self.description = description
self.external_docs = external_docs
self.security = security
self.request_body = request_body
self.deprecated = deprecated
self.operation_id = operation_id
self.tags = tags
self.servers = servers
def __getitem__(self, name):
return self.parameters[name]

View file

View file

@ -0,0 +1,14 @@
"""OpenAPI core security factories module"""
from openapi_core.schema.security.models import SecurityRequirement
class SecurityRequirementFactory(object):
def __init__(self, dereferencer):
self.dereferencer = dereferencer
def create(self, security_requirement_spec):
name = next(iter(security_requirement_spec))
scope_names = security_requirement_spec[name]
return SecurityRequirement(name, scope_names=scope_names)

View file

@ -0,0 +1,9 @@
"""OpenAPI core security models module"""
class SecurityRequirement(object):
"""Represents an OpenAPI Security Requirement."""
def __init__(self, name, scope_names=None):
self.name = name
self.scope_names = scope_names or []

View file

@ -62,6 +62,14 @@ paths:
$ref: "#/components/responses/PetsResponse"
post:
summary: Create a pet
description: Creates new pet entry
externalDocs:
url: https://example.com
description: Find more info here
security:
- petstore_auth:
- write:pets
- read:pets
operationId: createPets
tags:
- pets

View file

@ -9,6 +9,7 @@ from openapi_core.schema.paths.models import Path
from openapi_core.schema.request_bodies.models import RequestBody
from openapi_core.schema.responses.models import Response
from openapi_core.schema.schemas.models import Schema
from openapi_core.schema.security.models import SecurityRequirement
from openapi_core.schema.servers.models import Server, ServerVariable
from openapi_core.shortcuts import create_spec
from openapi_core.validation.request.validators import RequestValidator
@ -79,6 +80,25 @@ class TestPetstore(object):
assert operation.http_method == http_method
assert operation.operation_id is not None
assert operation.tags == operation_spec['tags']
assert operation.summary == operation_spec.get('summary')
assert operation.description == operation_spec.get(
'description')
ext_docs_spec = operation_spec.get('externalDocs')
if ext_docs_spec:
ext_docs = operation.external_docs
assert ext_docs.url == ext_docs_spec['url']
assert ext_docs.description == ext_docs_spec.get(
'description')
security_spec = operation_spec.get('security')
if security_spec:
for idx, sec_req in enumerate(operation.security):
assert type(sec_req) == SecurityRequirement
sec_req_spec = security_spec[idx]
sec_req_nam = next(iter(sec_req_spec))
assert sec_req.name == sec_req_nam
assert sec_req.scope_names == sec_req_spec[sec_req_nam]
responses_spec = operation_spec.get('responses')