Attach operation specification to connexion.request

[hubertgrzeskowiak]

Flask does only rudimentary JSON deserialisation to primitives or dictionaries. With named types defined in the swagger spec (object declarations) we could deserialise the dicts to custom DTO classes (which also could be easily generated). However during decoding in app.json_decoder we only get a dictionary and there is no way of telling which type we should transform to. It would be great if connexion exposed some kind of information about the spec for the current view.

Example code I played around with

class DataclassJSONDecoder(flask.json.JSONDecoder):
  def __init__(self, *args, **kwargs):
    kwargs["object_hook"] = self.map_to_object
    super(DataclassJSONDecoder, self).__init__(*args, **kwargs)

  def map_to_object(self, _dict):
    # What are the parameters for the current request?
	# To which type should we transform?
    pass


app = connexion.App(...)
app.app.json_decoder = DataclassJSONDecoder

Actual behaviour

I couldn't find any such meta information on the request.

Additional info:

• Python 3.7.3
• Connexion 1.1.15

[rafaelcaricio]

Right, I agree. That info could be exposed to the views, somehow. At this moment I don't know how we should do that though.

[frjonsen]

Was any progress made on this? I am attempting to implement role-based access by specifying allowed roles by adding x-roles in the specification, but need a way to actually read these either inside the controllers.

[brockhaywood]

+1 I'm trying to find a way to get at the schema of the current operation because the defaults are not provided in json body objects re: https://github.com/zalando/connexion/issues/1184

[brockhaywood]

Perhaps the same way that the request context is passed ? https://github.com/zalando/connexion/blob/master/connexion/decorators/parameter.py#L118

[frjonsen]

@brockhaywood We have since moved away from connexion to using FastAPI instead, but if it is of any help to you, this is how we solved this issue, specifically with our role based access functionality.

During the setup of the app, it looked like this. This is how we would get access to the OpenAPI specification at arbitrary points in the application.

Assume we had a specification which defines a path like this. Note the custom x-roles attribute.

paths:
  /admin/grant-tokens:
    post:
      tags:
        - tokens
      x-roles:
        - admin
      summary: Grant pre-paid tokens
      description: >
        Generate and save pre-paid tokens to a user
      requestBody:
        required: true
        description: Information about the tokens and who to grant tokens to
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/GrantTokensRequest"
      responses:
        "200":
          description: Returns an empty response if successful
        "403":
          description: Unauthorized
        "500":
          description: Internal server error

app = connexion.App(__name__, specification_dir="./openapi/", options=options)
openapi_definition = "openapi.yaml"
spec = app.add_api(openapi_definition, arguments={"title": "app"}, validate_responses=True).specification
app.app.config["specification"] = spec

Then, in our security controller, we had these functions. This first functions would figure out which is the operation we are currently in. There was a bug in this, in that it didn't handle non-existent operations (when the user would call something that didn't exist), but since these would always result in a 404 anyway, it didn't actually matter. The "error" prints in the console were annoying, but non-fatal.

def get_current_operation_id() -> str:
    spec = current_app.config["specification"]
    # connexion converts {} to <>, so have to convert it back for it to be a correct operationId
    operation_id = request.url_rule.rule.replace(spec.base_path, "").replace("<", "{").replace(">", "}")  # type: str

    return operation_id

This function would retrieve the the x-roles attribute for the current operation

def get_current_operation_roles() -> List[str]:
    spec = current_app.config["specification"]
    operation_id = get_current_operation_id()
    try:
        operation = spec.get_operation(operation_id, request.method.lower())
    except Exception:
        logging.getLogger(__name__).exception("Failed to match request to an operation")
        # Kind of ugly to throw an error like this, but better to return an internal server error
        # than to accidentally let users without the appropriate role get access
        raise Exception("Failed to match request to an operation")

    return operation.get("x-roles", [])  # type: ignore

We would then call this function during the authentication step in our security_controller_.py.

[brockhaywood]

@frjonsen thanks! that is quite helpful. I'm going to play around with this approach.