Integration with traversal

[silenius]

All examples and documentation focus on URL dispatch, it would be good to support Traversal too or at least provide an example on how integrate pyramid_openapi3 in a Traversal only application

[zupo]

I've only lived in the URL dispatch world for the last ~5 years, and wouldn't know how to even being. Can you provide a basic example and I can then make it pretty, add tests, etc.

[stevepiercy]

The pyramid-cookiecutter-starter and related tutorial might be helpful.

[jvitus]

Hi there is a code example:

import os
import random
from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPNotImplemented
from pyramid.interfaces import ILocation
from pyramid.router import Router
from zope.interface import implementer
from zope.interface import Interface
from wsgiref.simple_server import make_server

class IGETView(Interface):
    pass

class IPOSTView(Interface):
    pass

class GETView:
    def __init__(self, context, request):
        self.context = context
        self.request = request

    def __call__(self):
        return self.context.GET()

class POSTView:
    def __init__(self, context, request):
        self.context = context
        self.request = request

    def __call__(self):
        return self.context.POST()


@implementer(ILocation, IGETView, IPOSTView)
class TraversalRoot:
    def __init__(self, name, parent, request):
        self.__name__ = name
        self.__parent__ = parent
        self.__request__ = request

    def GET(self):
        raise  HTTPNotImplemented()
   
    def POST(self):
        raise  HTTPNotImplemented()
       
class Hello(TraversalRoot):
    def GET(self):
        return {"msg": "hello world"}

class Bad_Hello(TraversalRoot):
    def GET(self):
        return {"bad_field": "hello world"}

class Root(TraversalRoot):
    def __getitem__(self, name):
        if name == "hello":
            return Hello(name=name, parent=self, request=self.__request__)
        if name == "bad_hello":
            return Bad_Hello(name=name, parent=self, request=self.__request__)
        raise KeyError

    def POST(self):
        data = self.__request__.json_body

        content =  data.get('content', None)
        if content is None or content == '':
            response  = self.__request__.response
            response.status_code = 400
            response.content_type = "application/json"
            response.charset = "utf-8"
            response.json_body = {
                "field": "content",
                "message": "content field is required and can't be an empty string",
                "exception": "ValidationError"
            }
            return response

        toRet = {
            "ID": random.randint(1,100),
            "content" : data.get('content'),
            "comments": data.get('comments', 'NO_COMMENT_SEND')
        }
        return toRet


def api_factory(request):
    return Root(name="", parent=None, request=request)

def app() -> Router:
    """Prepare a Pyramid app."""
    with Configurator(root_factory=api_factory) as config:
        config.registry.settings["pyramid_openapi3.enable_endpoint_validation"] = False
        # config.registry.settings["pyramid_openapi3.enable_request_validation"] = False
        # config.registry.settings["pyramid_openapi3.enable_response_validation"] = False
        config.include("pyramid_openapi3")
        config.pyramid_openapi3_spec(
            os.path.join(os.path.dirname(__file__), "openapi.yaml"),
        )
        config.pyramid_openapi3_add_explorer()
        config.add_view(
            GETView,
            context=IGETView,
            request_method="GET",
            permission="read",
            name="",
            attr=None,
            renderer="json",
            openapi=True
        )
        config.add_view(
            POSTView,
            context=IPOSTView,
            request_method="POST",
            permission="create",
            name="",
            attr=None,
            renderer="json",
            openapi=True
        )

        return config.make_wsgi_app()


if __name__ == "__main__":
    """If app.py is called directly, start up the app."""
    print("Swagger UI available at http://127.0.0.1:6543/docs/")  # noqa: T001
    server = make_server("127.0.0.1", 6543, app())
    server.serve_forever()

and the openapi.yaml

openapi: "3.0.0"

info:
  version: "1.0.0"
  title: Example traversal api

paths:
  /:
    post:
      summary: Fake create message
      requestBody:
        required: true
        description: Data for creating a fake message
        content:
          application/json:
            schema:
              type: object
              $ref: '#/components/schemas/CreateMessage'

      responses:
        '200':
          description: Success message.
          content:
            application/json:
              schema:
                type: object
                $ref: "#/components/schemas/Message"

        '400':
          $ref: '#/components/responses/ValidationError'
  /hello:
    get:
      summary: Hello Word route
      responses:
        '200':
          description: Display Hello Word message
          content:
            application/json:
              schema:
                type: object
                $ref: "#/components/schemas/Hello"
  /bad_hello:
    get:
      summary: Hello Word route failed response validation
      responses:
        '200':
          description: failed response validation
          content:
            application/json:
              schema:
                type: object
                $ref: "#/components/schemas/Hello"

components:
  schemas:
    Hello:
      type: object
      required:
        - msg
      properties:
        msg:
          type: string
          maxLength: 40
    CreateMessage:
      type: object
      required:
        - content
      properties:
        content:
          type: string
        comments:
          type: string
    Message:
      type: object
      required:
        - ID
        - content
      properties:
        ID:
          type: integer
        content:
          type: string
        comments:
          type: string

    Error:
      type: object
      required:
        - message
      properties:
        field:
          type: string
        message:
          type: string
        exception:
          type: string

  responses:

    ValidationError:
      description: OpenAPI request/response validation failed
      content:
        application/json:
          schema:
            type: object
            items:
              $ref: "#/components/schemas/Error"

As you can see in the App method
pyramid_openapi3.enable_endpoint_validation= False

If we activate it, pyramid_openapi raise a MissingEndpointsError in check_all_routes method but we have no route declared in pyramid when use traversal right ?

we can activate pyramid_openapi3.enable_request_validation and pyramid_openapi3.enable_response_validation but:
That's will work for GET on 127.0.0.1:6543/bad_hello
Not for POST 127.0.0.1:6543/ (if you post a json not corresponding to the schema CreateMessage) excview_tween try to warn {route.name} but no route again

So is it working for traversal and i missing something in my understanding of how to implements it or it's an issue ?

[zupo]

Looks like some work is needed before it works with traversal out-of-the-box, but your examples shows that it should be possible.

[jvitus]

Ok,
I will keep using it for documentation feature.

Thanks for the answer