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
# ...