Skip to content

Dummy

Dummy adapter classes for the ODIN server.

The DummyAdapter class implements a dummy adapter for the ODIN server, demonstrating the basic adapter implementation and providing a loadable adapter for testing

The IacDummyAdapter class implements a dummy adapter for the ODIN server that can demonstrate the inter adapter communication, and how an adapter might use the dictionary of loaded adapters to communicate with them.

Tim Nicholls, STFC Application Engineering

DummyAdapter

Bases: ApiAdapter

Dummy adapter class for the ODIN server.

This dummy adapter implements the basic operation of an adapter including initialisation and HTTP verb methods (GET, PUT, DELETE) with various request and response types allowed.

Source code in src/odin_control/adapters/dummy.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
class DummyAdapter(ApiAdapter):
    """Dummy adapter class for the ODIN server.

    This dummy adapter implements the basic operation of an adapter including initialisation
    and HTTP verb methods (GET, PUT, DELETE) with various request and response types allowed.
    """

    version = __version__

    def __init__(self, **kwargs):
        """Initialize the DummyAdapter object.

        This constructor Initializes the DummyAdapter object, including launching a background
        task if enabled by the adapter options passed as arguments.

        :param kwargs: keyword arguments specifying options
        """
        super(DummyAdapter, self).__init__(**kwargs)

        # Set the background task counter to zero
        self.background_task_counter = 0

        # Launch the background task if enabled in options
        if self.options.get('background_task_enable', False):
            task_interval = float(
                self.options.get('background_task_interval', 1.0)
                )
            logging.debug(
                "Launching background task with interval %.2f secs", task_interval
            )
            self.background_task = PeriodicCallback(
                self.background_task_callback, task_interval * 1000
            )
            self.background_task.start()

        logging.debug('DummyAdapter loaded')

    def initialize(self, adapters):
        logging.debug("DummyAdapter initialized with %d adapters", len(adapters))

    def background_task_callback(self):
        """Run the adapter background task.

        This callback simply increments the background task counter.
        """
        logging.debug(
            "%s: background task running, count = %d", self.name, self.background_task_counter)
        self.background_task_counter += 1

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

        This method handles an HTTP GET request, returning a JSON response. To facilitate
        testing of the background task, if the URI path is set appropriately, the task
        counter is returned in the JSON response.

        :param path: URI path of request
        :param request: HTTP request object
        :return: an ApiAdapterResponse object containing the appropriate response
        """
        if path == 'background_task_count':
            response = {'response': {
                'background_task_count': self.background_task_counter}
            }
        else:
            response = {'response': 'DummyAdapter: GET on path {}'.format(path)}

        content_type = 'application/json'
        status_code = 200

        logging.debug(response)

        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')
    def put(self, path, request):
        """Handle an HTTP PUT request.

        This method handles an HTTP PUT request, returning a JSON response.

        :param path: URI path of request
        :param request: HTTP request object
        :return: an ApiAdapterResponse object containing the appropriate response
        """
        response = {'response': 'DummyAdapter: PUT on path {}'.format(path)}
        content_type = 'application/json'
        status_code = 200

        logging.debug(response)

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

    def delete(self, path, request):
        """Handle an HTTP DELETE request.

        This method handles an HTTP DELETE request, returning a JSON response.

        :param path: URI path of request
        :param request: HTTP request object
        :return: an ApiAdapterResponse object containing the appropriate response
        """
        response = 'DummyAdapter: DELETE on path {}'.format(path)
        status_code = 200

        logging.debug(response)

        return ApiAdapterResponse(response, status_code=status_code)

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

        This method cleans up the state of the adapter, which in this case is
        trivially setting the background task counter back to zero for test
        purposes.
        """
        logging.debug("DummyAdapter cleanup: stopping background task")
        self.background_task.stop()
        self.background_task_counter = 0

__init__(**kwargs)

Initialize the DummyAdapter object.

This constructor Initializes the DummyAdapter object, including launching a background task if enabled by the adapter options passed as arguments.

Parameters:

Name Type Description Default
kwargs

keyword arguments specifying options

 {}
Source code in src/odin_control/adapters/dummy.py 
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
def __init__(self, **kwargs):
    """Initialize the DummyAdapter object.

    This constructor Initializes the DummyAdapter object, including launching a background
    task if enabled by the adapter options passed as arguments.

    :param kwargs: keyword arguments specifying options
    """
    super(DummyAdapter, self).__init__(**kwargs)

    # Set the background task counter to zero
    self.background_task_counter = 0

    # Launch the background task if enabled in options
    if self.options.get('background_task_enable', False):
        task_interval = float(
            self.options.get('background_task_interval', 1.0)
            )
        logging.debug(
            "Launching background task with interval %.2f secs", task_interval
        )
        self.background_task = PeriodicCallback(
            self.background_task_callback, task_interval * 1000
        )
        self.background_task.start()

    logging.debug('DummyAdapter loaded')

