Skip to content

Routes

API route implementation for odin-control.

This module implements an API route object used by odin-control to handle API-related requests.

Tim Nicholls, STFC Application Engineering Group

ApiRoute

Bases: Route

ApiRoute - API route object used to map handlers onto adapter for API calls.

Source code in src/odin_control/http/routes/api.py 
 18
 19
 20
 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
class ApiRoute(Route):
    """ApiRoute - API route object used to map handlers onto adapter for API calls."""

    def __init__(self, enable_cors=False, cors_origin="*", api_version=None):
        """Initialize the ApiRoute object.

        This constructor initialises the API route object, defining handlers for version, adapter
        list and adapter API calls.

        :param enable_cors: flag to enable CORS request support
        :param cors_origin: CORS allowed origins
        :param api_version: API version string
        """
        super(ApiRoute, self).__init__()

        # Store the API version string so it can be used by handlers
        self.api_version = api_version

        # Build a dict of params to be passed to API handler initialisation calls
        handler_params = {
            "route": self, "enable_cors": enable_cors, "cors_origin": cors_origin
        }

        # Define a default handler which can return the supported API version
        self.add_handler((r"/api/?", ApiVersionHandler, handler_params))

        # Define the API handler URL specs depending on whether versioning is enabled
        if self.api_version:
            adapter_info_spec = r"/api/(.*?)/adapters/?"
            api_specs = [r"/api/(.*?)/(.*?)/(.*)", r"/api/(.*?)/(.*?)/?"]
        else:
            adapter_info_spec = r"/api/adapters/?"
            api_specs = [r"/api/(.*?)/(.*)", r"/api/(.*?)/?"]

        # Define a handler which can return a list of loaded adapters
        self.add_handler((adapter_info_spec, ApiAdapterInfoHandler, handler_params))

        # Define the handler for API calls. The expected URI syntax, which is enforced by the
        # validate_api_request decorator, is the following:
        #
        #    /api/<version>/<subsystem>/<action>....
        #
        # where the <version> part is optional depending on whether API versioning is enabled or
        # not. The second pattern allows an API adapter to be accessed with or without a trailing
        # slash for maximum compatibility
        for api_spec in api_specs:
            self.add_handler((api_spec, ApiHandler, handler_params))

        self.adapters = {}

    def register_adapter(self, adapter_config, fail_ok=True):
        """Register an API adapter with the APIRoute object.

        Based on the adapter_config object passed in as an argument, this method attempts to
        load the specified module and create an instance of the adapter class to be used
        to handle requests to the API route on the appropriate path.

        :param adapter_config: AdapterConfig object for the adapter
        :param fail_ok: Allow the adapter import and registration to fail without raising an error
        """
        # Resolve the adapter module and class name from the dotted module path in the config object
        (module_name, class_name) = adapter_config.module.rsplit(".", 1)

        # Try to import the module, resolve the class in the module and create an instance of it
        try:
            adapter_module = importlib.import_module(module_name)
            adapter_class = getattr(adapter_module, class_name)
            if adapter_class.is_async:
                adapter = run_async(adapter_class, **adapter_config.options())
            else:
                adapter = adapter_class(**adapter_config.options())
            self.adapters[adapter_config.name] = adapter

        except (ImportError, AttributeError) as e:
            logging.error(
                "Failed to register API adapter %s for path %s with dispatcher: %s",
                adapter_config.module,
                adapter_config.name,
                e,
            )
            if not fail_ok:
                raise ApiError(e)
        else:
            logging.debug(
                "Registered API adapter class %s from module %s for path %s",
                class_name,
                module_name,
                adapter_config.name,
            )

    def has_adapter(self, subsystem):
        """Determine if ApiRoute object has adapter for subsystem.

        :param subsystem: subsystem to check for adapter
        :return: True of adapter present for subsystem
        """
        return subsystem in self.adapters

    def adapter(self, subsystem):
        """Return adapter for subsystem.

        :param subsystem: subsystem to return adapter for
        :return: adapter for subsystem
        """
        return self.adapters[subsystem]

    def cleanup_adapters(self):
        """Clean up state of registered adapters.

        This calls the cleanup method present in any registered adapters, allowing them to
        clean up their state (e.g. connected hardware) in a controlled fashion at shutdown.
        """
        for adapter_name, adapter in self.adapters.items():
            try:
                cleanup_method = getattr(adapter, "cleanup")
                if adapter.is_async:
                    run_async(cleanup_method)
                else:
                    cleanup_method()
            except AttributeError:
                logging.debug("Adapter %s has no cleanup method", adapter_name)

    def initialize_adapters(self):
        """Initialize all the adapters after they have been registered by the route.

        This calls the initialize method present in any registered adapters, passing
        the dictionary of listed adapters to each, for inter adapter communication.
        """
        for adapter_name, adapter in self.adapters.items():
            try:
                initialize_method = getattr(adapter, "initialize")
                if adapter.is_async:
                    run_async(initialize_method, self.adapters)
                else:
                    initialize_method(self.adapters)
            except AttributeError:
                logging.debug("Adapter %s has no initialize method", adapter_name)

