Use web.AppKey instead of string keys (with backward compatibility)

Author: ruscoder

aidbox_python_sdk/app_keys.py

@@ -0,0 +1,11 @@
+from aiohttp import web
+
+from aidbox_python_sdk.aidboxpy import AsyncAidboxClient
+from aidbox_python_sdk.db import DBProxy
+from aidbox_python_sdk.sdk import SDK
+from aidbox_python_sdk.settings import Settings
+
+db: web.AppKey[DBProxy] = web.AppKey("db", DBProxy)
+client: web.AppKey[AsyncAidboxClient] = web.AppKey("client", AsyncAidboxClient)
+sdk: web.AppKey[SDK] = web.AppKey("sdk", SDK)
+settings: web.AppKey[Settings] = web.AppKey("settings", Settings)

aidbox_python_sdk/handlers.py

@@ -1,19 +1,22 @@
 import asyncio
 import logging
+from typing import Any
 
 from aiohttp import web
 from fhirpy.base.exceptions import OperationOutcome
 
+from . import app_keys as ak
+
 logger = logging.getLogger("aidbox_sdk")
 routes = web.RouteTableDef()
 
 
-async def subscription(request, data):
+async def subscription(request: web.Request, data: dict):
     logger.debug("Subscription handler: %s", data["handler"])
     if "handler" not in data or "event" not in data:
         logger.error("`handler` and/or `event` param is missing, data: %s", data)
         raise web.HTTPBadRequest()
-    handler = request.app["sdk"].get_subscription_handler(data["handler"])
+    handler = request.app[ak.sdk].get_subscription_handler(data["handler"])
     if not handler:
         logger.error("Subscription handler `%s` was not found", "handler")
         raise web.HTTPNotFound()
@@ -23,12 +26,12 @@ async def subscription(request, data):
     return web.json_response({})
 
 
-async def operation(request, data):
+async def operation(request: web.Request, data: dict[str, Any]):
     logger.debug("Operation handler: %s", data["operation"]["id"])
     if "operation" not in data or "id" not in data["operation"]:
         logger.error("`operation` or `operation[id]` param is missing, data: %s", data)
         raise web.HTTPBadRequest()
-    handler = request.app["sdk"].get_operation_handler(data["operation"]["id"])
+    handler = request.app[ak.sdk].get_operation_handler(data["operation"]["id"])
     if not handler:
         logger.error("Operation handler `%s` was not found", data["handler"])
         raise web.HTTPNotFound()

aidbox_python_sdk/main.py

@@ -3,10 +3,12 @@
 import logging
 import sys
 from pathlib import Path
+from typing import cast
 
 from aiohttp import BasicAuth, client_exceptions, web
 from fhirpy.base.exceptions import OperationOutcome
 
+from . import app_keys as ak
 from .aidboxpy import AsyncAidboxClient
 from .db import DBProxy
 from .handlers import routes
@@ -47,28 +49,31 @@ async def register_app(sdk: SDK, client: AsyncAidboxClient):
 async def init_client(settings: Settings):
     aidbox_client_cls = settings.AIDBOX_CLIENT_CLASS
     basic_auth = BasicAuth(
-        login=settings.APP_INIT_CLIENT_ID,
-        password=settings.APP_INIT_CLIENT_SECRET,
+        login=cast(str, settings.APP_INIT_CLIENT_ID),
+        password=cast(str, settings.APP_INIT_CLIENT_SECRET),
     )
 
     return aidbox_client_cls(f"{settings.APP_INIT_URL}", authorization=basic_auth.encode())
 
 
-async def init(app):
-    app["client"] = await init_client(app["settings"])
-    app["db"] = DBProxy(app["settings"])
-    await register_app(app["sdk"], app["client"])
-    await app["db"].initialize()
+async def init(app: web.Application):
+    app[ak.client] = await init_client(app[ak.settings])
+    app["client"] = app[ak.client]  # For backwards compatibility
+    app[ak.db] = DBProxy(app[ak.settings])
+    app["db"] = app[ak.db]  # For backwards compatibility
+    await register_app(app[ak.sdk], app[ak.client])
+    await app[ak.db].initialize()
     yield