background_task_callback()

Run the adapter background task.

This callback simply increments the background task counter.

Source code in src/odin_control/adapters/dummy.py 
61
62
63
64
65
66
67
68
def background_task_callback(self):
    """Run the adapter background task.

    This callback simply increments the background task counter.
    """
    logging.debug(
        "%s: background task running, count = %d", self.name, self.background_task_counter)
    self.background_task_counter += 1

cleanup()

Clean up the state of the adapter.

This method cleans up the state of the adapter, which in this case is trivially setting the background task counter back to zero for test purposes.

Source code in src/odin_control/adapters/dummy.py 
133
134
135
136
137
138
139
140
141
142
def cleanup(self):
    """Clean up the state of the adapter.

    This method cleans up the state of the adapter, which in this case is
    trivially setting the background task counter back to zero for test
    purposes.
    """
    logging.debug("DummyAdapter cleanup: stopping background task")
    self.background_task.stop()
    self.background_task_counter = 0

delete(path, request)

Handle an HTTP DELETE request.

This method handles an HTTP DELETE request, returning a JSON response.

Parameters:

Name Type Description Default
path

URI path of request

 required
request

HTTP request object

 required

Returns:

Type Description

an ApiAdapterResponse object containing the appropriate response
Source code in src/odin_control/adapters/dummy.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def delete(self, path, request):
    """Handle an HTTP DELETE request.

    This method handles an HTTP DELETE request, returning a JSON response.

    :param path: URI path of request
    :param request: HTTP request object
    :return: an ApiAdapterResponse object containing the appropriate response
    """
    response = 'DummyAdapter: DELETE on path {}'.format(path)
    status_code = 200

    logging.debug(response)

    return ApiAdapterResponse(response, status_code=status_code)

get(path, request)

Handle an HTTP GET request.

This method handles an HTTP GET request, returning a JSON response. To facilitate testing of the background task, if the URI path is set appropriately, the task counter is returned in the JSON response.

Parameters:

Name Type Description Default
path

URI path of request

 required
request

HTTP request object

 required

Returns:

Type Description

an ApiAdapterResponse object containing the appropriate response
Source code in src/odin_control/adapters/dummy.py 
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
@response_types('application/json', default='application/json')
def get(self, path, request):
    """Handle an HTTP GET request.

    This method handles an HTTP GET request, returning a JSON response. To facilitate
    testing of the background task, if the URI path is set appropriately, the task
    counter is returned in the JSON response.

    :param path: URI path of request
    :param request: HTTP request object
    :return: an ApiAdapterResponse object containing the appropriate response
    """
    if path == 'background_task_count':
        response = {'response': {
            'background_task_count': self.background_task_counter}
        }
    else:
        response = {'response': 'DummyAdapter: GET on path {}'.format(path)}

    content_type = 'application/json'
    status_code = 200

    logging.debug(response)

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

put(path, request)

Handle an HTTP PUT request.

This method handles an HTTP PUT request, returning a JSON response.

Parameters:

Name Type Description Default
path

URI path of request

 required
request

HTTP request object

 required

Returns:

Type Description

an ApiAdapterResponse object containing the appropriate response
Source code in src/odin_control/adapters/dummy.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@request_types('application/json', 'application/vnd.odin-native')
@response_types('application/json', default='application/json')
def put(self, path, request):
    """Handle an HTTP PUT request.

    This method handles an HTTP PUT request, returning a JSON response.

    :param path: URI path of request
    :param request: HTTP request object
    :return: an ApiAdapterResponse object containing the appropriate response
    """
    response = {'response': 'DummyAdapter: PUT on path {}'.format(path)}
    content_type = 'application/json'
    status_code = 200

    logging.debug(response)

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

IacDummyAdapter

Bases: ApiAdapter

Dummy adapter class for the Inter Adapter Communication changes.

This dummy adapter impelements the basic operations of GET and PUT, and allows another adapter to interact with it via these methods.