__init__(enable_cors=False, cors_origin='*', api_version=None)

Initialize the ApiRoute object.

This constructor initialises the API route object, defining handlers for version, adapter list and adapter API calls.

Parameters:

Name Type Description Default
enable_cors

flag to enable CORS request support

 False
cors_origin

CORS allowed origins

 '*'
api_version

API version string

 None
Source code in src/odin_control/http/routes/api.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
def __init__(self, enable_cors=False, cors_origin="*", api_version=None):
    """Initialize the ApiRoute object.

    This constructor initialises the API route object, defining handlers for version, adapter
    list and adapter API calls.

    :param enable_cors: flag to enable CORS request support
    :param cors_origin: CORS allowed origins
    :param api_version: API version string
    """
    super(ApiRoute, self).__init__()

    # Store the API version string so it can be used by handlers
    self.api_version = api_version

    # Build a dict of params to be passed to API handler initialisation calls
    handler_params = {
        "route": self, "enable_cors": enable_cors, "cors_origin": cors_origin
    }

    # Define a default handler which can return the supported API version
    self.add_handler((r"/api/?", ApiVersionHandler, handler_params))

    # Define the API handler URL specs depending on whether versioning is enabled
    if self.api_version:
        adapter_info_spec = r"/api/(.*?)/adapters/?"
        api_specs = [r"/api/(.*?)/(.*?)/(.*)", r"/api/(.*?)/(.*?)/?"]
    else:
        adapter_info_spec = r"/api/adapters/?"
        api_specs = [r"/api/(.*?)/(.*)", r"/api/(.*?)/?"]

    # Define a handler which can return a list of loaded adapters
    self.add_handler((adapter_info_spec, ApiAdapterInfoHandler, handler_params))

    # Define the handler for API calls. The expected URI syntax, which is enforced by the
    # validate_api_request decorator, is the following:
    #
    #    /api/<version>/<subsystem>/<action>....
    #
    # where the <version> part is optional depending on whether API versioning is enabled or
    # not. The second pattern allows an API adapter to be accessed with or without a trailing
    # slash for maximum compatibility
    for api_spec in api_specs:
        self.add_handler((api_spec, ApiHandler, handler_params))

    self.adapters = {}

adapter(subsystem)

Return adapter for subsystem.

Parameters:

Name Type Description Default
subsystem

subsystem to return adapter for

 required

Returns:

Type Description

adapter for subsystem
Source code in src/odin_control/http/routes/api.py 
116
117
118
119
120
121
122
def adapter(self, subsystem):
    """Return adapter for subsystem.

    :param subsystem: subsystem to return adapter for
    :return: adapter for subsystem
    """
    return self.adapters[subsystem]

