Coverage for core / src / sensorkit / webapi / __init__.py: 54%

1# SPDX-License-Identifier: Apache-2.0 

2import sensorkit.api as sk 

3from sensorkit.backend.base import KeyNotFound 

4 

5from .fastapi import WebAPI, WebAPIConfig 

6 

7sk.declare_config_section( 

8 "webapi", 

9 WebAPIConfig, 

10 id_source="by_key", 

11 id_default="webapi", 

12 service_path=__name__, 

13) 

14 

15 

16@sk.declare_entity 

17class WebAPIService: 

18 webapi: WebAPI | None = None 

19 

20 @sk.on_attach 

21 async def startup(self): 

22 entity = sk.entity() 

23 

24 try: 

25 config = await entity.kv_get_model(WebAPIConfig) 

26 except KeyNotFound: 

27 config = WebAPIConfig() 

28 

29 self.webapi = WebAPI(entity.sensorkit(), config) 

30 entity.task_group.create_task(self.webapi.serve(task_group=entity.task_group)) 

31 

32 @sk.on_detach 

33 async def shutdown(self): 

34 if self.webapi is not None: 

35 await self.webapi.shutdown() 

36 

37 

38@sk.service_entrypoint(version=sk.VERSION) 

39async def webapi_service(service: sk.Service): 

40 service.include(WebAPIService) 

41 await service.run()