Source code in src/odin_control/adapters/dummy.py 
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
class IacDummyAdapter(ApiAdapter):
    """Dummy adapter class for the Inter Adapter Communication changes.

    This dummy adapter impelements the basic operations of GET and PUT,
    and allows another adapter to interact with it via these methods.
    """

    version = __version__

    def __init__(self, **kwargs):
        """Initialize the dummy target adapter.

        Create the adapter using the base adapter class.
        Create an empty dictionary to store the references to other loaded adapters.
        """

        super(IacDummyAdapter, self).__init__(**kwargs)
        self.adapters = {}

        logging.debug("IAC Dummy Adapter Loaded")

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

        Call the get method of each other adapter that is loaded and return the responses
        in a dictionary.
        """
        logging.debug("IAC Dummy Get")
        response = {}
        request = ApiAdapterRequest(None, accept="application/json")
        for key, value in self.adapters.items():
            logging.debug("Calling Get of %s", key)
            response[key] = value.get(path=path, request=request).data
        logging.debug("Full response: %s", response)
        content_type = "application/json"
        status_code = 200

        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")
    def put(self, path, request):
        """Handle a HTTP PUT request.

        Calls the put method of each other adapter that has been loaded, and returns the responses
        in a dictionary.
        """
        logging.debug("IAC DUMMY PUT")
        body = decode_request_body(request)
        response = {}
        request = ApiAdapterRequest(body)

        for key, value in self.adapters.items():
            logging.debug("Calling Put of %s", key)
            response[key] = value.put(path="", request=request).data
        content_type = "application/json"
        status_code = 200

        logging.debug(response)

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

    def initialize(self, adapters):
        """Initialize the adapter after it has been loaded.

        Receive a dictionary of all loaded adapters so that they may be accessed by this adapter.
        Remove itself from the dictionary so that it does not reference itself, as doing so
        could end with an endless recursive loop.
        """

        self.adapters = dict((k, v) for k, v in adapters.items() if v is not self)

        logging.debug("Received following dict of Adapters: %s", self.adapters)

__init__(**kwargs)

Initialize the dummy target adapter.

Create the adapter using the base adapter class. Create an empty dictionary to store the references to other loaded adapters.

Source code in src/odin_control/adapters/dummy.py 
154
155
156
157
158
159
160
161
162
163
164
def __init__(self, **kwargs):
    """Initialize the dummy target adapter.

    Create the adapter using the base adapter class.
    Create an empty dictionary to store the references to other loaded adapters.
    """

    super(IacDummyAdapter, self).__init__(**kwargs)
    self.adapters = {}

    logging.debug("IAC Dummy Adapter Loaded")

get(path, request)

Handle a HTTP GET Request

Call the get method of each other adapter that is loaded and return the responses in a dictionary.

Source code in src/odin_control/adapters/dummy.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@response_types("application/json", default="application/json")
def get(self, path, request):
    """Handle a HTTP GET Request

    Call the get method of each other adapter that is loaded and return the responses
    in a dictionary.
    """
    logging.debug("IAC Dummy Get")
    response = {}
    request = ApiAdapterRequest(None, accept="application/json")
    for key, value in self.adapters.items():
        logging.debug("Calling Get of %s", key)
        response[key] = value.get(path=path, request=request).data
    logging.debug("Full response: %s", response)
    content_type = "application/json"
    status_code = 200

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

initialize(adapters)

Initialize the adapter after it has been loaded.

Receive a dictionary of all loaded adapters so that they may be accessed by this adapter. Remove itself from the dictionary so that it does not reference itself, as doing so could end with an endless recursive loop.

Source code in src/odin_control/adapters/dummy.py 
209
210
211
212
213
214
215
216
217
218
219
def initialize(self, adapters):
    """Initialize the adapter after it has been loaded.

    Receive a dictionary of all loaded adapters so that they may be accessed by this adapter.
    Remove itself from the dictionary so that it does not reference itself, as doing so
    could end with an endless recursive loop.
    """

    self.adapters = dict((k, v) for k, v in adapters.items() if v is not self)

    logging.debug("Received following dict of Adapters: %s", self.adapters)

put(path, request)

Handle a HTTP PUT request.

Calls the put method of each other adapter that has been loaded, and returns the responses in a dictionary.

Source code in src/odin_control/adapters/dummy.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@request_types("application/json", "application/vnd.odin-native")
@response_types("application/json", default="application/json")
def put(self, path, request):
    """Handle a HTTP PUT request.

    Calls the put method of each other adapter that has been loaded, and returns the responses
    in a dictionary.
    """
    logging.debug("IAC DUMMY PUT")
    body = decode_request_body(request)
    response = {}
    request = ApiAdapterRequest(body)

    for key, value in self.adapters.items():
        logging.debug("Calling Put of %s", key)
        response[key] = value.put(path="", request=request).data
    content_type = "application/json"
    status_code = 200

    logging.debug(response)

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