# designateclient.client

# Source code for designateclient.client

    # Copyright 2012 Managed I.T.
    #
    # Author: Kiall Mac Innes <kiall@managedit.ie>
    #
    # 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.

    import abc

    from stevedore import extension
    from urllib import parse

    from oslo_serialization import jsonutils

    from designateclient import exceptions

    class Controller(metaclass=abc.ABCMeta):

        def __init__(self, client):
            self.client = client

        def build_url(self, url, criterion=None, marker=None, limit=None):
            params = criterion or {}

            if marker is not None:
                params['marker'] = marker
            if limit is not None:
                params['limit'] = limit

            q = parse.urlencode(params) if params else ''
            return '{url}{params}'.format(
                url=url,
                params=f'?{q}'
            )

        def _serialize(self, kwargs):
            headers = kwargs.get('headers')
            content_type = headers.get('Content-Type') if headers else None

            if 'data' in kwargs and content_type in {None, 'application/json'}:
                kwargs['data'] = jsonutils.dumps(kwargs['data'])

        def _post(self, url, response_key=None, **kwargs):
            self._serialize(kwargs)

            resp, body = self.client.session.post(url, **kwargs)
            if response_key is not None:
                return body[response_key]
            return body

        def _get(self, url, response_key=None):
            resp, body = self.client.session.get(url)
            if response_key is not None:
                return body[response_key]
            return body

        def _patch(self, url, response_key=None, **kwargs):
            self._serialize(kwargs)

            resp, body = self.client.session.patch(url, **kwargs)
            if response_key is not None:
                return body[response_key]
            return body

        def _put(self, url, response_key=None, **kwargs):
            self._serialize(kwargs)

            resp, body = self.client.session.put(url, **kwargs)
            if response_key is not None:
                return body[response_key]
            return body

        def _delete(self, url, response_key=None, **kwargs):
            resp, body = self.client.session.delete(url, **kwargs)
            if response_key is not None:
                return body[response_key]
            return body

    class CrudController(Controller, metaclass=abc.ABCMeta):

        @abc.abstractmethod
        def list(self, *args, **kw):
            """
            List a resource
            """

        @abc.abstractmethod
        def get(self, *args, **kw):
            """
            Get a resource
            """

        @abc.abstractmethod
        def create(self, *args, **kw):
            """
            Create a resource
            """

        @abc.abstractmethod
        def update(self, *args, **kw):
            """
            Update a resource
            """

        @abc.abstractmethod
        def delete(self, *args, **kw):
            """
            Delete a resource
                """

    def get_versions():
        mgr = extension.ExtensionManager('designateclient.versions')
        return {ep.name: ep.plugin for ep in mgr.extensions}

    def Client(version, *args, **kwargs):  # noqa
        versions = get_versions()
        if version not in versions:
            msg = 'Version {} is not supported, use one of ({})'.format(
                version, list(versions.keys())
            )
            raise exceptions.UnsupportedVersion(msg)
        return versions[version](*args, **kwargs)

this page last updated: 2026-02-27 12:37:25

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