cleanup_adapters()

Clean up state of registered adapters.

This calls the cleanup method present in any registered adapters, allowing them to clean up their state (e.g. connected hardware) in a controlled fashion at shutdown.

Source code in src/odin_control/http/routes/api.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def cleanup_adapters(self):
    """Clean up state of registered adapters.

    This calls the cleanup method present in any registered adapters, allowing them to
    clean up their state (e.g. connected hardware) in a controlled fashion at shutdown.
    """
    for adapter_name, adapter in self.adapters.items():
        try:
            cleanup_method = getattr(adapter, "cleanup")
            if adapter.is_async:
                run_async(cleanup_method)
            else:
                cleanup_method()
        except AttributeError:
            logging.debug("Adapter %s has no cleanup method", adapter_name)

has_adapter(subsystem)

Determine if ApiRoute object has adapter for subsystem.

Parameters:

Name Type Description Default
subsystem

subsystem to check for adapter

 required

Returns:

Type Description

True of adapter present for subsystem
Source code in src/odin_control/http/routes/api.py 
108
109
110
111
112
113
114
def has_adapter(self, subsystem):
    """Determine if ApiRoute object has adapter for subsystem.

    :param subsystem: subsystem to check for adapter
    :return: True of adapter present for subsystem
    """
    return subsystem in self.adapters

initialize_adapters()

Initialize all the adapters after they have been registered by the route.

This calls the initialize method present in any registered adapters, passing the dictionary of listed adapters to each, for inter adapter communication.

Source code in src/odin_control/http/routes/api.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def initialize_adapters(self):
    """Initialize all the adapters after they have been registered by the route.

    This calls the initialize method present in any registered adapters, passing
    the dictionary of listed adapters to each, for inter adapter communication.
    """
    for adapter_name, adapter in self.adapters.items():
        try:
            initialize_method = getattr(adapter, "initialize")
            if adapter.is_async:
                run_async(initialize_method, self.adapters)
            else:
                initialize_method(self.adapters)
        except AttributeError:
            logging.debug("Adapter %s has no initialize method", adapter_name)

register_adapter(adapter_config, fail_ok=True)

Register an API adapter with the APIRoute object.

Based on the adapter_config object passed in as an argument, this method attempts to load the specified module and create an instance of the adapter class to be used to handle requests to the API route on the appropriate path.

Parameters:

Name Type Description Default
adapter_config

AdapterConfig object for the adapter

 required
fail_ok

Allow the adapter import and registration to fail without raising an error

 True
Source code in src/odin_control/http/routes/api.py 
 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
def register_adapter(self, adapter_config, fail_ok=True):
    """Register an API adapter with the APIRoute object.

    Based on the adapter_config object passed in as an argument, this method attempts to
    load the specified module and create an instance of the adapter class to be used
    to handle requests to the API route on the appropriate path.

    :param adapter_config: AdapterConfig object for the adapter
    :param fail_ok: Allow the adapter import and registration to fail without raising an error
    """
    # Resolve the adapter module and class name from the dotted module path in the config object
    (module_name, class_name) = adapter_config.module.rsplit(".", 1)

    # Try to import the module, resolve the class in the module and create an instance of it
    try:
        adapter_module = importlib.import_module(module_name)
        adapter_class = getattr(adapter_module, class_name)
        if adapter_class.is_async:
            adapter = run_async(adapter_class, **adapter_config.options())
        else:
            adapter = adapter_class(**adapter_config.options())
        self.adapters[adapter_config.name] = adapter

    except (ImportError, AttributeError) as e:
        logging.error(
            "Failed to register API adapter %s for path %s with dispatcher: %s",
            adapter_config.module,
            adapter_config.name,
            e,
        )
        if not fail_ok:
            raise ApiError(e)
    else:
        logging.debug(
            "Registered API adapter class %s from module %s for path %s",
            class_name,
            module_name,
            adapter_config.name,
        )