# octavia.api.common.hooks

# Source code for octavia.api.common.hooks

    #    Copyright 2014 Rackspace
    #
    #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    #    not use this file except in compliance with the License. You may obtain
    #    a copy of the License at
    #
    #         http://www.apache.org/licenses/LICENSE-2.0
    #
    #    Unless required by applicable law or agreed to in writing, software
    #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    #    License for the specific language governing permissions and limitations
    #    under the License.
    from pecan import hooks
    from webob import acceptparse
    from webob import exc

    from octavia.api.common import pagination
    from octavia.api.common import utils
    from octavia.common import constants
    from octavia.common import context
    from octavia.i18n import _

    _HEALTHCHECK_PATHS = ['/healthcheck', '/load-balancer/healthcheck']

    class ContentTypeHook(hooks.PecanHook):
        """Force the request content type to JSON if that is acceptable."""

        def on_route(self, state):
            # Oslo healthcheck middleware has its own content type handling
            # so we need to bypass the Octavia content type restrictions.
            if state.request.path in _HEALTHCHECK_PATHS:
                return
            # TODO(johnsom) Testing for an empty string is a workaround for an
            #               openstacksdk bug present up to the initial
            #               antelope release of openstacksdk. This means the
            #               octavia dashboard would also be impacted.
            #               This can be removed once antelope is EOL.
            # See: https://review.opendev.org/c/openstack/openstacksdk/+/876669
            if state.request.accept and state.request.accept.header_value != '':
                best_matches = state.request.accept.acceptable_offers(
                    [constants.APPLICATION_JSON])
                if not best_matches:
                    # The API reference says we always respond with JSON
                    state.request.accept = acceptparse.create_accept_header(
                        constants.APPLICATION_JSON)
                    msg = _('Only content type %s is accepted.')
                    raise exc.HTTPNotAcceptable(
                        msg % constants.APPLICATION_JSON,
                        json_formatter=utils.json_error_formatter)

            # Force application/json with no other options for the request
            state.request.accept = acceptparse.create_accept_header(
                constants.APPLICATION_JSON)

    class ContextHook(hooks.PecanHook):
        """Configures a request context and attaches it to the request."""

        def on_route(self, state):
            context_obj = context.RequestContext.from_environ(
                state.request.environ)
            state.request.context['octavia_context'] = context_obj

    class QueryParametersHook(hooks.PecanHook):

        def before(self, state):
            if state.request.method != 'GET':
                return

            state.request.context[
                constants.PAGINATION_HELPER] = pagination.PaginationHelper(
                state.request.params.mixed())

this page last updated: 2026-04-02 20:14:22

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug