Skip to content

Adapters

Base API adapter implementation for odin-control.

Tim Nicholls, STFC Detector System Software Group

ApiAdapter

API adapter base class.

This class defines the basis for all API adapters and provides default methods for supported HTTP verbs and for lifecycle management. Dervied adapters can either override these methods explciitly to provide custom behavior, or adopt the adapter-controller pattern by specifying controller and error clases, and allowing default methods in this class to interface requests to the controller.

Source code in src/odin_control/adapters/adapter.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
class ApiAdapter():
    """API adapter base class.

    This class defines the basis for all API adapters and provides default methods for supported
    HTTP verbs and for lifecycle management. Dervied adapters can either override these methods
    explciitly to provide custom behavior, or adopt the adapter-controller pattern by specifying
    controller and error clases, and allowing default methods in this class to interface requests
    to the controller.
    """

    # Derived classes can specify controller and error classes to use
    controller_cls = None
    error_cls = BaseError

    # Derived classes should override this with a relevant version string
    version = "unknown"

    # Flag to indicate if adapter is async; default to False
    is_async = False

    def __init__(self, **kwargs):
        """Initialise the ApiAdapter object.

        This method initialises the adapter, loading any keyword arguments into the options
        dictionary, and instantiating the controller if a controller class has been specified

        :param kwargs: keyword argument list that is copied into options dictionary
        """
        self.name = type(self).__name__

        # Load any keyword arguments into the adapter options dictionary
        self.options = {}
        for kw in kwargs:
            self.options[kw] = kwargs[kw]

        # Instantiate the controller if a controller class has been specified
        if self.controller_cls:
            self.controller = self.controller_cls(self.options)
        else:
            self.controller = None

        logging.debug("%s loaded", self.name)

    def initialize(self, adapters):
        """Initialize the ApiAdapter after it has been registered by the API Route.

        This method allows the adapter to perform any initialization that requires access to other
        loaded adapters. It receives a dictionary of loaded adapters from the API route and passes
        them to the controller's initialize method, if a controller has been configured.

        :param adapters: a dictionary of the adapters loaded by the API route.
        """
        logging.debug("%s initialize called with %d adapters", self.name, len(adapters))

        # Build a dictionary of other adapters excluding self
        adapters = {name: adapter for name, adapter in adapters.items() if adapter is not self}

        # Initialize the controller with access to other adapters
        if self.controller:
            try:
                self.controller.initialize(adapters)
            except AttributeError:
                logging.warning("%s controller has no initialize method", self.name)
        else:
            logging.warning("%s has no controller configured", self.name)


    @response_types("application/json", default="application/json")
    @require_controller
    def get(self, path, request):
        """Handle an HTTP GET request.

        This method is a default implementation of the GET request handler for adapters. It calls
        the get method of the controller and returns the result. Error handling is provided to
        catch controller errors and return appropriate error responses.

        :param path: URI path of resource
        :param request: HTTP request object passed from handler
        :return: ApiAdapterResponse container of data, content-type and status_code
        """
        content_type = "application/json"

        try:
            response = self.controller.get(path, wants_metadata(request))
            status_code = 200
        except self.error_cls as error:
            response = {"error": str(error)}
            status_code = 400

        return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)

    @request_types("application/json", "application/vnd.odin-native")
    @response_types("application/json", default="application/json")
    @require_controller
    def put(self, path, request):
        """Handle an HTTP PUT request.

        This method is a default implementation of the PUT request handler for adapters. It calls
        the set method of the controller with the specified data and returns the result of a
        subsequent get call. Error handling is provided to catch controller errors and return
        appropriate error responses.

        :param path: URI path of resource
        :param request: HTTP request object passed from handler
        :return: ApiAdapterResponse container of data, content-type and status_code
        """
        content_type = "application/json"

        try:
            data = json_decode(request.body)
            self.controller.set(path, data)
            response = self.controller.get(path)
            status_code = 200
        except self.error_cls as error:
            response = {"error": str(error)}
            status_code = 400
        except NotImplementedError:
            response = {
                "error": f"{self.name} does not support PUT requests"
            }
            status_code = 405
        except (TypeError, ValueError) as error:
            response = {"error": f"Failed to decode PUT request body: {str(error)}"}
            status_code = 400

        return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)


    @request_types("application/json", "application/vnd.odin-native")
    @response_types("application/json", default="application/json")
    @require_controller
    def post(self, path, request):
        """Handle an HTTP POST request.

        This method is a default implementation of the POST request handler for adapters. It calls
        the create method of the controller and returns the result. Error handling is provided to
        catch controller errors and return appropriate error responses.

        :param path: URI path of resource
        :param request: HTTP request object passed from handler
        :return: ApiAdapterResponse container of data, content-type and status_code
        """
        content_type = "application/json"

        try:
            data = json_decode(request.body)
            response = self.controller.create(path, data)
            status_code = 200
        except self.error_cls as error:
            response = {"error": str(error)}
            status_code = 400
        except NotImplementedError:
            response = {
                "error": f"{self.name} does not support POST requests"
            }
            status_code = 405
        except (TypeError, ValueError) as error:
            response = {"error": f"Failed to decode POST request body: {str(error)}"}
            status_code = 400

        return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)

    @response_types("application/json", default="application/json")
    @require_controller
    def delete(self, path, request):
        """Handle an HTTP DELETE request.

        This method is a default implementation of the DELETE request handler for adapters. It calls
        the delete method of the controller and returns the result. Error handling is provided to
        catch controller errors and return appropriate error responses.

        :param path: URI path of resource
        :param request: HTTP request object passed from handler
        :return: ApiAdapterResponse container of data, content-type and status_code
        """
        content_type = "application/json"

        try:
            response = self.controller.delete(path)
            status_code = 200
        except self.error_cls as error:
            response = {"error": str(error)}
            status_code = 400
        except NotImplementedError:
            response = {
                "error": f"{self.name} does not support DELETE requests"
            }
            status_code = 405

        return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)

    def cleanup(self):
        """Clean up adapter state.

        This method is a default implementation of the cleanup mechanism provided to allow adapters
        to clean up their state (e.g. disconnect cleanly from the device being controlled, set
        some status message).
        """
        logging.debug("%s cleanup called",  self.name)

        # If a controller is configured, call its cleanup method
        if self.controller:
            try:
                self.controller.cleanup()
            except AttributeError:
                logging.warning("%s controller has no cleanup method", self.name)
        else:
            logging.warning("%s has no controller configured", self.name)

__init__(**kwargs)

Initialise the ApiAdapter object.

This method initialises the adapter, loading any keyword arguments into the options dictionary, and instantiating the controller if a controller class has been specified

Parameters:

Name Type Description Default
kwargs

keyword argument list that is copied into options dictionary

 {}
Source code in src/odin_control/adapters/adapter.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(self, **kwargs):
    """Initialise the ApiAdapter object.

    This method initialises the adapter, loading any keyword arguments into the options
    dictionary, and instantiating the controller if a controller class has been specified

    :param kwargs: keyword argument list that is copied into options dictionary
    """
    self.name = type(self).__name__

    # Load any keyword arguments into the adapter options dictionary
    self.options = {}
    for kw in kwargs:
        self.options[kw] = kwargs[kw]

    # Instantiate the controller if a controller class has been specified
    if self.controller_cls:
        self.controller = self.controller_cls(self.options)
    else:
        self.controller = None

    logging.debug("%s loaded", self.name)

cleanup()

Clean up adapter state.

This method is a default implementation of the cleanup mechanism provided to allow adapters to clean up their state (e.g. disconnect cleanly from the device being controlled, set some status message).

Source code in src/odin_control/adapters/adapter.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def cleanup(self):
    """Clean up adapter state.

    This method is a default implementation of the cleanup mechanism provided to allow adapters
    to clean up their state (e.g. disconnect cleanly from the device being controlled, set
    some status message).
    """
    logging.debug("%s cleanup called",  self.name)

    # If a controller is configured, call its cleanup method
    if self.controller:
        try:
            self.controller.cleanup()
        except AttributeError:
            logging.warning("%s controller has no cleanup method", self.name)
    else:
        logging.warning("%s has no controller configured", self.name)

delete(path, request)

Handle an HTTP DELETE request.

This method is a default implementation of the DELETE request handler for adapters. It calls the delete method of the controller and returns the result. Error handling is provided to catch controller errors and return appropriate error responses.

Parameters:

Name Type Description Default
path

URI path of resource

 required
request

HTTP request object passed from handler

 required

Returns:

Type Description

ApiAdapterResponse container of data, content-type and status_code
Source code in src/odin_control/adapters/adapter.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@response_types("application/json", default="application/json")
@require_controller
def delete(self, path, request):
    """Handle an HTTP DELETE request.

    This method is a default implementation of the DELETE request handler for adapters. It calls
    the delete method of the controller and returns the result. Error handling is provided to
    catch controller errors and return appropriate error responses.

    :param path: URI path of resource
    :param request: HTTP request object passed from handler
    :return: ApiAdapterResponse container of data, content-type and status_code
    """
    content_type = "application/json"

    try:
        response = self.controller.delete(path)
        status_code = 200
    except self.error_cls as error:
        response = {"error": str(error)}
        status_code = 400
    except NotImplementedError:
        response = {
            "error": f"{self.name} does not support DELETE requests"
        }
        status_code = 405

    return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)

get(path, request)

Handle an HTTP GET request.

This method is a default implementation of the GET request handler for adapters. It calls the get method of the controller and returns the result. Error handling is provided to catch controller errors and return appropriate error responses.

Parameters:

Name Type Description Default
path

URI path of resource

 required
request

HTTP request object passed from handler

 required

Returns:

Type Description

ApiAdapterResponse container of data, content-type and status_code
Source code in src/odin_control/adapters/adapter.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@response_types("application/json", default="application/json")
@require_controller
def get(self, path, request):
    """Handle an HTTP GET request.

    This method is a default implementation of the GET request handler for adapters. It calls
    the get method of the controller and returns the result. Error handling is provided to
    catch controller errors and return appropriate error responses.

    :param path: URI path of resource
    :param request: HTTP request object passed from handler
    :return: ApiAdapterResponse container of data, content-type and status_code
    """
    content_type = "application/json"

    try:
        response = self.controller.get(path, wants_metadata(request))
        status_code = 200
    except self.error_cls as error:
        response = {"error": str(error)}
        status_code = 400

    return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)

initialize(adapters)

Initialize the ApiAdapter after it has been registered by the API Route.

This method allows the adapter to perform any initialization that requires access to other loaded adapters. It receives a dictionary of loaded adapters from the API route and passes them to the controller's initialize method, if a controller has been configured.

Parameters:

Name Type Description Default
adapters

a dictionary of the adapters loaded by the API route.

 required
Source code in src/odin_control/adapters/adapter.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def initialize(self, adapters):
    """Initialize the ApiAdapter after it has been registered by the API Route.

    This method allows the adapter to perform any initialization that requires access to other
    loaded adapters. It receives a dictionary of loaded adapters from the API route and passes
    them to the controller's initialize method, if a controller has been configured.

    :param adapters: a dictionary of the adapters loaded by the API route.
    """
    logging.debug("%s initialize called with %d adapters", self.name, len(adapters))

    # Build a dictionary of other adapters excluding self
    adapters = {name: adapter for name, adapter in adapters.items() if adapter is not self}

    # Initialize the controller with access to other adapters
    if self.controller:
        try:
            self.controller.initialize(adapters)
        except AttributeError:
            logging.warning("%s controller has no initialize method", self.name)
    else:
        logging.warning("%s has no controller configured", self.name)

post(path, request)

Handle an HTTP POST request.

This method is a default implementation of the POST request handler for adapters. It calls the create method of the controller and returns the result. Error handling is provided to catch controller errors and return appropriate error responses.

Parameters:

Name Type Description Default
path

URI path of resource

 required
request

HTTP request object passed from handler

 required

Returns:

Type Description

ApiAdapterResponse container of data, content-type and status_code
Source code in src/odin_control/adapters/adapter.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@request_types("application/json", "application/vnd.odin-native")
@response_types("application/json", default="application/json")
@require_controller
def post(self, path, request):
    """Handle an HTTP POST request.

    This method is a default implementation of the POST request handler for adapters. It calls
    the create method of the controller and returns the result. Error handling is provided to
    catch controller errors and return appropriate error responses.

    :param path: URI path of resource
    :param request: HTTP request object passed from handler
    :return: ApiAdapterResponse container of data, content-type and status_code
    """
    content_type = "application/json"

    try:
        data = json_decode(request.body)
        response = self.controller.create(path, data)
        status_code = 200
    except self.error_cls as error:
        response = {"error": str(error)}
        status_code = 400
    except NotImplementedError:
        response = {
            "error": f"{self.name} does not support POST requests"
        }
        status_code = 405
    except (TypeError, ValueError) as error:
        response = {"error": f"Failed to decode POST request body: {str(error)}"}
        status_code = 400

    return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)

put(path, request)

Handle an HTTP PUT request.

This method is a default implementation of the PUT request handler for adapters. It calls the set method of the controller with the specified data and returns the result of a subsequent get call. Error handling is provided to catch controller errors and return appropriate error responses.

Parameters:

Name Type Description Default
path

URI path of resource

 required
request

HTTP request object passed from handler

 required

Returns:

Type Description

ApiAdapterResponse container of data, content-type and status_code
Source code in src/odin_control/adapters/adapter.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@request_types("application/json", "application/vnd.odin-native")
@response_types("application/json", default="application/json")
@require_controller
def put(self, path, request):
    """Handle an HTTP PUT request.

    This method is a default implementation of the PUT request handler for adapters. It calls
    the set method of the controller with the specified data and returns the result of a
    subsequent get call. Error handling is provided to catch controller errors and return
    appropriate error responses.

    :param path: URI path of resource
    :param request: HTTP request object passed from handler
    :return: ApiAdapterResponse container of data, content-type and status_code
    """
    content_type = "application/json"

    try:
        data = json_decode(request.body)
        self.controller.set(path, data)
        response = self.controller.get(path)
        status_code = 200
    except self.error_cls as error:
        response = {"error": str(error)}
        status_code = 400
    except NotImplementedError:
        response = {
            "error": f"{self.name} does not support PUT requests"
        }
        status_code = 405
    except (TypeError, ValueError) as error:
        response = {"error": f"Failed to decode PUT request body: {str(error)}"}
        status_code = 400

    return ApiAdapterResponse(response, content_type=content_type, status_code=status_code)