-    await app["db"].deinitialize()
+    await app[ak.db].deinitialize()
 
 
 def create_app(sdk: SDK):
     app = web.Application()
     app.cleanup_ctx.append(init)
-    app.update(
-        settings=sdk.settings,
-        sdk=sdk,
-    )
+    app[ak.sdk] = sdk
+    app["sdk"] = app[ak.sdk]  # For backwards compatibility
+    app[ak.settings] = sdk.settings
+    app["settings"] = app[ak.settings]  # For backwards compatibility
+
     setup_routes(app)
     return app

aidbox_python_sdk/pytest_plugin.py

@@ -1,16 +1,19 @@
 import os
+from typing import cast
 
 import pytest
 import pytest_asyncio
-from aiohttp import BasicAuth, ClientSession
+from aiohttp import BasicAuth, ClientSession, web
 from yarl import URL
 
 from main import create_app as _create_app
 
+from . import app_keys as ak
+
 
 async def start_app(aiohttp_client):
     app = await aiohttp_client(_create_app(), server_kwargs={"host": "0.0.0.0", "port": 8081})
-    sdk = app.server.app["sdk"]
+    sdk = cast(web.Application, app.server.app)[ak.sdk]
     sdk._test_start_txid = -1
 
     return app
@@ -40,20 +43,18 @@ async def _request(self, method, path, *args, **kwargs):
 @pytest_asyncio.fixture
 async def aidbox(client):
     """HTTP client for making requests to Aidbox"""
-    app = client.server.app
+    app = cast(web.Application, client.server.app)
     basic_auth = BasicAuth(
-        login=app["settings"].APP_INIT_CLIENT_ID,
-        password=app["settings"].APP_INIT_CLIENT_SECRET,
+        login=app[ak.settings].APP_INIT_CLIENT_ID,
+        password=app[ak.settings].APP_INIT_CLIENT_SECRET,
     )
-    session = AidboxSession(auth=basic_auth)
+    session = AidboxSession(auth=basic_auth, base_url=app[ak.settings].APP_INIT_URL)
     yield session
     await session.close()
 
 
 @pytest_asyncio.fixture
-async def safe_db(aidbox, client):
-    sdk = client.server.app["sdk"]
-
+async def safe_db(aidbox, client, sdk):
     resp = await aidbox.post(
         "/$psql",
         json={"query": "SELECT last_value from transaction_id_seq;"},
@@ -76,14 +77,14 @@ async def safe_db(aidbox, client):
 
 @pytest.fixture()
 def sdk(client):
-    return client.server.app["sdk"]
+    return cast(web.Application, client.server.app)[ak.sdk]
 
 
 @pytest.fixture()
 def aidbox_client(client):
-    return client.server.app["client"]
+    return cast(web.Application, client.server.app)[ak.client]
 
 
 @pytest.fixture()
 def aidbox_db(client):
-    return client.server.app["db"]
+    return cast(web.Application, client.server.app)[ak.db]

aidbox_python_sdk/types.py

@@ -1,16 +1,13 @@
-from typing_extensions import TypedDict
-
-from aidbox_python_sdk.aidboxpy import AsyncAidboxClient
-from aidbox_python_sdk.db import DBProxy
-from aidbox_python_sdk.sdk import SDK
-
-
-class SDKOperationRequestApp(TypedDict):
-    client: AsyncAidboxClient
-    db: DBProxy
-    sdk: SDK
+from typing import Any
 
+from aiohttp import web
+from typing_extensions import TypedDict
 
 SDKOperationRequest = TypedDict(
-    "SDKOperationRequest", {"app": SDKOperationRequestApp, "route-params": dict, "resource": dict}
+    "SDKOperationRequest",
+    {"app": web.Application, "params": dict, "route-params": dict, "resource": Any},
 )
+
+
+class SDKOperation(TypedDict):
+    pass