ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-22 09:10:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

reduces complexity of create_api_application method

Browse Source
This commit is contained in:
Mose Müller 2024-12-02 14:57:23 +01:00
parent 6d12e5c939
commit 374a930745
1 changed files with 65 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
src/pydase/server/web_server/api/v1/application.py
View File

@ -24,57 +24,75 @@ STATUS_OK = 200
STATUS_FAILED = 400
async def _get_value(
state_manager: StateManager, request: aiohttp.web.Request
) -> aiohttp.web.Response:
logger.info("Handle api request: %s", request)
access_path = request.rel_url.query["access_path"]
status = STATUS_OK
try:
result = get_value(state_manager, access_path)
except Exception as e:
logger.exception(e)
result = dump(e)
status = STATUS_FAILED
return aiohttp.web.json_response(result, status=status)
async def _update_value(
state_manager: StateManager, request: aiohttp.web.Request
) -> aiohttp.web.Response:
data: UpdateDict = await request.json()
try:
update_value(state_manager, data)
return aiohttp.web.json_response()
except Exception as e:
logger.exception(e)
return aiohttp.web.json_response(dump(e), status=STATUS_FAILED)
async def _trigger_method(
state_manager: StateManager, request: aiohttp.web.Request
) -> aiohttp.web.Response:
data: TriggerMethodDict = await request.json()
method = get_object_attr_from_path(state_manager.service, data["access_path"])
try:
if inspect.iscoroutinefunction(method):
method_return = await trigger_async_method(
state_manager=state_manager, data=data
)
else:
method_return = trigger_method(state_manager=state_manager, data=data)
return aiohttp.web.json_response(method_return)
except Exception as e:
logger.exception(e)
return aiohttp.web.json_response(dump(e), status=STATUS_FAILED)
def create_api_application(state_manager: StateManager) -> aiohttp.web.Application:
api_application = aiohttp.web.Application(
middlewares=(aiohttp_middlewares.error.error_middleware(),)
)
async def _get_value(request: aiohttp.web.Request) -> aiohttp.web.Response:
logger.info("Handle api request: %s", request)
access_path = request.rel_url.query["access_path"]
status = STATUS_OK
try:
result = get_value(state_manager, access_path)
except Exception as e:
logger.exception(e)
result = dump(e)
status = STATUS_FAILED
return aiohttp.web.json_response(result, status=status)
async def _update_value(request: aiohttp.web.Request) -> aiohttp.web.Response:
data: UpdateDict = await request.json()
try:
update_value(state_manager, data)
return aiohttp.web.json_response()
except Exception as e:
logger.exception(e)
return aiohttp.web.json_response(dump(e), status=STATUS_FAILED)
async def _trigger_method(request: aiohttp.web.Request) -> aiohttp.web.Response:
data: TriggerMethodDict = await request.json()
method = get_object_attr_from_path(state_manager.service, data["access_path"])
try:
if inspect.iscoroutinefunction(method):
method_return = await trigger_async_method(
state_manager=state_manager, data=data
)
else:
method_return = trigger_method(state_manager=state_manager, data=data)
return aiohttp.web.json_response(method_return)
except Exception as e:
logger.exception(e)
return aiohttp.web.json_response(dump(e), status=STATUS_FAILED)
api_application.router.add_get("/get_value", _get_value)
api_application.router.add_put("/update_value", _update_value)
api_application.router.add_put("/trigger_method", _trigger_method)
api_application.router.add_get(
"/get_value",
lambda request: _get_value(state_manager=state_manager, request=request),
)
api_application.router.add_put(
"/update_value",
lambda request: _update_value(state_manager=state_manager, request=request),
)
api_application.router.add_put(
"/trigger_method",
lambda request: _trigger_method(state_manager=state_manager, request=request),
)
return api_application