OpenAPI Schema
Pipeline views support OpenAPI schema by default.
Here is an example pipeline.
from django.urls import path
from rest_framework import serializers
from pipeline_views.views import BasePipelineView
from openapi_schema.views import get_schema_view
class InputSerializer(serializers.Serializer):
"""Example Input"""
name = serializers.CharField(help_text="foo")
age = serializers.IntegerField(help_text="bar")
class OutputSerializer(serializers.Serializer):
"""Example Output"""
email = serializers.EmailField(help_text="fizz")
age = serializers.IntegerField(help_text="buzz")
def example_method(name: str, age: int):
return {"email": f"{name.lower()}@email.com", "age": age}
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [
InputSerializer,
example_method,
OutputSerializer,
],
}
urlpatterns = [
path("example", ExampleView.as_view(), name="example"),
path(
"openapi/",
get_schema_view(
title="Your Project",
root_url="api/v1/",
description="API for all things",
version="1.0.0",
contact={"email": "user@example.com"},
license={"name": "MIT"},
terms_of_service="example.com",
),
name="openapi-schema",
),
]
This gets converted to the following OpenAPI schema.
openapi: 3.0.2
info:
title: Your Project
version: 1.0.0
description: API for all things
contact:
email: user@example.com
license:
name: MIT
termsOfService: example.com
paths:
/example/:
post:
operationId: createInput
description: Example Input
parameters: []
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/Input'
application/x-www-form-urlencoded:
schema:
$ref: '#/components/schemas/Input'
multipart/form-data:
schema:
$ref: '#/components/schemas/Input'
responses:
'200':
content:
application/json:
schema:
$ref: '#/components/schemas/Output'
description: Example Output
tags:
- example
components:
schemas:
Input:
type: object
properties:
name:
type: string
description: foo
age:
type: integer
description: bar
required:
- name
- age
Output:
type: object
properties:
email:
type: string
format: email
description: fizz
age:
type: integer
description: buzz
required:
- email
- age
Additional responses
You can add additional responses in the initialization of the schema.
from rest_framework import serializers
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ErrorSerializer(serializers.Serializer):
"""This is a custom error"""
loc = serializers.ListField(child=serializers.CharField())
msg = serializers.CharField()
type = serializers.CharField()
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
responses={
"POST": {
400: ErrorSerializer,
404: "This is the error message"
}
}
)
# ...
paths:
/example/:
post:
# ...
responses:
# ...
'400':
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
description: This is a custom error
'404':
content:
application/json:
schema:
type: object
properties:
detail:
type: string
default: error message
description: This is the error message
components:
schemas:
# ...
Error:
type: object
properties:
loc:
type: array
items:
type: string
msg:
type: string
type:
type: string
required:
- loc
- msg
- type
In case your pipeline returns a list response, a default 204 response will be added automatically.
Dynamic responses
You can also define dynamic responses with the help of MockSerializer.
from openapi_schema.schema import PipelineSchema
from pipeline_views.views import BasePipelineView
from serializer_inference.serializers import MockSerializer
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
responses={
"POST": {
404: MockSerializer.with_example(
description="This is the error message",
response={
"{date}": {
"{time}": [
"'free' or 'not free'",
],
},
},
)
}
}
)
# ...
paths:
/example:
post:
# ...
responses:
# ...
'404':
content:
application/json:
schema:
type: object
properties:
'{date}':
type: object
properties:
'{time}':
type: array
items:
type: string
default: '''free'' or ''not free'''
description: This is the error message
# ...
Deprecation
You can deprecate endpoints on a method by method basis.
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
deprecated=["POST"],
)
You can also use the pipeline_views.schema.deprecate
on the view.
from pipeline_views.views import BasePipelineView
from openapi_schema.utils import deprecate
@deprecate(methods=["POST"])
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
You can deprecate all methods by omitting the
methods
argument.
Security schemes
Add security schemes to the endpoints.
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
security={
"POST": {
"my_security": [],
},
},
)
The value for the security scheme defines its scopes.
The security scheme also needs to be added to the schema view. You can do this by adding the following:
from django.urls import path
from openapi_schema.views import get_schema_view
urlpatterns = [
path("example/", ExampleView.as_view(), name="test_view"),
path(
"openapi/",
get_schema_view(
title="Your Project",
description="API for all things",
version="1.0.0",
security_schemes={
"my_security": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
},
},
),
name="openapi-schema",
),
]
# ...
components:
# ...
securitySchemes:
my_security:
type: http
scheme: bearer
bearerFormat: JWT
# ...
Automatic security schemes
You can also define rules that will automatically add certain security schemes to views based on their authentication and permission classes. The key for the security rules is either a single authentication and permission class, or a tuple of them. If the view already has any of the schemas defined for it, the view's configuration will take precedence.
from django.urls import path
from rest_framework.permissions import IsAuthenticated
from openapi_schema.views import get_schema_view
from pipeline_views.views import BasePipelineView
class ExampleView(BasePipelineView):
"""Example View"""
permission_classes = [IsAuthenticated]
pipelines = {
"POST": [...],
}
urlpatterns = [
path("example/", ExampleView.as_view(), name="test_view"),
path(
"openapi/",
get_schema_view(
title="Your Project",
description="API for all things",
version="1.0.0",
security_schemes={
"my_security": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
},
},
security_rules={
IsAuthenticated: {
"my_security": [],
},
},
),
name="openapi-schema",
),
]
# ...
paths:
/example:
post:
# ...
security:
- my_security: []
components:
# ...
securitySchemes:
my_security:
type: http
scheme: bearer
bearerFormat: JWT
# ...
Query, path, header, and cookie parameters
For pipelines using the GET method, input serializer fields are interpreted automatically as query parameters. If the endpoint has path parameters, those are used in the schema instead, but with the documentation from the input serializer.
For other HTTP methods, you need to explicitly state that if a value is given as a parameter instead of in the request body. This is just for schema definition, the endpoints will actually accept the input from both places.
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
query_parameters={
"POST": ["name"],
},
)
You can also declare a parameter as a header or as cookie parameter.
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
header_parameters={
"POST": ["name"],
},
cookie_parameters={
"POST": ["name"],
},
)
However, you'll need to handle getting the parameter from cookies in
the input serializer separately. You can use HeaderAndCookieSerializer
for this. This will ass fields to the serializer to accept the defined
headers and cookies from the incoming request as strings, or None if
they were not given.
from pipeline_views.serializers import HeaderAndCookieSerializer
class TestSerialzer(HeaderAndCookieSerializer):
take_from_headers = ["foo"]
take_from_cookies = ["bar"]
External docs
External docs for an endpoint can also be added.
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
external_docs={
"POST": {
"description": "Look here for more information",
"url": "...",
},
},
)
Public
Endpoints can be set public/private for the whole API (public by default). Private endpoints are not visible to users that do not have the appropriate permissions.
from django.urls import path
from openapi_schema.views import get_schema_view
urlpatterns = [
path("example/", ExampleView.as_view(), name="test_view"),
path(
"openapi/",
get_schema_view(
title="Your Project",
description="API for all things",
version="1.0.0",
public=False,
),
name="openapi-schema",
),
]
You can also set individual endpoints public/private. This will override the API-wide configuration.
from pipeline_views.views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
public={
"POST": False,
},
)
Links
Using links, you can describe how various values returned by one operation can be used as input for other operations.
from pipeline_views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
links={
"POST": {
200: {
"LinkTitle": {
"description": "Description",
"operationId": "partialUpdateInput",
"parameters": {
"age": "$request.body#/age",
},
}
},
},
},
)
# ...
paths:
/example/:
post:
# ...
responses:
'200':
# ...
links:
LinkTitle:
description: Description
operationId: partialUpdateInput
parameters:
age: $request.body#/age
# ...
Callbacks
Callbacks are asynchronous, out-of-band requests that your service will send to some other service in response to certain events.
from rest_framework import serializers
from pipeline_views import BasePipelineView
from openapi_schema.schema import PipelineSchema
class InputSerializer(serializers.Serializer):
"""Example Input"""
name = serializers.CharField()
age = serializers.IntegerField()
class OutputSerializer(serializers.Serializer):
"""Example Output"""
email = serializers.EmailField()
age = serializers.IntegerField()
class ExampleView(BasePipelineView):
"""Example View"""
pipelines = {
"POST": [...],
}
schema = PipelineSchema(
callbacks={
"EventName": {
"CallbackUrl": {
"POST": {
"request_body": InputSerializer,
"responses": {
200: OutputSerializer,
},
},
},
},
},
)
# ...
paths:
/example/:
post:
# ...
callbacks:
EventName:
CallbackUrl:
post:
requestBody:
content:
application/json:
schema:
type: object
properties:
name:
type: string
age:
type: integer
required:
- name
- age
responses:
200:
content:
application/json:
schema:
type: object
properties:
email:
type: string
format: email
age:
type: integer
required:
- email
- age
# ...
Webhooks
Webhooks describe requests initiated other than by an API call, for example by an out-of-band registration. You can define them in the PipelineSchemaGenerator.
from django.urls import path
from rest_framework import serializers
from openapi_schema.views import get_schema_view
class InputSerializer(serializers.Serializer):
"""Example Input"""
name = serializers.CharField()
age = serializers.IntegerField()
class OutputSerializer(serializers.Serializer):
"""Example Output"""
email = serializers.EmailField()
age = serializers.IntegerField()
urlpatterns = [
path(
"openapi/",
get_schema_view(
title="Your Project",
description="API for all things",
version="1.0.0",
webhooks={
"ExampleWebhook": {
"method": "POST",
"request_data": InputSerializer,
"responses": {
200: OutputSerializer,
400: "Failure",
},
},
},
),
name="openapi-schema",
),
]
# ...
webhooks:
ExampleWebhook:
POST:
requestBody:
description: Example Input
content:
application/json:
schema:
type: object
properties:
name:
type: string
age:
type: integer
required:
- name
- age
responses:
'200':
description: Example Output
content:
application/json:
type: object
properties:
email:
type: string
format: email
age:
type: integer
required:
- email
- age
'400':
description: Failure
# ...