try publish demo app as artefact

otdemo/otdemo_adapter.py

@@ -0,0 +1,59 @@
+"""
+This module provides the CleverSwarm AMQP Adapter for handling API requests via AMQP.
+The code IS MEANT to be part of CleverSwarm codebase, and has direct dependencies on the CleverSwarm codebase.
+It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the CleverSwarm endpoints context.
+"""
+
+from threading import Thread
+from typing import Any, List
+
+# CleverSwarm specific imports
+from fastapi.routing import APIRoute
+
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
+from amqp.config.amq_configuration import AMQConfiguration
+from amqp.service.amq_service import AMQService
+
+PREFERRED_SUFFIX = "_json"
+
+
+# =================================================================================================
+# ======================== C l e v e r S w a r m   A M Q    A d a p t e r  ========================
+# =================================================================================================
+class OTDemoAmqpAdapter(CleverThisServiceAdapter):
+    """
+    CleverSwarm AMQP Adapter for CleverSwarm API. Implements the CleverSwarm specific logic that overrides
+    common logic defined by AMQ Adapter library. The specific logic here is:
+
+    1. Extract subject from JWT token and convert it to UserSchema to avail authorized user argument
+    2. Find JSON wrapper for endpoint Callable, and if exists, use it instead of the original endpoint.
+
+    """
+
+    def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]):
+        super().__init__(amq_configuration, routes=routes, session=None)
+        _amq_service: AMQService = AMQService(amq_configuration, self)
+        self.thread = Thread(target=_amq_service.run)
+        self.thread.start()
+
+    async def get_current_user(self, token: str) -> str:
+        """
+        Overrides the default get_current_user method to use the token from the AMQP message.
+        In CleverSwarm context, it means simply to call provided get_current_user function.
+
+        :param token: JWT token from the AMQP message
+
+        :return: UserSchema object
+        """
+        return token
+
+    def get_endpoint(self, endpoint: Any) -> Any:
+        """
+        If the specific endpoint has a matching JSON wrapper (identified as {endpoint.name}_json ),
+        then use it as preferred way to invoke the endpoint.
+
+        :param endpoint: the Callable for the endpoint
+
+        :return: new Callable representing the JSON wrapper endpoint, if it exists or the original Callable.
+        """
+        return endpoint