hat.event.manager.rest

  1import collections
  2import functools
  3
  4import aiohttp.web
  5
  6from hat import aio
  7from hat import json
  8from hat import juggler
  9from hat.drivers import net
 10
 11from hat.event import eventer
 12from hat.event.manager import common
 13
 14
 15async def run(addr: net.StreamAddress,
 16              client_name: str,
 17              client_token: str | None,
 18              http_addr: net.TcpAddress):
 19    async_group = aio.Group()
 20
 21    try:
 22        client = await eventer.connect(addr=addr,
 23                                       client_name=client_name,
 24                                       client_token=client_token)
 25        await _bind_resource(async_group, client)
 26
 27        server = await _create_web_server(http_addr=http_addr,
 28                                          client=client)
 29        await _bind_resource(async_group, server)
 30
 31        await async_group.wait_closing()
 32
 33    finally:
 34        await aio.uncancellable(async_group.async_close())
 35
 36
 37async def _create_web_server(http_addr, client):
 38    query_routes = [
 39        aiohttp.web.get(
 40            f'/query/{path}',
 41            functools.partial(_process_get_query, parse_params, client))
 42        for path, parse_params in [('latest', _parse_latest_params),
 43                                   ('timeseries', _parse_timeseries_params),
 44                                   ('server', _parse_server_params)]]
 45
 46    return await juggler.listen(
 47        host=http_addr.host,
 48        port=http_addr.port,
 49        ws_path=None,
 50        index_path=None,
 51        additional_routes=query_routes)
 52
 53
 54async def _process_get_query(parse_params, client, req):
 55    try:
 56        body = await req.json()
 57        params = parse_params(body)
 58
 59    except Exception as e:
 60        raise aiohttp.web.HTTPBadRequest(text=f"error reading params: {e}")
 61
 62    try:
 63        result = await client.query(params)
 64
 65        return aiohttp.web.Response(
 66            content_type='application/json',
 67            text=json.encode(common.query_result_to_json(result)))
 68
 69    except Exception as e:
 70        raise aiohttp.web.HTTPInternalServerError(text=f"query error: {e}")
 71
 72
 73def _parse_latest_params(params):
 74    event_types = (_parse_event_types(params['event_types'])
 75                   if 'event_types' in params else None)
 76
 77    return common.QueryLatestParams(event_types=event_types)
 78
 79
 80def _parse_timeseries_params(params):
 81    event_types = (_parse_event_types(params['event_types'])
 82                   if 'event_types' in params else None)
 83    t_from = _parse_timestamp(params['t_from']) if 't_from' in params else None
 84    t_to = _parse_timestamp(params['t_to']) if 't_to' in params else None
 85    source_t_from = (_parse_timestamp(params['source_t_from'])
 86                     if 'source_t_from' in params else None)
 87    source_t_to = (_parse_timestamp(params['source_t_to'])
 88                   if 'source_t_to' in params else None)
 89    order = common.Order(params.get('order', 'descending'))
 90    order_by = common.OrderBy(params.get('order_by', 'timestamp'))
 91    max_results = (_parse_max_results(params['max_results'])
 92                   if 'max_results' in params else None)
 93    last_event_id = (_parse_event_id(params['last_event_id'])
 94                     if 'last_event_id' in params else None)
 95
 96    return common.QueryTimeseriesParams(event_types=event_types,
 97                                        t_from=t_from,
 98                                        t_to=t_to,
 99                                        source_t_from=source_t_from,
100                                        source_t_to=source_t_to,
101                                        order=order,
102                                        order_by=order_by,
103                                        max_results=max_results,
104                                        last_event_id=last_event_id)
105
106
107def _parse_server_params(params):
108    server_id = _parse_server_id(params['server_id'])
109    persisted = (_parse_persisted(params['persisted'])
110                 if 'persisted' in params else False)
111    max_results = (_parse_max_results(params['max_results'])
112                   if 'max_results' in params else None)
113    last_event_id = (_parse_event_id(params['last_event_id'])
114                     if 'last_event_id' in params else None)
115
116    return common.QueryServerParams(server_id=server_id,
117                                    persisted=persisted,
118                                    max_results=max_results,
119                                    last_event_id=last_event_id)
120
121
122def _parse_event_types(event_types):
123    result = collections.deque()
124
125    for event_type in event_types:
126        event_type = tuple(event_type)
127
128        for segment in event_type:
129            if not isinstance(segment, str):
130                raise Exception('invalid event type')
131
132        result.append(event_type)
133
134    return result
135
136
137def _parse_server_id(server_id):
138    if not isinstance(server_id, int):
139        raise Exception('invalid server id')
140
141    return int(server_id)
142
143
144def _parse_max_results(max_results):
145    if not isinstance(max_results, int):
146        raise Exception('invalid max results')
147
148    return int(max_results)
149
150
151def _parse_persisted(persisted):
152    if not isinstance(persisted, bool):
153        raise Exception('invalid persisted')
154
155    return persisted
156
157
158def _parse_timestamp(t):
159    if not isinstance(t['s'], int):
160        raise Exception('invalid timestamp seconds')
161
162    if not isinstance(t['us'], int) or not (0 <= t['us'] < 1_000_000):
163        raise Exception('invalid timestamp micro seconds')
164
165    return common.Timestamp(s=int(t['s']),
166                            us=int(t['us']))
167
168
169def _parse_event_id(event_id):
170    if (not isinstance(event_id['server'], int) or
171            not isinstance(event_id['session'], int) or
172            not isinstance(event_id['instance'], int)):
173        raise Exception('invalid event id')
174
175    return common.EventId(server=int(event_id['server']),
176                          session=int(event_id['session']),
177                          instance=int(event_id['instance']))
178
179
180async def _bind_resource(async_group, resource):
181    try:
182        async_group.spawn(aio.call_on_cancel, resource.async_close)
183        async_group.spawn(aio.call_on_done, resource.wait_closing(),
184                          async_group.close)
185
186    except Exception:
187        await aio.uncancellable(resource.async_close())
188        raise